* [PATCH 0/2] fsck.erofs: implement multi-threaded extraction
@ 2026-06-21 12:01 Nithurshen
2026-06-21 12:01 ` [PATCH 1/2] fsck.erofs: add multi-threaded decompression Nithurshen
2026-06-21 12:01 ` [PATCH 2/2] fsck.erofs: implement concurrent directory traversal Nithurshen
0 siblings, 2 replies; 13+ messages in thread
From: Nithurshen @ 2026-06-21 12:01 UTC (permalink / raw)
To: linux-erofs; +Cc: hsiangkao, xiang, Nithurshen
Hi Xiang,
This series introduces multi-threaded decompression and extraction to
fsck.erofs.
The architecture is divided into two decoupled workqueues to prevent
thread pool exhaustion gridlock:
1. `erofs_traverse_wq`: Handles the asynchronous directory walk. It
localizes the historically global `extract_path` and `dirstack`
states into individual payloads, safely traversing the tree and
verifying inodes.
2. `erofs_wq`: Dedicated strictly to processing `z_erofs_decompress_req`
payloads. Decompression tasks take strict ownership of the raw and
output buffers, preventing data races.
Testing on heavily packed LZ4HC images demonstrates smooth asynchronous
fan-out, successfully overlapping I/O traversal with decompression
compute.
Nithurshen (2):
fsck.erofs: add multi-threaded decompression
fsck.erofs: implement concurrent directory traversal
fsck/main.c | 450 ++++++++++++++++++++++++---------------
include/erofs/cond.h | 31 +++
include/erofs/internal.h | 20 +-
include/erofs/lock.h | 3 +
lib/data.c | 216 ++++++++++++++-----
5 files changed, 488 insertions(+), 232 deletions(-)
create mode 100644 include/erofs/cond.h
--
2.52.0
^ permalink raw reply [flat|nested] 13+ messages in thread
* [PATCH 1/2] fsck.erofs: add multi-threaded decompression
2026-06-21 12:01 [PATCH 0/2] fsck.erofs: implement multi-threaded extraction Nithurshen
@ 2026-06-21 12:01 ` Nithurshen
2026-06-22 2:08 ` Gao Xiang
` (2 more replies)
2026-06-21 12:01 ` [PATCH 2/2] fsck.erofs: implement concurrent directory traversal Nithurshen
1 sibling, 3 replies; 13+ messages in thread
From: Nithurshen @ 2026-06-21 12:01 UTC (permalink / raw)
To: linux-erofs; +Cc: hsiangkao, xiang, Nithurshen
Currently, fsck.erofs extracts files synchronously. When decompressing
heavily packed images, the main thread spends the majority of its time
blocked on a combination of synchronous vfs_write() syscalls and
decompression routines, bottlenecking overall extraction speed.
This patch introduces a scalable, multi-threaded decompression framework
using the existing erofs_workqueue infrastructure to decouple compute
from the main thread's I/O.
To prevent massive scheduling overhead (futex contention) where worker
threads spend more CPU time waking up than actually decompressing small
clusters, this implementation introduces a batching context. Because
different compression algorithms exhibit vastly different scheduling
thresholds, the batch size is algorithm-aware:
- Fast algorithms like LZ4 utilize a larger batch limit (up to 32
pclusters) to effectively hide synchronization overhead.
- Compute-heavy algorithms like LZMA or ZSTD trigger at a lower
threshold (8 pclusters) to prevent memory bloat and thread starvation.
Key details of this implementation:
- The worker pool is dynamically sized based on available system CPUs.
- Decompression tasks take strict ownership of the raw and output
buffers (safely tracking memory via a `free_out` flag) to prevent
data races and memory leaks.
- Output buffers are explicitly zero-initialized via calloc() to
prevent trailing garbage bytes from leaking into extracted files.
- Tail-end packed fragments are processed synchronously by the main
thread, as their minimal overhead does not benefit from asynchronous
offloading.
Signed-off-by: Nithurshen <nithurshen.dev@gmail.com>
---
fsck/main.c | 150 ++++++++++++---------------
include/erofs/cond.h | 31 ++++++
include/erofs/internal.h | 20 +++-
include/erofs/lock.h | 3 +
lib/data.c | 216 +++++++++++++++++++++++++++++----------
5 files changed, 277 insertions(+), 143 deletions(-)
create mode 100644 include/erofs/cond.h
diff --git a/fsck/main.c b/fsck/main.c
index 16cc627..ffe7e29 100644
--- a/fsck/main.c
+++ b/fsck/main.c
@@ -8,14 +8,18 @@
#include <time.h>
#include <utime.h>
#include <unistd.h>
+#include "erofs/lock.h"
#include <sys/stat.h>
#include "erofs/print.h"
#include "erofs/decompress.h"
#include "erofs/dir.h"
#include "erofs/xattr.h"
+#include "erofs/workqueue.h"
#include "../lib/compressor.h"
#include "../lib/liberofs_compress.h"
+extern struct erofs_workqueue erofs_wq;
+
static int erofsfsck_check_inode(erofs_nid_t pnid, erofs_nid_t nid);
struct erofsfsck_dirstack {
@@ -505,44 +509,36 @@ out:
static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd)
{
- struct erofs_map_blocks map = {
- .buf = __EROFS_BUF_INITIALIZER,
- };
+ struct erofs_map_blocks map = { .buf = __EROFS_BUF_INITIALIZER };
bool needdecode = fsckcfg.check_decomp && !erofs_is_packed_inode(inode);
int ret = 0;
- bool compressed;
+ bool compressed = erofs_inode_is_data_compressed(inode->datalayout);
erofs_off_t pos = 0;
u64 pchunk_len = 0;
- unsigned int raw_size = 0, buffer_size = 0;
- char *raw = NULL, *buffer = NULL;
- erofs_dbg("verify data chunk of nid(%llu): type(%d)",
- inode->nid | 0ULL, inode->datalayout);
+ struct z_erofs_read_ctx ctx = {
+ .pending_tasks = 0,
+ .final_err = 0,
+ .outfd = outfd,
+ .free_out = true,
+ .current_task = NULL
+ };
+ erofs_mutex_init(&ctx.lock);
+ erofs_cond_init(&ctx.cond);
- compressed = erofs_inode_is_data_compressed(inode->datalayout);
- while (pos < inode->i_size) {
- unsigned int alloc_rawsize;
+ erofs_dbg("verify data chunk of nid(%llu): type(%d)", inode->nid | 0ULL, inode->datalayout);
+ while (pos < inode->i_size) {
map.m_la = pos;
ret = erofs_map_blocks(inode, &map, EROFS_GET_BLOCKS_FIEMAP);
- if (ret)
- goto out;
-
- if (!compressed && map.m_llen != map.m_plen) {
- erofs_err("broken chunk length m_la %" PRIu64 " m_llen %" PRIu64 " m_plen %" PRIu64,
- map.m_la, map.m_llen, map.m_plen);
- ret = -EFSCORRUPTED;
- goto out;
- }
+ if (ret) goto out;
- /* the last lcluster can be divided into 3 parts */
if (map.m_la + map.m_llen > inode->i_size)
map.m_llen = inode->i_size - map.m_la;
pchunk_len += map.m_plen;
pos += map.m_llen;
- /* should skip decomp? */
if (map.m_la >= inode->i_size || !needdecode)
continue;
@@ -555,85 +551,53 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd)
continue;
}
- if (map.m_plen > Z_EROFS_PCLUSTER_MAX_SIZE) {
- if (compressed && !(map.m_flags & __EROFS_MAP_FRAGMENT)) {
- erofs_err("invalid pcluster size %" PRIu64 " @ offset %" PRIu64 " of nid %" PRIu64,
- map.m_plen, map.m_la,
- inode->nid | 0ULL);
- ret = -EFSCORRUPTED;
- goto out;
- }
- alloc_rawsize = Z_EROFS_PCLUSTER_MAX_SIZE;
- } else {
- alloc_rawsize = map.m_plen;
- }
-
- if (alloc_rawsize > raw_size) {
- char *newraw = realloc(raw, alloc_rawsize);
-
- if (!newraw) {
+ if (compressed) {
+ char *raw = malloc(map.m_plen);
+ size_t buffer_size = map.m_llen > erofs_blksiz(inode->sbi) ? map.m_llen : erofs_blksiz(inode->sbi);
+ char *buffer = calloc(1, buffer_size);
+
+ if (!raw || !buffer) {
+ free(raw); free(buffer);
ret = -ENOMEM;
goto out;
}
- raw = newraw;
- raw_size = alloc_rawsize;
- }
- if (compressed) {
- if (map.m_llen > buffer_size) {
- char *newbuffer;
-
- buffer_size = map.m_llen;
- newbuffer = realloc(buffer, buffer_size);
- if (!newbuffer) {
- ret = -ENOMEM;
- goto out;
- }
- buffer = newbuffer;
- }
- ret = z_erofs_read_one_data(inode, &map, raw, buffer,
- 0, map.m_llen, false);
- if (ret)
+ ret = z_erofs_read_one_data(inode, &map, raw, buffer, 0, map.m_llen, false, map.m_la, &ctx);
+ if (ret) {
+ /* DO NOT free(raw) or free(buffer) here. z_erofs_read_one_data took ownership! */
goto out;
-
- if (outfd >= 0 && write(outfd, buffer, map.m_llen) < 0)
- goto fail_eio;
+ }
} else {
- u64 p = 0;
-
- do {
- u64 count = min_t(u64, alloc_rawsize,
- map.m_llen);
-
- ret = erofs_read_one_data(inode, &map, raw, p, count);
- if (ret)
- goto out;
-
- if (outfd >= 0 && write(outfd, raw, count) < 0)
- goto fail_eio;
- map.m_llen -= count;
- p += count;
- } while (map.m_llen);
+ char *raw = calloc(1, map.m_llen);
+ ret = erofs_read_one_data(inode, &map, raw, 0, map.m_llen);
+ if (ret >= 0 && outfd >= 0)
+ pwrite(outfd, raw, map.m_llen, map.m_la);
+ free(raw);
+ if (ret) goto out;
}
}
+ z_erofs_read_ctx_enqueue(&ctx);
+
+out:
+ erofs_mutex_lock(&ctx.lock);
+ while (ctx.pending_tasks > 0)
+ erofs_cond_wait(&ctx.cond, &ctx.lock);
+ if (ctx.final_err < 0 && ret >= 0)
+ ret = ctx.final_err;
+ erofs_mutex_unlock(&ctx.lock);
if (fsckcfg.print_comp_ratio) {
if (!erofs_is_packed_inode(inode))
fsckcfg.logical_blocks += BLK_ROUND_UP(inode->sbi, inode->i_size);
fsckcfg.physical_blocks += BLK_ROUND_UP(inode->sbi, pchunk_len);
}
-out:
- if (raw)
- free(raw);
- if (buffer)
- free(buffer);
- return ret < 0 ? ret : 0;
-fail_eio:
- erofs_err("I/O error occurred when verifying data chunk @ nid %llu",
- inode->nid | 0ULL);
- ret = -EIO;
- goto out;
+ if (outfd >= 0 && ret >= 0)
+ ftruncate(outfd, inode->i_size);
+
+ erofs_mutex_destroy(&ctx.lock);
+ erofs_cond_destroy(&ctx.cond);
+ return ret < 0 ? ret : 0;
}
static inline int erofs_extract_dir(struct erofs_inode *inode)
@@ -1043,10 +1007,21 @@ int erofsfsck_fuzz_one(int argc, char *argv[])
int main(int argc, char *argv[])
#endif
{
+
int err;
+#ifdef EROFS_MT_ENABLED
+ int workers;
+#endif
erofs_init_configure();
+#ifdef EROFS_MT_ENABLED
+ workers = erofs_get_available_processors();
+ if (workers < 1)
+ workers = 1;
+ erofs_alloc_workqueue(&erofs_wq, workers, 256, NULL, NULL);
+#endif
+
fsckcfg.physical_blocks = 0;
fsckcfg.logical_blocks = 0;
fsckcfg.extract_path = NULL;
@@ -1181,6 +1156,9 @@ exit_dev_close:
exit:
erofs_blob_closeall(&g_sbi);
erofs_exit_configure();
+#ifdef EROFS_MT_ENABLED
+ erofs_destroy_workqueue(&erofs_wq);
+#endif
return err ? 1 : 0;
}
diff --git a/include/erofs/cond.h b/include/erofs/cond.h
new file mode 100644
index 0000000..90ec838
--- /dev/null
+++ b/include/erofs/cond.h
@@ -0,0 +1,31 @@
+/* SPDX-License-Identifier: GPL-2.0+ OR Apache-2.0 */
+#ifndef __EROFS_COND_H
+#define __EROFS_COND_H
+
+#include "lock.h"
+
+#if defined(HAVE_PTHREAD_H) && defined(EROFS_MT_ENABLED)
+#include <pthread.h>
+
+typedef pthread_cond_t erofs_cond_t;
+
+static inline void erofs_cond_init(erofs_cond_t *cond)
+{
+ pthread_cond_init(cond, NULL);
+}
+#define erofs_cond_wait pthread_cond_wait
+#define erofs_cond_signal pthread_cond_signal
+#define erofs_cond_broadcast pthread_cond_broadcast
+#define erofs_cond_destroy pthread_cond_destroy
+
+#else
+typedef struct {} erofs_cond_t;
+
+static inline void erofs_cond_init(erofs_cond_t *cond) {}
+static inline int erofs_cond_wait(erofs_cond_t *cond, erofs_mutex_t *mutex) { return 0; }
+static inline int erofs_cond_signal(erofs_cond_t *cond) { return 0; }
+static inline int erofs_cond_broadcast(erofs_cond_t *cond) { return 0; }
+static inline int erofs_cond_destroy(erofs_cond_t *cond) { return 0; }
+#endif
+
+#endif
\ No newline at end of file
diff --git a/include/erofs/internal.h b/include/erofs/internal.h
index 671880f..94f14da 100644
--- a/include/erofs/internal.h
+++ b/include/erofs/internal.h
@@ -25,6 +25,8 @@ typedef unsigned short umode_t;
#ifdef HAVE_PTHREAD_H
#include <pthread.h>
#endif
+#include <erofs/lock.h>
+#include "erofs/cond.h"
#include <stdlib.h>
#include <string.h>
#include "atomic.h"
@@ -62,6 +64,7 @@ struct erofs_buf {
#define erofs_pos(sbi, nr) ((erofs_off_t)(nr) << (sbi)->blkszbits)
#define BLK_ROUND_UP(sbi, addr) \
(roundup(addr, erofs_blksiz(sbi)) >> (sbi)->blkszbits)
+#define Z_EROFS_PCLUSTER_MAX_BATCH_SIZE 32
struct erofs_buffer_head;
struct erofs_bufmgr;
@@ -442,6 +445,20 @@ struct z_erofs_paramset {
char *extraopts;
};
+struct z_erofs_decompress_task;
+
+struct z_erofs_read_ctx {
+ erofs_mutex_t lock;
+ erofs_cond_t cond;
+ int pending_tasks;
+ int final_err;
+ int outfd;
+ bool free_out;
+ struct z_erofs_decompress_task *current_task;
+};
+
+void z_erofs_read_ctx_enqueue(struct z_erofs_read_ctx *ctx);
+
int liberofs_global_init(void);
void liberofs_global_exit(void);
@@ -478,7 +495,8 @@ int erofs_read_one_data(struct erofs_inode *inode, struct erofs_map_blocks *map,
char *buffer, u64 offset, size_t len);
int z_erofs_read_one_data(struct erofs_inode *inode,
struct erofs_map_blocks *map, char *raw, char *buffer,
- erofs_off_t skip, erofs_off_t length, bool trimmed);
+ erofs_off_t skip, erofs_off_t length, bool trimmed,
+ erofs_off_t out_offset, struct z_erofs_read_ctx *ctx);
void *erofs_read_metadata(struct erofs_sb_info *sbi, erofs_nid_t nid,
erofs_off_t *offset, int *lengthp);
int z_erofs_parse_cfgs(struct erofs_sb_info *sbi, struct erofs_super_block *dsb);
diff --git a/include/erofs/lock.h b/include/erofs/lock.h
index c6e3093..2e79d52 100644
--- a/include/erofs/lock.h
+++ b/include/erofs/lock.h
@@ -15,6 +15,7 @@ static inline void erofs_mutex_init(erofs_mutex_t *lock)
}
#define erofs_mutex_lock pthread_mutex_lock
#define erofs_mutex_unlock pthread_mutex_unlock
+#define erofs_mutex_destroy pthread_mutex_destroy
#define EROFS_DEFINE_MUTEX(lock) \
erofs_mutex_t lock = PTHREAD_MUTEX_INITIALIZER
@@ -29,12 +30,14 @@ static inline void erofs_init_rwsem(erofs_rwsem_t *lock)
#define erofs_down_write pthread_rwlock_wrlock
#define erofs_up_read pthread_rwlock_unlock
#define erofs_up_write pthread_rwlock_unlock
+
#else
typedef struct {} erofs_mutex_t;
static inline void erofs_mutex_init(erofs_mutex_t *lock) {}
static inline void erofs_mutex_lock(erofs_mutex_t *lock) {}
static inline void erofs_mutex_unlock(erofs_mutex_t *lock) {}
+static inline void erofs_mutex_destroy(erofs_mutex_t *lock) {}
#define EROFS_DEFINE_MUTEX(lock) \
erofs_mutex_t lock = {}
diff --git a/lib/data.c b/lib/data.c
index 6fd1389..e9d2218 100644
--- a/lib/data.c
+++ b/lib/data.c
@@ -9,6 +9,73 @@
#include "erofs/trace.h"
#include "erofs/decompress.h"
#include "liberofs_fragments.h"
+#include "erofs/workqueue.h"
+#include "erofs/lock.h"
+
+struct erofs_workqueue erofs_wq;
+
+struct z_erofs_decompress_item {
+ struct z_erofs_decompress_req req;
+ char *raw_buf;
+ char *out_buf;
+ erofs_off_t out_offset;
+ unsigned int out_length;
+};
+
+struct z_erofs_decompress_task {
+ struct erofs_work work;
+ struct z_erofs_read_ctx *ctx;
+ struct z_erofs_decompress_item items[Z_EROFS_PCLUSTER_MAX_BATCH_SIZE];
+ unsigned int nr_reqs;
+};
+
+static void z_erofs_decompress_worker(struct erofs_work *work, void *tlsp)
+{
+ struct z_erofs_decompress_task *task = (struct z_erofs_decompress_task *)work;
+ struct z_erofs_read_ctx *ctx = task->ctx;
+ int i, ret = 0, first_err = 0;
+
+ for (i = 0; i < task->nr_reqs; ++i) {
+ struct z_erofs_decompress_item *item = &task->items[i];
+ ret = z_erofs_decompress(&item->req);
+
+ if (ret >= 0 && ctx && ctx->outfd >= 0) {
+ if (pwrite(ctx->outfd, item->out_buf,
+ item->out_length, item->out_offset) < 0)
+ ret = -errno;
+ }
+
+ if (ret < 0 && first_err == 0)
+ first_err = ret;
+
+ free(item->raw_buf);
+ if (ctx && ctx->free_out)
+ free(item->out_buf);
+ }
+
+ if (ctx) {
+ erofs_mutex_lock(&ctx->lock);
+ if (first_err < 0 && ctx->final_err == 0)
+ ctx->final_err = first_err;
+ ctx->pending_tasks--;
+ if (ctx->pending_tasks == 0)
+ erofs_cond_signal(&ctx->cond);
+ erofs_mutex_unlock(&ctx->lock);
+ }
+ free(task);
+}
+
+void z_erofs_read_ctx_enqueue(struct z_erofs_read_ctx *ctx)
+{
+ if (ctx && ctx->current_task) {
+#ifdef EROFS_MT_ENABLED
+ erofs_queue_work(&erofs_wq, &ctx->current_task->work);
+#else
+ z_erofs_decompress_worker(&ctx->current_task->work, NULL);
+#endif
+ ctx->current_task = NULL;
+ }
+}
void *erofs_bread(struct erofs_buf *buf, erofs_off_t offset, bool need_kmap)
{
@@ -277,7 +344,8 @@ static int erofs_read_raw_data(struct erofs_inode *inode, char *buffer,
int z_erofs_read_one_data(struct erofs_inode *inode,
struct erofs_map_blocks *map, char *raw, char *buffer,
- erofs_off_t skip, erofs_off_t length, bool trimmed)
+ erofs_off_t skip, erofs_off_t length, bool trimmed,
+ erofs_off_t out_offset, struct z_erofs_read_ctx *ctx)
{
struct erofs_sb_info *sbi = inode->sbi;
struct erofs_map_dev mdev;
@@ -285,77 +353,107 @@ int z_erofs_read_one_data(struct erofs_inode *inode,
if (map->m_flags & __EROFS_MAP_FRAGMENT) {
if (__erofs_unlikely(inode->nid == sbi->packed_nid)) {
- erofs_err("fragment should not exist in the packed inode %llu",
- sbi->packed_nid | 0ULL);
- return -EFSCORRUPTED;
+ ret = -EFSCORRUPTED;
+ goto err_out;
+ }
+ ret = erofs_packedfile_read(sbi, buffer, length - skip,
+ inode->fragmentoff + skip);
+
+ if (ret >= 0 && ctx && ctx->outfd >= 0) {
+ if (pwrite(ctx->outfd, buffer, length - skip, out_offset) < 0)
+ ret = -errno;
}
- return erofs_packedfile_read(sbi, buffer, length - skip,
- inode->fragmentoff + skip);
+ goto err_out;
}
- /* no device id here, thus it will always succeed */
- mdev = (struct erofs_map_dev) {
- .m_pa = map->m_pa,
- };
+ mdev = (struct erofs_map_dev) { .m_pa = map->m_pa };
ret = erofs_map_dev(sbi, &mdev);
- if (ret) {
- DBG_BUGON(1);
- return ret;
- }
+ if (ret)
+ goto err_out;
ret = erofs_dev_read(sbi, mdev.m_deviceid, raw, mdev.m_pa, map->m_plen);
if (ret < 0)
- return ret;
+ goto err_out;
+ struct z_erofs_decompress_task *task = ctx->current_task;
+ if (!task) {
+ task = calloc(1, sizeof(*task));
+ task->ctx = ctx;
+ task->work.fn = z_erofs_decompress_worker;
+ ctx->current_task = task;
+
+ erofs_mutex_lock(&ctx->lock);
+ ctx->pending_tasks++;
+ erofs_mutex_unlock(&ctx->lock);
+ }
- ret = z_erofs_decompress(&(struct z_erofs_decompress_req) {
- .sbi = sbi,
- .in = raw,
- .out = buffer,
- .decodedskip = skip,
- .interlaced_offset =
- map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
- erofs_blkoff(sbi, map->m_la) : 0,
- .inputsize = map->m_plen,
- .decodedlength = length,
- .alg = map->m_algorithmformat,
- .partial_decoding = trimmed ? true :
- !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
- (map->m_flags & EROFS_MAP_PARTIAL_REF),
- });
- if (ret < 0)
- return ret;
+ int idx = task->nr_reqs++;
+ struct z_erofs_decompress_item *item = &task->items[idx];
+
+ item->req = (struct z_erofs_decompress_req) {
+ .sbi = sbi,
+ .in = raw,
+ .out = buffer,
+ .decodedskip = skip,
+ .interlaced_offset =
+ map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
+ erofs_blkoff(sbi, map->m_la) : 0,
+ .inputsize = map->m_plen,
+ .decodedlength = length,
+ .alg = map->m_algorithmformat,
+ .partial_decoding = trimmed ? true :
+ !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
+ (map->m_flags & EROFS_MAP_PARTIAL_REF),
+ };
+ item->raw_buf = raw;
+ item->out_buf = buffer;
+ item->out_offset = out_offset;
+ item->out_length = length;
+
+ int batch_limit = (map->m_algorithmformat == Z_EROFS_COMPRESSION_LZ4) ?
+ Z_EROFS_PCLUSTER_MAX_BATCH_SIZE : 8;
+
+ if (task->nr_reqs >= batch_limit) {
+ z_erofs_read_ctx_enqueue(ctx);
+ }
return 0;
+
+err_out:
+ if (ctx && ctx->free_out) free(buffer);
+ free(raw);
+ return ret;
}
static int z_erofs_read_data(struct erofs_inode *inode, char *buffer,
- erofs_off_t size, erofs_off_t offset)
+ erofs_off_t size, erofs_off_t offset)
{
erofs_off_t end, length, skip;
struct erofs_map_blocks map = {
.buf = __EROFS_BUF_INITIALIZER,
};
bool trimmed;
- unsigned int bufsize = 0;
- char *raw = NULL;
int ret = 0;
+ struct z_erofs_read_ctx ctx = {
+ .pending_tasks = 0,
+ .final_err = 0,
+ .outfd = -1,
+ .free_out = false,
+ .current_task = NULL
+ };
+ erofs_mutex_init(&ctx.lock);
+ erofs_cond_init(&ctx.cond);
+
end = offset + size;
while (end > offset) {
map.m_la = end - 1;
ret = z_erofs_map_blocks_iter(inode, &map, 0);
- if (ret)
- break;
+ if (ret) break;
- /*
- * trim to the needed size if the returned extent is quite
- * larger than requested, and set up partial flag as well.
- */
if (end < map.m_la + map.m_llen) {
length = end - map.m_la;
trimmed = true;
} else {
- DBG_BUGON(end != map.m_la + map.m_llen);
length = map.m_llen;
trimmed = false;
}
@@ -374,25 +472,31 @@ static int z_erofs_read_data(struct erofs_inode *inode, char *buffer,
continue;
}
- if (map.m_plen > bufsize) {
- char *newraw;
-
- bufsize = map.m_plen;
- newraw = realloc(raw, bufsize);
- if (!newraw) {
- ret = -ENOMEM;
- break;
- }
- raw = newraw;
+ char *raw = malloc(map.m_plen);
+ if (!raw) {
+ ret = -ENOMEM;
+ break;
}
ret = z_erofs_read_one_data(inode, &map, raw,
- buffer + end - offset, skip, length, trimmed);
- if (ret < 0)
+ buffer + end - offset, skip, length, trimmed, 0, &ctx);
+ if (ret < 0) {
break;
+ }
}
- if (raw)
- free(raw);
+ z_erofs_read_ctx_enqueue(&ctx);
+
+ erofs_mutex_lock(&ctx.lock);
+ while (ctx.pending_tasks > 0)
+ erofs_cond_wait(&ctx.cond, &ctx.lock);
+
+ if (ctx.final_err < 0 && ret == 0)
+ ret = ctx.final_err;
+
+ erofs_mutex_unlock(&ctx.lock);
+ erofs_mutex_destroy(&ctx.lock);
+ erofs_cond_destroy(&ctx.cond);
+
return ret < 0 ? ret : 0;
}
--
2.52.0
^ permalink raw reply related [flat|nested] 13+ messages in thread
* [PATCH 2/2] fsck.erofs: implement concurrent directory traversal
2026-06-21 12:01 [PATCH 0/2] fsck.erofs: implement multi-threaded extraction Nithurshen
2026-06-21 12:01 ` [PATCH 1/2] fsck.erofs: add multi-threaded decompression Nithurshen
@ 2026-06-21 12:01 ` Nithurshen
2026-07-05 14:05 ` Gao Xiang
2026-07-14 1:10 ` [PATCH v5] fsck.erofs: add multi-threaded decompression Nithurshen
1 sibling, 2 replies; 13+ messages in thread
From: Nithurshen @ 2026-06-21 12:01 UTC (permalink / raw)
To: linux-erofs; +Cc: hsiangkao, xiang, Nithurshen
Currently, fsck.erofs traverses the filesystem tree and verifies
inodes synchronously on the main thread. While decompression
compute is offloaded, the main thread remains a bottleneck
during the I/O-heavy directory walk.
This patch parallelizes the directory traversal and inode
extraction processes. To achieve this safely, the globally shared
fsckcfg.extract_path and fsckcfg.dirstack states are decoupled
and localized into individual struct erofsfsck_inode_task
payloads, which are deep-copied and handed off to the worker
pool. Global statistics and hardlink tables are secured using
native erofs_mutex_t primitives.
To prevent thread pool exhaustion deadlocks—where workers
processing a deep directory tree occupy all available execution
slots and block on erofs_cond_wait, starving their own spawned
decompression tasks—this patch introduces a dedicated
erofs_traverse_wq. By isolating the producers (traversal and
verification) from the consumers (pcluster decompression), the
pipeline avoids gridlock.
Signed-off-by: Nithurshen <nithurshen.dev@gmail.com>
---
fsck/main.c | 300 ++++++++++++++++++++++++++++++++++++----------------
1 file changed, 211 insertions(+), 89 deletions(-)
diff --git a/fsck/main.c b/fsck/main.c
index ffe7e29..0ed0ddd 100644
--- a/fsck/main.c
+++ b/fsck/main.c
@@ -19,14 +19,45 @@
#include "../lib/liberofs_compress.h"
extern struct erofs_workqueue erofs_wq;
+struct erofs_workqueue erofs_traverse_wq;
-static int erofsfsck_check_inode(erofs_nid_t pnid, erofs_nid_t nid);
+static EROFS_DEFINE_MUTEX(fsck_global_lock);
+static EROFS_DEFINE_MUTEX(fsck_hardlink_lock);
+
+static int fsck_pending_tasks = 0;
+static EROFS_DEFINE_MUTEX(fsck_task_lock);
+static erofs_cond_t fsck_task_cond;
+
+static void fsck_inc_task(void) {
+ erofs_mutex_lock(&fsck_task_lock);
+ fsck_pending_tasks++;
+ erofs_mutex_unlock(&fsck_task_lock);
+}
+
+static void fsck_dec_task(void) {
+ erofs_mutex_lock(&fsck_task_lock);
+ fsck_pending_tasks--;
+ if (fsck_pending_tasks == 0)
+ erofs_cond_signal(&fsck_task_cond);
+ erofs_mutex_unlock(&fsck_task_lock);
+}
+
+static void fsck_wait_tasks(void) {
+ erofs_mutex_lock(&fsck_task_lock);
+ while (fsck_pending_tasks > 0)
+ erofs_cond_wait(&fsck_task_cond, &fsck_task_lock);
+ erofs_mutex_unlock(&fsck_task_lock);
+}
struct erofsfsck_dirstack {
erofs_nid_t dirs[PATH_MAX];
int top;
};
+static int erofsfsck_check_inode(erofs_nid_t pnid, erofs_nid_t nid,
+ char *current_path, size_t pos,
+ struct erofsfsck_dirstack *dirstack);
+
struct erofsfsck_cfg {
struct erofsfsck_dirstack dirstack;
u64 physical_blocks;
@@ -425,7 +456,7 @@ out:
return ret;
}
-static int erofsfsck_dump_xattrs(struct erofs_inode *inode)
+static int erofsfsck_dump_xattrs(struct erofs_inode *inode, const char *current_path)
{
static bool ignore_xattrs = false;
char *keylst, *key;
@@ -472,7 +503,7 @@ static int erofsfsck_dump_xattrs(struct erofs_inode *inode)
break;
}
if (fsckcfg.extract_path)
- ret = erofs_sys_lsetxattr(fsckcfg.extract_path, key,
+ ret = erofs_sys_lsetxattr(current_path, key,
value, size);
else
ret = 0;
@@ -587,9 +618,11 @@ out:
erofs_mutex_unlock(&ctx.lock);
if (fsckcfg.print_comp_ratio) {
+ erofs_mutex_lock(&fsck_global_lock);
if (!erofs_is_packed_inode(inode))
fsckcfg.logical_blocks += BLK_ROUND_UP(inode->sbi, inode->i_size);
fsckcfg.physical_blocks += BLK_ROUND_UP(inode->sbi, pchunk_len);
+ erofs_mutex_unlock(&fsck_global_lock);
}
if (outfd >= 0 && ret >= 0)
@@ -600,11 +633,11 @@ out:
return ret < 0 ? ret : 0;
}
-static inline int erofs_extract_dir(struct erofs_inode *inode)
+static inline int erofs_extract_dir(struct erofs_inode *inode, const char *current_path)
{
int ret;
- erofs_dbg("create directory %s", fsckcfg.extract_path);
+ erofs_dbg("create directory %s", current_path);
/* verify data chunk layout */
ret = erofs_verify_inode_data(inode, -1);
@@ -617,19 +650,19 @@ static inline int erofs_extract_dir(struct erofs_inode *inode)
* write/execute permission. These are fixed up later in
* erofsfsck_set_attributes().
*/
- if (mkdir(fsckcfg.extract_path, 0700) < 0) {
+ if (mkdir(current_path, 0700) < 0) {
struct stat st;
if (errno != EEXIST) {
erofs_err("failed to create directory: %s (%s)",
- fsckcfg.extract_path, strerror(errno));
+ current_path, strerror(errno));
return -errno;
}
- if (lstat(fsckcfg.extract_path, &st) ||
+ if (lstat(current_path, &st) ||
!S_ISDIR(st.st_mode)) {
erofs_err("path is not a directory: %s",
- fsckcfg.extract_path);
+ current_path);
return -ENOTDIR;
}
@@ -637,9 +670,9 @@ static inline int erofs_extract_dir(struct erofs_inode *inode)
* Try to change permissions of existing directory so
* that we can write to it
*/
- if (chmod(fsckcfg.extract_path, 0700) < 0) {
+ if (chmod(current_path, 0700) < 0) {
erofs_err("failed to set permissions: %s (%s)",
- fsckcfg.extract_path, strerror(errno));
+ current_path, strerror(errno));
return -errno;
}
}
@@ -651,11 +684,17 @@ static char *erofsfsck_hardlink_find(erofs_nid_t nid)
struct list_head *head =
&erofsfsck_link_hashtable[nid % NR_HARDLINK_HASHTABLE];
struct erofsfsck_hardlink_entry *entry;
+ char *path = NULL;
- list_for_each_entry(entry, head, list)
- if (entry->nid == nid)
+ erofs_mutex_lock(&fsck_hardlink_lock);
+ list_for_each_entry(entry, head, list){
+ if (entry->nid == nid){
return entry->path;
- return NULL;
+ break;
+ }
+ }
+ erofs_mutex_unlock(&fsck_hardlink_lock);
+ return path;
}
static int erofsfsck_hardlink_insert(erofs_nid_t nid, const char *path)
@@ -673,8 +712,10 @@ static int erofsfsck_hardlink_insert(erofs_nid_t nid, const char *path)
return -ENOMEM;
}
+ erofs_mutex_lock(&fsck_hardlink_lock);
list_add_tail(&entry->list,
&erofsfsck_link_hashtable[nid % NR_HARDLINK_HASHTABLE]);
+ erofs_mutex_unlock(&fsck_hardlink_lock);
return 0;
}
@@ -703,37 +744,37 @@ static void erofsfsck_hardlink_exit(void)
}
}
-static inline int erofs_extract_file(struct erofs_inode *inode)
+static inline int erofs_extract_file(struct erofs_inode *inode, const char *current_path)
{
bool tryagain = true;
int ret, fd;
- erofs_dbg("extract file to path: %s", fsckcfg.extract_path);
+ erofs_dbg("extract file to path: %s", current_path);
again:
- fd = open(fsckcfg.extract_path,
+ fd = open(current_path,
O_WRONLY | O_CREAT | O_NOFOLLOW |
(fsckcfg.overwrite ? O_TRUNC : O_EXCL), 0700);
if (fd < 0) {
if (fsckcfg.overwrite && tryagain) {
if (errno == EISDIR) {
erofs_warn("try to forcely remove directory %s",
- fsckcfg.extract_path);
- if (rmdir(fsckcfg.extract_path) < 0) {
+ current_path);
+ if (rmdir(current_path) < 0) {
erofs_err("failed to remove: %s (%s)",
- fsckcfg.extract_path, strerror(errno));
+ current_path, strerror(errno));
return -EISDIR;
}
} else if (errno == EACCES &&
- chmod(fsckcfg.extract_path, 0700) < 0) {
+ chmod(current_path, 0700) < 0) {
erofs_err("failed to set permissions: %s (%s)",
- fsckcfg.extract_path, strerror(errno));
+ current_path, strerror(errno));
return -errno;
}
tryagain = false;
goto again;
}
- erofs_err("failed to open: %s (%s)", fsckcfg.extract_path,
+ erofs_err("failed to open: %s (%s)", current_path,
strerror(errno));
return -errno;
}
@@ -744,14 +785,14 @@ again:
return ret;
}
-static inline int erofs_extract_symlink(struct erofs_inode *inode)
+static inline int erofs_extract_symlink(struct erofs_inode *inode, const char *current_path)
{
struct erofs_vfile vf;
bool tryagain = true;
int ret;
char *buf = NULL;
- erofs_dbg("extract symlink to path: %s", fsckcfg.extract_path);
+ erofs_dbg("extract symlink to path: %s", current_path);
/* verify data chunk layout */
ret = erofs_verify_inode_data(inode, -1);
@@ -777,13 +818,13 @@ static inline int erofs_extract_symlink(struct erofs_inode *inode)
buf[inode->i_size] = '\0';
again:
- if (symlink(buf, fsckcfg.extract_path) < 0) {
+ if (symlink(buf, current_path) < 0) {
if (errno == EEXIST && fsckcfg.overwrite && tryagain) {
erofs_warn("try to forcely remove file %s",
- fsckcfg.extract_path);
- if (unlink(fsckcfg.extract_path) < 0) {
+ current_path);
+ if (unlink(current_path) < 0) {
erofs_err("failed to remove: %s",
- fsckcfg.extract_path);
+ current_path);
ret = -errno;
goto out;
}
@@ -791,7 +832,7 @@ again:
goto again;
}
erofs_err("failed to create symlink: %s",
- fsckcfg.extract_path);
+ current_path);
ret = -errno;
}
out:
@@ -800,12 +841,12 @@ out:
return ret;
}
-static int erofs_extract_special(struct erofs_inode *inode)
+static int erofs_extract_special(struct erofs_inode *inode, const char *current_path)
{
bool tryagain = true;
int ret;
- erofs_dbg("extract special to path: %s", fsckcfg.extract_path);
+ erofs_dbg("extract special to path: %s", current_path);
/* verify data chunk layout */
ret = erofs_verify_inode_data(inode, -1);
@@ -813,13 +854,13 @@ static int erofs_extract_special(struct erofs_inode *inode)
return ret;
again:
- if (mknod(fsckcfg.extract_path, inode->i_mode, inode->u.i_rdev) < 0) {
+ if (mknod(current_path, inode->i_mode, inode->u.i_rdev) < 0) {
if (errno == EEXIST && fsckcfg.overwrite && tryagain) {
erofs_warn("try to forcely remove file %s",
- fsckcfg.extract_path);
- if (unlink(fsckcfg.extract_path) < 0) {
+ current_path);
+ if (unlink(current_path) < 0) {
erofs_err("failed to remove: %s",
- fsckcfg.extract_path);
+ current_path);
return -errno;
}
tryagain = false;
@@ -827,11 +868,11 @@ again:
}
if (errno == EEXIST || fsckcfg.superuser) {
erofs_err("failed to create special file: %s",
- fsckcfg.extract_path);
+ current_path);
ret = -errno;
} else {
erofs_warn("failed to create special file: %s, skipped",
- fsckcfg.extract_path);
+ current_path);
ret = -ECANCELED;
}
}
@@ -854,47 +895,82 @@ static int erofsfsck_get_parent_cb(struct erofs_dir_context *ctx)
return 0;
}
+struct erofsfsck_inode_task {
+ struct erofs_work work;
+ erofs_nid_t pnid;
+ erofs_nid_t nid;
+ char *path;
+ size_t path_pos;
+ struct erofsfsck_dirstack dirstack;
+};
+
+static void erofsfsck_check_inode_worker(struct erofs_work *work, void *tlsp);
+
+struct erofsfsck_dir_iter_ctx {
+ struct erofs_dir_context ctx;
+ char *base_path;
+ size_t base_pos;
+ struct erofsfsck_dirstack *dirstack;
+};
+
static int erofsfsck_dirent_iter(struct erofs_dir_context *ctx)
{
- int ret;
- size_t prev_pos, curr_pos;
+ struct erofsfsck_dir_iter_ctx *ictx = (void *)ctx;
+ size_t curr_pos = ictx->base_pos;
+ char *new_path = NULL;
if (ctx->dot_dotdot)
return 0;
- prev_pos = fsckcfg.extract_pos;
- curr_pos = prev_pos;
-
- if (prev_pos + ctx->de_namelen >= PATH_MAX) {
- erofs_err("unable to fsck since the path is too long (%llu)",
- (curr_pos + ctx->de_namelen) | 0ULL);
- return -EOPNOTSUPP;
- }
+ if (curr_pos + ctx->de_namelen >= PATH_MAX) {
+ erofs_err("unable to fsck since the path is too long");
+ return -EOPNOTSUPP;
+ }
if (fsckcfg.extract_path) {
- fsckcfg.extract_path[curr_pos++] = '/';
- strncpy(fsckcfg.extract_path + curr_pos, ctx->dname,
- ctx->de_namelen);
+ new_path = malloc(PATH_MAX);
+ if (!new_path) return -ENOMEM;
+
+ if (ictx->base_path)
+ memcpy(new_path, ictx->base_path, curr_pos);
+
+ new_path[curr_pos++] = '/';
+ strncpy(new_path + curr_pos, ctx->dname, ctx->de_namelen);
curr_pos += ctx->de_namelen;
- fsckcfg.extract_path[curr_pos] = '\0';
+ new_path[curr_pos] = '\0';
} else {
curr_pos += ctx->de_namelen;
}
- fsckcfg.extract_pos = curr_pos;
- ret = erofsfsck_check_inode(ctx->dir->nid, ctx->de_nid);
- if (fsckcfg.extract_path)
- fsckcfg.extract_path[prev_pos] = '\0';
- fsckcfg.extract_pos = prev_pos;
- return ret;
+ struct erofsfsck_inode_task *task = malloc(sizeof(*task));
+ if (!task) {
+ free(new_path);
+ return -ENOMEM;
+ }
+
+ task->work.fn = erofsfsck_check_inode_worker;
+ task->work.next = NULL;
+ task->pnid = ctx->dir->nid;
+ task->nid = ctx->de_nid;
+ task->path = new_path;
+ task->path_pos = curr_pos;
+ task->dirstack = *(ictx->dirstack);
+
+ fsck_inc_task();
+#ifdef EROFS_MT_ENABLED
+ erofs_queue_work(&erofs_traverse_wq, &task->work);
+#else
+ task->work.fn(&task->work, NULL);
+#endif
+ return 0;
}
-static int erofsfsck_extract_inode(struct erofs_inode *inode)
+static int erofsfsck_extract_inode(struct erofs_inode *inode, const char *current_path)
{
int ret;
char *oldpath;
- if (!fsckcfg.extract_path || erofs_is_packed_inode(inode)) {
+ if (!current_path || erofs_is_packed_inode(inode)) {
verify:
/* verify data chunk layout */
return erofs_verify_inode_data(inode, -1);
@@ -902,9 +978,9 @@ verify:
oldpath = erofsfsck_hardlink_find(inode->nid);
if (oldpath) {
- if (link(oldpath, fsckcfg.extract_path) == -1) {
+ if (link(oldpath, current_path) == -1) {
erofs_err("failed to extract hard link: %s (%s)",
- fsckcfg.extract_path, strerror(errno));
+ current_path, strerror(errno));
return -errno;
}
return 0;
@@ -912,19 +988,19 @@ verify:
switch (inode->i_mode & S_IFMT) {
case S_IFDIR:
- ret = erofs_extract_dir(inode);
+ ret = erofs_extract_dir(inode, current_path);
break;
case S_IFREG:
- ret = erofs_extract_file(inode);
+ ret = erofs_extract_file(inode, current_path);
break;
case S_IFLNK:
- ret = erofs_extract_symlink(inode);
+ ret = erofs_extract_symlink(inode, current_path);
break;
case S_IFCHR:
case S_IFBLK:
case S_IFIFO:
case S_IFSOCK:
- ret = erofs_extract_special(inode);
+ ret = erofs_extract_special(inode, current_path);
break;
default:
/* TODO */
@@ -936,11 +1012,13 @@ verify:
/* record nid and old path for hardlink */
if (inode->i_nlink > 1 && !S_ISDIR(inode->i_mode))
ret = erofsfsck_hardlink_insert(inode->nid,
- fsckcfg.extract_path);
+ current_path);
return ret;
}
-static int erofsfsck_check_inode(erofs_nid_t pnid, erofs_nid_t nid)
+static int erofsfsck_check_inode(erofs_nid_t pnid, erofs_nid_t nid,
+ char *current_path, size_t pos,
+ struct erofsfsck_dirstack *dirstack)
{
int ret, i;
struct erofs_inode inode = {.sbi = &g_sbi, .nid = nid};
@@ -961,44 +1039,63 @@ static int erofsfsck_check_inode(erofs_nid_t pnid, erofs_nid_t nid)
goto out;
}
- ret = erofsfsck_extract_inode(&inode);
+ ret = erofsfsck_extract_inode(&inode, current_path);
if (ret && ret != -ECANCELED)
goto out;
if (fsckcfg.check_decomp && fsckcfg.dump_xattrs) {
- ret = erofsfsck_dump_xattrs(&inode);
+ ret = erofsfsck_dump_xattrs(&inode, current_path);
if (ret)
return ret;
}
if (S_ISDIR(inode.i_mode)) {
- struct erofs_dir_context ctx = {
- .flags = EROFS_READDIR_VALID_PNID,
- .pnid = pnid,
- .dir = &inode,
- .cb = erofsfsck_dirent_iter,
+ struct erofsfsck_dir_iter_ctx ctx = {
+ .ctx.flags = EROFS_READDIR_VALID_PNID,
+ .ctx.pnid = pnid,
+ .ctx.dir = &inode,
+ .ctx.cb = erofsfsck_dirent_iter,
+ .base_path = current_path,
+ .base_pos = pos,
+ .dirstack = dirstack
};
- /* XXX: support the deeper cases later */
- if (fsckcfg.dirstack.top >= ARRAY_SIZE(fsckcfg.dirstack.dirs))
+ if (dirstack->top >= ARRAY_SIZE(dirstack->dirs))
return -ENAMETOOLONG;
- for (i = 0; i < fsckcfg.dirstack.top; ++i)
- if (inode.nid == fsckcfg.dirstack.dirs[i])
+ for (i = 0; i < dirstack->top; ++i)
+ if (inode.nid == dirstack->dirs[i])
return -ELOOP;
- fsckcfg.dirstack.dirs[fsckcfg.dirstack.top++] = pnid;
- ret = erofs_iterate_dir(&ctx, true);
- --fsckcfg.dirstack.top;
+
+ dirstack->dirs[dirstack->top++] = pnid;
+ ret = erofs_iterate_dir(&ctx.ctx, true);
+ --dirstack->top;
}
- if (!ret && !erofs_is_packed_inode(&inode))
- erofsfsck_set_attributes(&inode, fsckcfg.extract_path);
+ if (!ret && !erofs_is_packed_inode(&inode) && current_path)
+ erofsfsck_set_attributes(&inode, current_path);
if (ret == -ECANCELED)
ret = 0;
out:
- if (ret && ret != -EIO)
- fsckcfg.corrupted = true;
- return ret;
+ if (ret && ret != -EIO) {
+ erofs_mutex_lock(&fsck_global_lock);
+ fsckcfg.corrupted = true;
+ erofs_mutex_unlock(&fsck_global_lock);
+ }
+ return ret;
+}
+
+static void erofsfsck_check_inode_worker(struct erofs_work *work, void *tlsp)
+{
+ struct erofsfsck_inode_task *task = (struct erofsfsck_inode_task *)work;
+
+ erofsfsck_check_inode(task->pnid, task->nid, task->path,
+ task->path_pos, &task->dirstack);
+
+ if (task->path)
+ free(task->path);
+ free(task);
+ fsck_dec_task();
}
#ifdef FUZZING
@@ -1020,6 +1117,7 @@ int main(int argc, char *argv[])
if (workers < 1)
workers = 1;
erofs_alloc_workqueue(&erofs_wq, workers, 256, NULL, NULL);
+ erofs_alloc_workqueue(&erofs_traverse_wq, workers, 256, NULL, NULL);
#endif
fsckcfg.physical_blocks = 0;
@@ -1086,6 +1184,8 @@ int main(int argc, char *argv[])
fsckcfg.nid = g_sbi.root_nid;
}
+ erofs_cond_init(&fsck_task_cond);
+
if (!fsckcfg.inode_path && fsckcfg.nid == g_sbi.root_nid) {
if (erofs_sb_has_fragments(&g_sbi) && g_sbi.packed_nid > 0) {
err = erofs_packedfile_init(&g_sbi, false);
@@ -1095,7 +1195,7 @@ int main(int argc, char *argv[])
goto exit_hardlink;
}
- err = erofsfsck_check_inode(g_sbi.packed_nid, g_sbi.packed_nid);
+ err = erofsfsck_check_inode(g_sbi.packed_nid, g_sbi.packed_nid, NULL, 0, &fsckcfg.dirstack);
if (err) {
erofs_err("failed to verify packed file");
goto exit_packedinode;
@@ -1120,7 +1220,26 @@ int main(int argc, char *argv[])
pnid = ctx.pnid;
}
}
- err = erofsfsck_check_inode(pnid, fsckcfg.nid);
+
+ fsck_inc_task();
+ struct erofsfsck_inode_task *root_task = calloc(1, sizeof(*root_task));
+ root_task->work.fn = erofsfsck_check_inode_worker;
+ root_task->pnid = pnid;
+ root_task->nid = fsckcfg.nid;
+
+ if (fsckcfg.extract_path)
+ root_task->path = strdup(fsckcfg.extract_path);
+
+ root_task->path_pos = fsckcfg.extract_pos;
+ root_task->dirstack = fsckcfg.dirstack;
+
+#ifdef EROFS_MT_ENABLED
+ erofs_queue_work(&erofs_traverse_wq, &root_task->work);
+#else
+ root_task->work.fn(&root_task->work, NULL);
+#endif
+
+ fsck_wait_tasks();
}
if (fsckcfg.corrupted) {
@@ -1144,6 +1263,8 @@ int main(int argc, char *argv[])
}
}
+ erofs_cond_destroy(&fsck_task_cond);
+
exit_packedinode:
erofs_packedfile_exit(&g_sbi);
exit_hardlink:
@@ -1158,6 +1279,7 @@ exit:
erofs_exit_configure();
#ifdef EROFS_MT_ENABLED
erofs_destroy_workqueue(&erofs_wq);
+ erofs_destroy_workqueue(&erofs_traverse_wq);
#endif
return err ? 1 : 0;
}
--
2.52.0
^ permalink raw reply related [flat|nested] 13+ messages in thread
* Re: [PATCH 1/2] fsck.erofs: add multi-threaded decompression
2026-06-21 12:01 ` [PATCH 1/2] fsck.erofs: add multi-threaded decompression Nithurshen
@ 2026-06-22 2:08 ` Gao Xiang
2026-06-29 9:55 ` [PATCH 1/2 v2] " Nithurshen
2026-07-06 6:05 ` [PATCH 1/2 v3] " Nithurshen
2 siblings, 0 replies; 13+ messages in thread
From: Gao Xiang @ 2026-06-22 2:08 UTC (permalink / raw)
To: Nithurshen, linux-erofs; +Cc: xiang
Hi Nithurshen,
On 2026/6/21 20:01, Nithurshen wrote:
> Currently, fsck.erofs extracts files synchronously. When decompressing
> heavily packed images, the main thread spends the majority of its time
> blocked on a combination of synchronous vfs_write() syscalls and
> decompression routines, bottlenecking overall extraction speed.
>
> This patch introduces a scalable, multi-threaded decompression framework
> using the existing erofs_workqueue infrastructure to decouple compute
> from the main thread's I/O.
>
> To prevent massive scheduling overhead (futex contention) where worker
> threads spend more CPU time waking up than actually decompressing small
> clusters, this implementation introduces a batching context. Because
> different compression algorithms exhibit vastly different scheduling
> thresholds, the batch size is algorithm-aware:
> - Fast algorithms like LZ4 utilize a larger batch limit (up to 32
> pclusters) to effectively hide synchronization overhead.
> - Compute-heavy algorithms like LZMA or ZSTD trigger at a lower
> threshold (8 pclusters) to prevent memory bloat and thread starvation.
>
> Key details of this implementation:
> - The worker pool is dynamically sized based on available system CPUs.
> - Decompression tasks take strict ownership of the raw and output
> buffers (safely tracking memory via a `free_out` flag) to prevent
> data races and memory leaks.
> - Output buffers are explicitly zero-initialized via calloc() to
> prevent trailing garbage bytes from leaking into extracted files.
> - Tail-end packed fragments are processed synchronously by the main
> thread, as their minimal overhead does not benefit from asynchronous
> offloading.
>
> Signed-off-by: Nithurshen <nithurshen.dev@gmail.com>
> ---
> fsck/main.c | 150 ++++++++++++---------------
> include/erofs/cond.h | 31 ++++++
> include/erofs/internal.h | 20 +++-
> include/erofs/lock.h | 3 +
> lib/data.c | 216 +++++++++++++++++++++++++++++----------
> 5 files changed, 277 insertions(+), 143 deletions(-)
> create mode 100644 include/erofs/cond.h
>
> diff --git a/fsck/main.c b/fsck/main.c
> index 16cc627..ffe7e29 100644
> --- a/fsck/main.c
> +++ b/fsck/main.c
> @@ -8,14 +8,18 @@
> #include <time.h>
> #include <utime.h>
> #include <unistd.h>
> +#include "erofs/lock.h"
> #include <sys/stat.h>
> #include "erofs/print.h"
> #include "erofs/decompress.h"
> #include "erofs/dir.h"
> #include "erofs/xattr.h"
> +#include "erofs/workqueue.h"
> #include "../lib/compressor.h"
> #include "../lib/liberofs_compress.h"
>
> +extern struct erofs_workqueue erofs_wq;
> +
> static int erofsfsck_check_inode(erofs_nid_t pnid, erofs_nid_t nid);
>
> struct erofsfsck_dirstack {
> @@ -505,44 +509,36 @@ out:
>
> static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd)
> {
> - struct erofs_map_blocks map = {
> - .buf = __EROFS_BUF_INITIALIZER,
> - };
> + struct erofs_map_blocks map = { .buf = __EROFS_BUF_INITIALIZER };
If it's unrelated to the change, let's not touch it.
> bool needdecode = fsckcfg.check_decomp && !erofs_is_packed_inode(inode);
> int ret = 0;
> - bool compressed;
> + bool compressed = erofs_inode_is_data_compressed(inode->datalayout);
> erofs_off_t pos = 0;
> u64 pchunk_len = 0;
> - unsigned int raw_size = 0, buffer_size = 0;
> - char *raw = NULL, *buffer = NULL;
>
> - erofs_dbg("verify data chunk of nid(%llu): type(%d)",
> - inode->nid | 0ULL, inode->datalayout);
> + struct z_erofs_read_ctx ctx = {
maybe rename it as z_erofs_mt_read_ctx since it relates to multi-threading.
> + .pending_tasks = 0,
> + .final_err = 0,
> + .outfd = outfd,
> + .free_out = true,
> + .current_task = NULL
> + };
> + erofs_mutex_init(&ctx.lock);
> + erofs_cond_init(&ctx.cond);
>
> - compressed = erofs_inode_is_data_compressed(inode->datalayout);
> - while (pos < inode->i_size) {
> - unsigned int alloc_rawsize;
> + erofs_dbg("verify data chunk of nid(%llu): type(%d)", inode->nid | 0ULL, inode->datalayout);
Can we move this to the original position? Unnecessary diff.
>
> + while (pos < inode->i_size) {
> map.m_la = pos;
> ret = erofs_map_blocks(inode, &map, EROFS_GET_BLOCKS_FIEMAP);
> - if (ret)
> - goto out;
> -
> - if (!compressed && map.m_llen != map.m_plen) {
> - erofs_err("broken chunk length m_la %" PRIu64 " m_llen %" PRIu64 " m_plen %" PRIu64,
> - map.m_la, map.m_llen, map.m_plen);
> - ret = -EFSCORRUPTED;
> - goto out;
> - }
> + if (ret) goto out;
The code style is still wrong here.
>
> - /* the last lcluster can be divided into 3 parts */
> if (map.m_la + map.m_llen > inode->i_size)
> map.m_llen = inode->i_size - map.m_la;
>
> pchunk_len += map.m_plen;
> pos += map.m_llen;
>
> - /* should skip decomp? */
> if (map.m_la >= inode->i_size || !needdecode)
> continue;
>
> @@ -555,85 +551,53 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd)
> continue;
> }
>
> - if (map.m_plen > Z_EROFS_PCLUSTER_MAX_SIZE) {
> - if (compressed && !(map.m_flags & __EROFS_MAP_FRAGMENT)) {
> - erofs_err("invalid pcluster size %" PRIu64 " @ offset %" PRIu64 " of nid %" PRIu64,
> - map.m_plen, map.m_la,
> - inode->nid | 0ULL);
> - ret = -EFSCORRUPTED;
> - goto out;
> - }
> - alloc_rawsize = Z_EROFS_PCLUSTER_MAX_SIZE;
> - } else {
> - alloc_rawsize = map.m_plen;
> - }
> -
> - if (alloc_rawsize > raw_size) {
> - char *newraw = realloc(raw, alloc_rawsize);
> -
> - if (!newraw) {
> + if (compressed) {
> + char *raw = malloc(map.m_plen);
> + size_t buffer_size = map.m_llen > erofs_blksiz(inode->sbi) ? map.m_llen : erofs_blksiz(inode->sbi);
Please avoid overly long lines (< 80 chars recommended, but should
less than 100 chars)
> + char *buffer = calloc(1, buffer_size);
I hope it could be:
char *raw, *buffers;
raw = malloc(map.m_plen);
buffer = calloc(1, buffer_size);
> +
> + if (!raw || !buffer) {
> + free(raw); free(buffer);
> ret = -ENOMEM;
> goto out;
> }
> - raw = newraw;
> - raw_size = alloc_rawsize;
> - }
>
> - if (compressed) {
> - if (map.m_llen > buffer_size) {
> - char *newbuffer;
> -
> - buffer_size = map.m_llen;
> - newbuffer = realloc(buffer, buffer_size);
> - if (!newbuffer) {
> - ret = -ENOMEM;
> - goto out;
> - }
> - buffer = newbuffer;
> - }
> - ret = z_erofs_read_one_data(inode, &map, raw, buffer,
> - 0, map.m_llen, false);
> - if (ret)
> + ret = z_erofs_read_one_data(inode, &map, raw, buffer, 0, map.m_llen, false, map.m_la, &ctx);
Same code styling issue here.
> + if (ret) {
> + /* DO NOT free(raw) or free(buffer) here. z_erofs_read_one_data took ownership! */
> goto out;
> -
> - if (outfd >= 0 && write(outfd, buffer, map.m_llen) < 0)
> - goto fail_eio;
> + }
> } else {
> - u64 p = 0;
> -
> - do {
> - u64 count = min_t(u64, alloc_rawsize,
> - map.m_llen);
> -
> - ret = erofs_read_one_data(inode, &map, raw, p, count);
> - if (ret)
> - goto out;
> -
> - if (outfd >= 0 && write(outfd, raw, count) < 0)
> - goto fail_eio;
> - map.m_llen -= count;
> - p += count;
> - } while (map.m_llen);
> + char *raw = calloc(1, map.m_llen);
> + ret = erofs_read_one_data(inode, &map, raw, 0, map.m_llen);
> + if (ret >= 0 && outfd >= 0)
> + pwrite(outfd, raw, map.m_llen, map.m_la);
> + free(raw);
> + if (ret) goto out;
> }
> }
> + z_erofs_read_ctx_enqueue(&ctx);
Maybe just call it as `z_erofs_mt_read_enqueue(&ctx);`
> +
> +out:
> + erofs_mutex_lock(&ctx.lock);
> + while (ctx.pending_tasks > 0)
> + erofs_cond_wait(&ctx.cond, &ctx.lock);
> + if (ctx.final_err < 0 && ret >= 0)
> + ret = ctx.final_err;
> + erofs_mutex_unlock(&ctx.lock);
>
> if (fsckcfg.print_comp_ratio) {
> if (!erofs_is_packed_inode(inode))
> fsckcfg.logical_blocks += BLK_ROUND_UP(inode->sbi, inode->i_size);
> fsckcfg.physical_blocks += BLK_ROUND_UP(inode->sbi, pchunk_len);
> }
> -out:
> - if (raw)
> - free(raw);
> - if (buffer)
> - free(buffer);
> - return ret < 0 ? ret : 0;
>
> -fail_eio:
> - erofs_err("I/O error occurred when verifying data chunk @ nid %llu",
> - inode->nid | 0ULL);
> - ret = -EIO;
> - goto out;
> + if (outfd >= 0 && ret >= 0)
> + ftruncate(outfd, inode->i_size);
> +
> + erofs_mutex_destroy(&ctx.lock);
> + erofs_cond_destroy(&ctx.cond);
> + return ret < 0 ? ret : 0;
> }
>
> static inline int erofs_extract_dir(struct erofs_inode *inode)
> @@ -1043,10 +1007,21 @@ int erofsfsck_fuzz_one(int argc, char *argv[])
> int main(int argc, char *argv[])
> #endif
> {
> +
> int err;
> +#ifdef EROFS_MT_ENABLED
> + int workers;
> +#endif
>
> erofs_init_configure();
>
> +#ifdef EROFS_MT_ENABLED
> + workers = erofs_get_available_processors();
> + if (workers < 1)
> + workers = 1;
> + erofs_alloc_workqueue(&erofs_wq, workers, 256, NULL, NULL);
> +#endif
It shouldn't be worked as this, we should have a helper
in liberofs to wrap up the initialization.
And erofs_wq naming is ambigious.
> +
> fsckcfg.physical_blocks = 0;
> fsckcfg.logical_blocks = 0;
> fsckcfg.extract_path = NULL;
> @@ -1181,6 +1156,9 @@ exit_dev_close:
> exit:
> erofs_blob_closeall(&g_sbi);
> erofs_exit_configure();
> +#ifdef EROFS_MT_ENABLED
> + erofs_destroy_workqueue(&erofs_wq);
> +#endif
Same here.
> return err ? 1 : 0;
> }
>
> diff --git a/include/erofs/cond.h b/include/erofs/cond.h
> new file mode 100644
> index 0000000..90ec838
> --- /dev/null
> +++ b/include/erofs/cond.h
> @@ -0,0 +1,31 @@
> +/* SPDX-License-Identifier: GPL-2.0+ OR Apache-2.0 */
> +#ifndef __EROFS_COND_H
> +#define __EROFS_COND_H
> +
> +#include "lock.h"
> +
> +#if defined(HAVE_PTHREAD_H) && defined(EROFS_MT_ENABLED)
> +#include <pthread.h>
> +
> +typedef pthread_cond_t erofs_cond_t;
> +
> +static inline void erofs_cond_init(erofs_cond_t *cond)
> +{
> + pthread_cond_init(cond, NULL);
> +}
> +#define erofs_cond_wait pthread_cond_wait
> +#define erofs_cond_signal pthread_cond_signal
> +#define erofs_cond_broadcast pthread_cond_broadcast
> +#define erofs_cond_destroy pthread_cond_destroy
> +
> +#else
> +typedef struct {} erofs_cond_t;
> +
> +static inline void erofs_cond_init(erofs_cond_t *cond) {}
> +static inline int erofs_cond_wait(erofs_cond_t *cond, erofs_mutex_t *mutex) { return 0; }
> +static inline int erofs_cond_signal(erofs_cond_t *cond) { return 0; }
> +static inline int erofs_cond_broadcast(erofs_cond_t *cond) { return 0; }
> +static inline int erofs_cond_destroy(erofs_cond_t *cond) { return 0; }
> +#endif
> +
> +#endif
> \ No newline at end of file
> diff --git a/include/erofs/internal.h b/include/erofs/internal.h
> index 671880f..94f14da 100644
> --- a/include/erofs/internal.h
> +++ b/include/erofs/internal.h
> @@ -25,6 +25,8 @@ typedef unsigned short umode_t;
> #ifdef HAVE_PTHREAD_H
> #include <pthread.h>
> #endif
> +#include <erofs/lock.h>
> +#include "erofs/cond.h"
> #include <stdlib.h>
> #include <string.h>
> #include "atomic.h"
> @@ -62,6 +64,7 @@ struct erofs_buf {
> #define erofs_pos(sbi, nr) ((erofs_off_t)(nr) << (sbi)->blkszbits)
> #define BLK_ROUND_UP(sbi, addr) \
> (roundup(addr, erofs_blksiz(sbi)) >> (sbi)->blkszbits)
> +#define Z_EROFS_PCLUSTER_MAX_BATCH_SIZE 32
>
> struct erofs_buffer_head;
> struct erofs_bufmgr;
> @@ -442,6 +445,20 @@ struct z_erofs_paramset {
> char *extraopts;
> };
>
> +struct z_erofs_decompress_task;
> +
> +struct z_erofs_read_ctx {
> + erofs_mutex_t lock;
> + erofs_cond_t cond;
> + int pending_tasks;
> + int final_err;
> + int outfd;
> + bool free_out;
> + struct z_erofs_decompress_task *current_task;
> +};
Is it necessary to expose the internal structure fields
to users?
> +
> +void z_erofs_read_ctx_enqueue(struct z_erofs_read_ctx *ctx);
z_erofs_mt_read_enqueue
> +
> int liberofs_global_init(void);
> void liberofs_global_exit(void);
>
> @@ -478,7 +495,8 @@ int erofs_read_one_data(struct erofs_inode *inode, struct erofs_map_blocks *map,
> char *buffer, u64 offset, size_t len);
> int z_erofs_read_one_data(struct erofs_inode *inode,
> struct erofs_map_blocks *map, char *raw, char *buffer,
> - erofs_off_t skip, erofs_off_t length, bool trimmed);
> + erofs_off_t skip, erofs_off_t length, bool trimmed,
> + erofs_off_t out_offset, struct z_erofs_read_ctx *ctx);
> void *erofs_read_metadata(struct erofs_sb_info *sbi, erofs_nid_t nid,
> erofs_off_t *offset, int *lengthp);
> int z_erofs_parse_cfgs(struct erofs_sb_info *sbi, struct erofs_super_block *dsb);
> diff --git a/include/erofs/lock.h b/include/erofs/lock.h
> index c6e3093..2e79d52 100644
> --- a/include/erofs/lock.h
> +++ b/include/erofs/lock.h
> @@ -15,6 +15,7 @@ static inline void erofs_mutex_init(erofs_mutex_t *lock)
> }
> #define erofs_mutex_lock pthread_mutex_lock
> #define erofs_mutex_unlock pthread_mutex_unlock
> +#define erofs_mutex_destroy pthread_mutex_destroy
>
> #define EROFS_DEFINE_MUTEX(lock) \
> erofs_mutex_t lock = PTHREAD_MUTEX_INITIALIZER
> @@ -29,12 +30,14 @@ static inline void erofs_init_rwsem(erofs_rwsem_t *lock)
> #define erofs_down_write pthread_rwlock_wrlock
> #define erofs_up_read pthread_rwlock_unlock
> #define erofs_up_write pthread_rwlock_unlock
> +
> #else
> typedef struct {} erofs_mutex_t;
>
> static inline void erofs_mutex_init(erofs_mutex_t *lock) {}
> static inline void erofs_mutex_lock(erofs_mutex_t *lock) {}
> static inline void erofs_mutex_unlock(erofs_mutex_t *lock) {}
> +static inline void erofs_mutex_destroy(erofs_mutex_t *lock) {}
>
> #define EROFS_DEFINE_MUTEX(lock) \
> erofs_mutex_t lock = {}
> diff --git a/lib/data.c b/lib/data.c
> index 6fd1389..e9d2218 100644
> --- a/lib/data.c
> +++ b/lib/data.c
> @@ -9,6 +9,73 @@
> #include "erofs/trace.h"
> #include "erofs/decompress.h"
> #include "liberofs_fragments.h"
> +#include "erofs/workqueue.h"
> +#include "erofs/lock.h"
> +
> +struct erofs_workqueue erofs_wq;
> +
> +struct z_erofs_decompress_item {
> + struct z_erofs_decompress_req req;
> + char *raw_buf;
> + char *out_buf;
> + erofs_off_t out_offset;
> + unsigned int out_length;
> +};
> +
> +struct z_erofs_decompress_task {
> + struct erofs_work work;
> + struct z_erofs_read_ctx *ctx;
> + struct z_erofs_decompress_item items[Z_EROFS_PCLUSTER_MAX_BATCH_SIZE];
> + unsigned int nr_reqs;
> +};
> +
> +static void z_erofs_decompress_worker(struct erofs_work *work, void *tlsp)
> +{
> + struct z_erofs_decompress_task *task = (struct z_erofs_decompress_task *)work;
> + struct z_erofs_read_ctx *ctx = task->ctx;
> + int i, ret = 0, first_err = 0;
> +
> + for (i = 0; i < task->nr_reqs; ++i) {
> + struct z_erofs_decompress_item *item = &task->items[i];
> + ret = z_erofs_decompress(&item->req);
> +
> + if (ret >= 0 && ctx && ctx->outfd >= 0) {
> + if (pwrite(ctx->outfd, item->out_buf,
> + item->out_length, item->out_offset) < 0)
> + ret = -errno;
> + }
> +
> + if (ret < 0 && first_err == 0)
We don't use first_err == 0 style, it should be !first_err.
> + first_err = ret;
> +
> + free(item->raw_buf);
> + if (ctx && ctx->free_out)
> + free(item->out_buf);
> + }
> +
> + if (ctx) {
> + erofs_mutex_lock(&ctx->lock);
> + if (first_err < 0 && ctx->final_err == 0)
> + ctx->final_err = first_err;
> + ctx->pending_tasks--;
> + if (ctx->pending_tasks == 0)
Same here.
> + erofs_cond_signal(&ctx->cond);
> + erofs_mutex_unlock(&ctx->lock);
> + }
> + free(task);
> +}
> +
> +void z_erofs_read_ctx_enqueue(struct z_erofs_read_ctx *ctx)
> +{
> + if (ctx && ctx->current_task) {
> +#ifdef EROFS_MT_ENABLED
> + erofs_queue_work(&erofs_wq, &ctx->current_task->work);
> +#else
> + z_erofs_decompress_worker(&ctx->current_task->work, NULL);
> +#endif
> + ctx->current_task = NULL;
> + }
> +}
>
> void *erofs_bread(struct erofs_buf *buf, erofs_off_t offset, bool need_kmap)
> {
> @@ -277,7 +344,8 @@ static int erofs_read_raw_data(struct erofs_inode *inode, char *buffer,
>
> int z_erofs_read_one_data(struct erofs_inode *inode,
> struct erofs_map_blocks *map, char *raw, char *buffer,
> - erofs_off_t skip, erofs_off_t length, bool trimmed)
> + erofs_off_t skip, erofs_off_t length, bool trimmed,
> + erofs_off_t out_offset, struct z_erofs_read_ctx *ctx)
> {
> struct erofs_sb_info *sbi = inode->sbi;
> struct erofs_map_dev mdev;
> @@ -285,77 +353,107 @@ int z_erofs_read_one_data(struct erofs_inode *inode,
>
> if (map->m_flags & __EROFS_MAP_FRAGMENT) {
> if (__erofs_unlikely(inode->nid == sbi->packed_nid)) {
> - erofs_err("fragment should not exist in the packed inode %llu",
> - sbi->packed_nid | 0ULL);
> - return -EFSCORRUPTED;
> + ret = -EFSCORRUPTED;
> + goto err_out;
> + }
> + ret = erofs_packedfile_read(sbi, buffer, length - skip,
> + inode->fragmentoff + skip);
> +
> + if (ret >= 0 && ctx && ctx->outfd >= 0) {
> + if (pwrite(ctx->outfd, buffer, length - skip, out_offset) < 0)
> + ret = -errno;
> }
> - return erofs_packedfile_read(sbi, buffer, length - skip,
> - inode->fragmentoff + skip);
> + goto err_out;
> }
>
> - /* no device id here, thus it will always succeed */
> - mdev = (struct erofs_map_dev) {
> - .m_pa = map->m_pa,
> - };
> + mdev = (struct erofs_map_dev) { .m_pa = map->m_pa };
Please don't change unrelated code.
> ret = erofs_map_dev(sbi, &mdev);
> - if (ret) {
> - DBG_BUGON(1);
> - return ret;
> - }
> + if (ret)
> + goto err_out;
>
> ret = erofs_dev_read(sbi, mdev.m_deviceid, raw, mdev.m_pa, map->m_plen);
> if (ret < 0)
> - return ret;
> + goto err_out;
> + struct z_erofs_decompress_task *task = ctx->current_task;
> + if (!task) {
> + task = calloc(1, sizeof(*task));
> + task->ctx = ctx;
> + task->work.fn = z_erofs_decompress_worker;
> + ctx->current_task = task;
> +
> + erofs_mutex_lock(&ctx->lock);
> + ctx->pending_tasks++;
> + erofs_mutex_unlock(&ctx->lock);
> + }
>
> - ret = z_erofs_decompress(&(struct z_erofs_decompress_req) {
> - .sbi = sbi,
> - .in = raw,
> - .out = buffer,
> - .decodedskip = skip,
> - .interlaced_offset =
> - map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
> - erofs_blkoff(sbi, map->m_la) : 0,
> - .inputsize = map->m_plen,
> - .decodedlength = length,
> - .alg = map->m_algorithmformat,
> - .partial_decoding = trimmed ? true :
> - !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
> - (map->m_flags & EROFS_MAP_PARTIAL_REF),
> - });
> - if (ret < 0)
> - return ret;
> + int idx = task->nr_reqs++;
> + struct z_erofs_decompress_item *item = &task->items[idx];
Definitions should be at the beginning of the code blocks.
> +
> + item->req = (struct z_erofs_decompress_req) {
> + .sbi = sbi,
> + .in = raw,
> + .out = buffer,
> + .decodedskip = skip,
> + .interlaced_offset =
> + map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
> + erofs_blkoff(sbi, map->m_la) : 0,
> + .inputsize = map->m_plen,
> + .decodedlength = length,
> + .alg = map->m_algorithmformat,
> + .partial_decoding = trimmed ? true :
> + !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
> + (map->m_flags & EROFS_MAP_PARTIAL_REF),
> + };
> + item->raw_buf = raw;
> + item->out_buf = buffer;
> + item->out_offset = out_offset;
> + item->out_length = length;
> +
> + int batch_limit = (map->m_algorithmformat == Z_EROFS_COMPRESSION_LZ4) ?
> + Z_EROFS_PCLUSTER_MAX_BATCH_SIZE : 8;
Same here.
> +
> + if (task->nr_reqs >= batch_limit) {
> + z_erofs_read_ctx_enqueue(ctx);
> + }
> return 0;
> +
> +err_out:
> + if (ctx && ctx->free_out) free(buffer);
Wrong style, please fix them all.
> + free(raw);
> + return ret;
> }
>
> static int z_erofs_read_data(struct erofs_inode *inode, char *buffer,
> - erofs_off_t size, erofs_off_t offset)
> + erofs_off_t size, erofs_off_t offset)
> {
> erofs_off_t end, length, skip;
> struct erofs_map_blocks map = {
> .buf = __EROFS_BUF_INITIALIZER,
> };
> bool trimmed;
> - unsigned int bufsize = 0;
> - char *raw = NULL;
> int ret = 0;
>
> + struct z_erofs_read_ctx ctx = {
> + .pending_tasks = 0,
> + .final_err = 0,
> + .outfd = -1,
> + .free_out = false,
> + .current_task = NULL
> + };
> + erofs_mutex_init(&ctx.lock);
> + erofs_cond_init(&ctx.cond);
> +
> end = offset + size;
> while (end > offset) {
> map.m_la = end - 1;
>
> ret = z_erofs_map_blocks_iter(inode, &map, 0);
> - if (ret)
> - break;
> + if (ret) break;
Wrong style again.
Thanks,
Gao Xiang
^ permalink raw reply [flat|nested] 13+ messages in thread
* [PATCH 1/2 v2] fsck.erofs: add multi-threaded decompression
2026-06-21 12:01 ` [PATCH 1/2] fsck.erofs: add multi-threaded decompression Nithurshen
2026-06-22 2:08 ` Gao Xiang
@ 2026-06-29 9:55 ` Nithurshen
2026-07-05 14:00 ` Gao Xiang
2026-07-06 6:05 ` [PATCH 1/2 v3] " Nithurshen
2 siblings, 1 reply; 13+ messages in thread
From: Nithurshen @ 2026-06-29 9:55 UTC (permalink / raw)
To: nithurshen.dev; +Cc: hsiangkao, linux-erofs, xiang
Currently, fsck.erofs extracts files synchronously. When decompressing
heavily packed images, the main thread spends the majority of its time
blocked on a combination of synchronous vfs_write() syscalls and
decompression routines, bottlenecking overall extraction speed.
This patch introduces a scalable, multi-threaded decompression framework
using the existing erofs_workqueue infrastructure to decouple compute
from the main thread's I/O.
To prevent massive scheduling overhead (futex contention) where worker
threads spend more CPU time waking up than actually decompressing small
clusters, this implementation introduces a batching context. Because
different compression algorithms exhibit vastly different scheduling
thresholds, the batch size is algorithm-aware:
- Fast algorithms like LZ4 utilize a larger batch limit (up to 32
pclusters) to effectively hide synchronization overhead.
- Compute-heavy algorithms like LZMA or ZSTD trigger at a lower
threshold (8 pclusters) to prevent memory bloat and thread starvation.
Key details of this implementation:
- The worker pool is dynamically sized based on available system CPUs.
- Decompression tasks take strict ownership of the raw and output
buffers (safely tracking memory via a `free_out` flag) to prevent
data races and memory leaks.
- Output buffers are explicitly zero-initialized via calloc() to
prevent trailing garbage bytes from leaking into extracted files.
- Tail-end packed fragments are processed synchronously by the main
thread, as their minimal overhead does not benefit from asynchronous
offloading.
Signed-off-by: Nithurshen <nithurshen.dev@gmail.com>
---
fsck/main.c | 122 +++++++------------
include/erofs/cond.h | 31 +++++
include/erofs/internal.h | 15 ++-
include/erofs/lock.h | 3 +
lib/data.c | 254 +++++++++++++++++++++++++++++++--------
lib/global.c | 15 +++
6 files changed, 307 insertions(+), 133 deletions(-)
create mode 100644 include/erofs/cond.h
diff --git a/fsck/main.c b/fsck/main.c
index 16cc627..c427a70 100644
--- a/fsck/main.c
+++ b/fsck/main.c
@@ -8,11 +8,13 @@
#include <time.h>
#include <utime.h>
#include <unistd.h>
+#include "erofs/lock.h"
#include <sys/stat.h>
#include "erofs/print.h"
#include "erofs/decompress.h"
#include "erofs/dir.h"
#include "erofs/xattr.h"
+#include "erofs/workqueue.h"
#include "../lib/compressor.h"
#include "../lib/liberofs_compress.h"
@@ -509,40 +511,31 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd)
.buf = __EROFS_BUF_INITIALIZER,
};
bool needdecode = fsckcfg.check_decomp && !erofs_is_packed_inode(inode);
- int ret = 0;
- bool compressed;
+ int ret = 0, wait_err;
+ bool compressed = erofs_inode_is_data_compressed(inode->datalayout);
erofs_off_t pos = 0;
u64 pchunk_len = 0;
- unsigned int raw_size = 0, buffer_size = 0;
- char *raw = NULL, *buffer = NULL;
+ struct z_erofs_mt_read_ctx *ctx;
+
+ ctx = z_erofs_mt_read_ctx_alloc(outfd, true);
+ if (!ctx)
+ return -ENOMEM;
erofs_dbg("verify data chunk of nid(%llu): type(%d)",
- inode->nid | 0ULL, inode->datalayout);
+ inode->nid | 0ULL, inode->datalayout);
- compressed = erofs_inode_is_data_compressed(inode->datalayout);
while (pos < inode->i_size) {
- unsigned int alloc_rawsize;
-
map.m_la = pos;
ret = erofs_map_blocks(inode, &map, EROFS_GET_BLOCKS_FIEMAP);
if (ret)
goto out;
- if (!compressed && map.m_llen != map.m_plen) {
- erofs_err("broken chunk length m_la %" PRIu64 " m_llen %" PRIu64 " m_plen %" PRIu64,
- map.m_la, map.m_llen, map.m_plen);
- ret = -EFSCORRUPTED;
- goto out;
- }
-
- /* the last lcluster can be divided into 3 parts */
if (map.m_la + map.m_llen > inode->i_size)
map.m_llen = inode->i_size - map.m_la;
pchunk_len += map.m_plen;
pos += map.m_llen;
- /* should skip decomp? */
if (map.m_la >= inode->i_size || !needdecode)
continue;
@@ -555,85 +548,53 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd)
continue;
}
- if (map.m_plen > Z_EROFS_PCLUSTER_MAX_SIZE) {
- if (compressed && !(map.m_flags & __EROFS_MAP_FRAGMENT)) {
- erofs_err("invalid pcluster size %" PRIu64 " @ offset %" PRIu64 " of nid %" PRIu64,
- map.m_plen, map.m_la,
- inode->nid | 0ULL);
- ret = -EFSCORRUPTED;
- goto out;
- }
- alloc_rawsize = Z_EROFS_PCLUSTER_MAX_SIZE;
- } else {
- alloc_rawsize = map.m_plen;
- }
-
- if (alloc_rawsize > raw_size) {
- char *newraw = realloc(raw, alloc_rawsize);
-
- if (!newraw) {
+ if (compressed) {
+ size_t buffer_size = map.m_llen > erofs_blksiz(inode->sbi) ?
+ map.m_llen : erofs_blksiz(inode->sbi);
+ char *raw, *buffer;
+ raw = malloc(map.m_plen);
+ buffer = calloc(1, buffer_size);
+
+ if (!raw || !buffer) {
+ free(raw);
+ free(buffer);
ret = -ENOMEM;
goto out;
}
- raw = newraw;
- raw_size = alloc_rawsize;
- }
- if (compressed) {
- if (map.m_llen > buffer_size) {
- char *newbuffer;
-
- buffer_size = map.m_llen;
- newbuffer = realloc(buffer, buffer_size);
- if (!newbuffer) {
- ret = -ENOMEM;
- goto out;
- }
- buffer = newbuffer;
+ ret = z_erofs_read_one_data(inode, &map, raw, buffer, 0, map.m_llen,
+ false, map.m_la, ctx);
+ if (ret) {
+ goto out;
}
- ret = z_erofs_read_one_data(inode, &map, raw, buffer,
- 0, map.m_llen, false);
+ } else {
+ char *raw = calloc(1, map.m_llen);
+ ret = erofs_read_one_data(inode, &map, raw, 0, map.m_llen);
+ if (ret >= 0 && outfd >= 0)
+ pwrite(outfd, raw, map.m_llen, map.m_la);
+ free(raw);
if (ret)
goto out;
-
- if (outfd >= 0 && write(outfd, buffer, map.m_llen) < 0)
- goto fail_eio;
- } else {
- u64 p = 0;
-
- do {
- u64 count = min_t(u64, alloc_rawsize,
- map.m_llen);
-
- ret = erofs_read_one_data(inode, &map, raw, p, count);
- if (ret)
- goto out;
-
- if (outfd >= 0 && write(outfd, raw, count) < 0)
- goto fail_eio;
- map.m_llen -= count;
- p += count;
- } while (map.m_llen);
}
}
+ z_erofs_mt_read_enqueue(ctx);
+
+out:
+ wait_err = z_erofs_mt_read_ctx_wait(ctx);
+ if (wait_err < 0 && ret >= 0)
+ ret = wait_err;
if (fsckcfg.print_comp_ratio) {
if (!erofs_is_packed_inode(inode))
fsckcfg.logical_blocks += BLK_ROUND_UP(inode->sbi, inode->i_size);
fsckcfg.physical_blocks += BLK_ROUND_UP(inode->sbi, pchunk_len);
}
-out:
- if (raw)
- free(raw);
- if (buffer)
- free(buffer);
- return ret < 0 ? ret : 0;
-fail_eio:
- erofs_err("I/O error occurred when verifying data chunk @ nid %llu",
- inode->nid | 0ULL);
- ret = -EIO;
- goto out;
+ if (outfd >= 0 && ret >= 0)
+ ftruncate(outfd, inode->i_size);
+
+ z_erofs_mt_read_ctx_free(ctx);
+ return ret < 0 ? ret : 0;
}
static inline int erofs_extract_dir(struct erofs_inode *inode)
@@ -1043,6 +1004,7 @@ int erofsfsck_fuzz_one(int argc, char *argv[])
int main(int argc, char *argv[])
#endif
{
+
int err;
erofs_init_configure();
diff --git a/include/erofs/cond.h b/include/erofs/cond.h
new file mode 100644
index 0000000..90ec838
--- /dev/null
+++ b/include/erofs/cond.h
@@ -0,0 +1,31 @@
+/* SPDX-License-Identifier: GPL-2.0+ OR Apache-2.0 */
+#ifndef __EROFS_COND_H
+#define __EROFS_COND_H
+
+#include "lock.h"
+
+#if defined(HAVE_PTHREAD_H) && defined(EROFS_MT_ENABLED)
+#include <pthread.h>
+
+typedef pthread_cond_t erofs_cond_t;
+
+static inline void erofs_cond_init(erofs_cond_t *cond)
+{
+ pthread_cond_init(cond, NULL);
+}
+#define erofs_cond_wait pthread_cond_wait
+#define erofs_cond_signal pthread_cond_signal
+#define erofs_cond_broadcast pthread_cond_broadcast
+#define erofs_cond_destroy pthread_cond_destroy
+
+#else
+typedef struct {} erofs_cond_t;
+
+static inline void erofs_cond_init(erofs_cond_t *cond) {}
+static inline int erofs_cond_wait(erofs_cond_t *cond, erofs_mutex_t *mutex) { return 0; }
+static inline int erofs_cond_signal(erofs_cond_t *cond) { return 0; }
+static inline int erofs_cond_broadcast(erofs_cond_t *cond) { return 0; }
+static inline int erofs_cond_destroy(erofs_cond_t *cond) { return 0; }
+#endif
+
+#endif
\ No newline at end of file
diff --git a/include/erofs/internal.h b/include/erofs/internal.h
index 671880f..a3ddcfb 100644
--- a/include/erofs/internal.h
+++ b/include/erofs/internal.h
@@ -25,6 +25,8 @@ typedef unsigned short umode_t;
#ifdef HAVE_PTHREAD_H
#include <pthread.h>
#endif
+#include <erofs/lock.h>
+#include "erofs/cond.h"
#include <stdlib.h>
#include <string.h>
#include "atomic.h"
@@ -62,6 +64,7 @@ struct erofs_buf {
#define erofs_pos(sbi, nr) ((erofs_off_t)(nr) << (sbi)->blkszbits)
#define BLK_ROUND_UP(sbi, addr) \
(roundup(addr, erofs_blksiz(sbi)) >> (sbi)->blkszbits)
+#define Z_EROFS_PCLUSTER_MAX_BATCH_SIZE 32
struct erofs_buffer_head;
struct erofs_bufmgr;
@@ -442,6 +445,15 @@ struct z_erofs_paramset {
char *extraopts;
};
+struct z_erofs_decompress_task;
+
+struct z_erofs_mt_read_ctx;
+
+struct z_erofs_mt_read_ctx *z_erofs_mt_read_ctx_alloc(int outfd, bool free_out);
+int z_erofs_mt_read_ctx_wait(struct z_erofs_mt_read_ctx *ctx);
+void z_erofs_mt_read_ctx_free(struct z_erofs_mt_read_ctx *ctx);
+void z_erofs_mt_read_enqueue(struct z_erofs_mt_read_ctx *ctx);
+
int liberofs_global_init(void);
void liberofs_global_exit(void);
@@ -478,7 +490,8 @@ int erofs_read_one_data(struct erofs_inode *inode, struct erofs_map_blocks *map,
char *buffer, u64 offset, size_t len);
int z_erofs_read_one_data(struct erofs_inode *inode,
struct erofs_map_blocks *map, char *raw, char *buffer,
- erofs_off_t skip, erofs_off_t length, bool trimmed);
+ erofs_off_t skip, erofs_off_t length, bool trimmed,
+ erofs_off_t out_offset, struct z_erofs_mt_read_ctx *ctx);
void *erofs_read_metadata(struct erofs_sb_info *sbi, erofs_nid_t nid,
erofs_off_t *offset, int *lengthp);
int z_erofs_parse_cfgs(struct erofs_sb_info *sbi, struct erofs_super_block *dsb);
diff --git a/include/erofs/lock.h b/include/erofs/lock.h
index c6e3093..2e79d52 100644
--- a/include/erofs/lock.h
+++ b/include/erofs/lock.h
@@ -15,6 +15,7 @@ static inline void erofs_mutex_init(erofs_mutex_t *lock)
}
#define erofs_mutex_lock pthread_mutex_lock
#define erofs_mutex_unlock pthread_mutex_unlock
+#define erofs_mutex_destroy pthread_mutex_destroy
#define EROFS_DEFINE_MUTEX(lock) \
erofs_mutex_t lock = PTHREAD_MUTEX_INITIALIZER
@@ -29,12 +30,14 @@ static inline void erofs_init_rwsem(erofs_rwsem_t *lock)
#define erofs_down_write pthread_rwlock_wrlock
#define erofs_up_read pthread_rwlock_unlock
#define erofs_up_write pthread_rwlock_unlock
+
#else
typedef struct {} erofs_mutex_t;
static inline void erofs_mutex_init(erofs_mutex_t *lock) {}
static inline void erofs_mutex_lock(erofs_mutex_t *lock) {}
static inline void erofs_mutex_unlock(erofs_mutex_t *lock) {}
+static inline void erofs_mutex_destroy(erofs_mutex_t *lock) {}
#define EROFS_DEFINE_MUTEX(lock) \
erofs_mutex_t lock = {}
diff --git a/lib/data.c b/lib/data.c
index 6fd1389..fe0005c 100644
--- a/lib/data.c
+++ b/lib/data.c
@@ -9,6 +9,121 @@
#include "erofs/trace.h"
#include "erofs/decompress.h"
#include "liberofs_fragments.h"
+#include "erofs/workqueue.h"
+#include "erofs/lock.h"
+
+struct erofs_workqueue z_erofs_mt_wq;
+
+struct z_erofs_mt_read_ctx {
+ erofs_mutex_t lock;
+ erofs_cond_t cond;
+ int pending_tasks;
+ int final_err;
+ int outfd;
+ bool free_out;
+ struct z_erofs_decompress_task *current_task;
+};
+
+struct z_erofs_mt_read_ctx *z_erofs_mt_read_ctx_alloc(int outfd, bool free_out)
+{
+ struct z_erofs_mt_read_ctx *ctx = calloc(1, sizeof(*ctx));
+ if (!ctx)
+ return NULL;
+
+ erofs_mutex_init(&ctx->lock);
+ erofs_cond_init(&ctx->cond);
+ ctx->outfd = outfd;
+ ctx->free_out = free_out;
+ return ctx;
+}
+
+int z_erofs_mt_read_ctx_wait(struct z_erofs_mt_read_ctx *ctx)
+{
+ int err;
+ if (!ctx)
+ return 0;
+
+ erofs_mutex_lock(&ctx->lock);
+ while (ctx->pending_tasks > 0)
+ erofs_cond_wait(&ctx->cond, &ctx->lock);
+ err = ctx->final_err;
+ erofs_mutex_unlock(&ctx->lock);
+
+ return err;
+}
+
+void z_erofs_mt_read_ctx_free(struct z_erofs_mt_read_ctx *ctx)
+{
+ if (!ctx)
+ return;
+
+ erofs_mutex_destroy(&ctx->lock);
+ erofs_cond_destroy(&ctx->cond);
+ free(ctx);
+}
+
+struct z_erofs_decompress_item {
+ struct z_erofs_decompress_req req;
+ char *raw_buf;
+ char *out_buf;
+ erofs_off_t out_offset;
+ unsigned int out_length;
+};
+
+struct z_erofs_decompress_task {
+ struct erofs_work work;
+ struct z_erofs_mt_read_ctx *ctx;
+ struct z_erofs_decompress_item items[Z_EROFS_PCLUSTER_MAX_BATCH_SIZE];
+ unsigned int nr_reqs;
+};
+
+static void z_erofs_decompress_worker(struct erofs_work *work, void *tlsp)
+{
+ struct z_erofs_decompress_task *task = (struct z_erofs_decompress_task *)work;
+ struct z_erofs_mt_read_ctx *ctx = task->ctx;
+ int i, ret = 0, first_err = 0;
+
+ for (i = 0; i < task->nr_reqs; ++i) {
+ struct z_erofs_decompress_item *item = &task->items[i];
+ ret = z_erofs_decompress(&item->req);
+
+ if (ret >= 0 && ctx && ctx->outfd >= 0) {
+ if (pwrite(ctx->outfd, item->out_buf,
+ item->out_length, item->out_offset) < 0)
+ ret = -errno;
+ }
+
+ if (ret < 0 && !first_err)
+ first_err = ret;
+
+ free(item->raw_buf);
+ if (ctx && ctx->free_out)
+ free(item->out_buf);
+ }
+
+ if (ctx) {
+ erofs_mutex_lock(&ctx->lock);
+ if (first_err < 0 && !ctx->final_err)
+ ctx->final_err = first_err;
+ ctx->pending_tasks--;
+ if (!ctx->pending_tasks)
+ erofs_cond_signal(&ctx->cond);
+ erofs_mutex_unlock(&ctx->lock);
+ }
+ free(task);
+}
+
+void z_erofs_mt_read_enqueue(struct z_erofs_mt_read_ctx *ctx)
+{
+ if (ctx && ctx->current_task) {
+#ifdef EROFS_MT_ENABLED
+ erofs_queue_work(&z_erofs_mt_wq, &ctx->current_task->work);
+#else
+ z_erofs_decompress_worker(&ctx->current_task->work, NULL);
+#endif
+ ctx->current_task = NULL;
+ }
+}
void *erofs_bread(struct erofs_buf *buf, erofs_off_t offset, bool need_kmap)
{
@@ -277,20 +392,28 @@ static int erofs_read_raw_data(struct erofs_inode *inode, char *buffer,
int z_erofs_read_one_data(struct erofs_inode *inode,
struct erofs_map_blocks *map, char *raw, char *buffer,
- erofs_off_t skip, erofs_off_t length, bool trimmed)
+ erofs_off_t skip, erofs_off_t length, bool trimmed,
+ erofs_off_t out_offset, struct z_erofs_mt_read_ctx *ctx)
{
struct erofs_sb_info *sbi = inode->sbi;
struct erofs_map_dev mdev;
- int ret = 0;
+ struct z_erofs_decompress_task *task;
+ struct z_erofs_decompress_item *item;
+ int ret = 0, idx, batch_limit;
if (map->m_flags & __EROFS_MAP_FRAGMENT) {
if (__erofs_unlikely(inode->nid == sbi->packed_nid)) {
- erofs_err("fragment should not exist in the packed inode %llu",
- sbi->packed_nid | 0ULL);
- return -EFSCORRUPTED;
+ ret = -EFSCORRUPTED;
+ goto err_out;
+ }
+ ret = erofs_packedfile_read(sbi, buffer, length - skip,
+ inode->fragmentoff + skip);
+
+ if (ret >= 0 && ctx && ctx->outfd >= 0) {
+ if (pwrite(ctx->outfd, buffer, length - skip, out_offset) < 0)
+ ret = -errno;
}
- return erofs_packedfile_read(sbi, buffer, length - skip,
- inode->fragmentoff + skip);
+ goto err_out;
}
/* no device id here, thus it will always succeed */
@@ -298,46 +421,77 @@ int z_erofs_read_one_data(struct erofs_inode *inode,
.m_pa = map->m_pa,
};
ret = erofs_map_dev(sbi, &mdev);
- if (ret) {
- DBG_BUGON(1);
- return ret;
- }
+ if (ret)
+ goto err_out;
ret = erofs_dev_read(sbi, mdev.m_deviceid, raw, mdev.m_pa, map->m_plen);
if (ret < 0)
- return ret;
+ goto err_out;
+
+ task = ctx->current_task;
+ if (!task) {
+ task = calloc(1, sizeof(*task));
+ task->ctx = ctx;
+ task->work.fn = z_erofs_decompress_worker;
+ ctx->current_task = task;
+
+ erofs_mutex_lock(&ctx->lock);
+ ctx->pending_tasks++;
+ erofs_mutex_unlock(&ctx->lock);
+ }
- ret = z_erofs_decompress(&(struct z_erofs_decompress_req) {
- .sbi = sbi,
- .in = raw,
- .out = buffer,
- .decodedskip = skip,
- .interlaced_offset =
- map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
- erofs_blkoff(sbi, map->m_la) : 0,
- .inputsize = map->m_plen,
- .decodedlength = length,
- .alg = map->m_algorithmformat,
- .partial_decoding = trimmed ? true :
- !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
- (map->m_flags & EROFS_MAP_PARTIAL_REF),
- });
- if (ret < 0)
- return ret;
+ idx = task->nr_reqs++;
+ item = &task->items[idx];
+
+ item->req = (struct z_erofs_decompress_req) {
+ .sbi = sbi,
+ .in = raw,
+ .out = buffer,
+ .decodedskip = skip,
+ .interlaced_offset =
+ map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
+ erofs_blkoff(sbi, map->m_la) : 0,
+ .inputsize = map->m_plen,
+ .decodedlength = length,
+ .alg = map->m_algorithmformat,
+ .partial_decoding = trimmed ? true :
+ !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
+ (map->m_flags & EROFS_MAP_PARTIAL_REF),
+ };
+ item->raw_buf = raw;
+ item->out_buf = buffer;
+ item->out_offset = out_offset;
+ item->out_length = length;
+
+ batch_limit = (map->m_algorithmformat == Z_EROFS_COMPRESSION_LZ4) ?
+ Z_EROFS_PCLUSTER_MAX_BATCH_SIZE : 8;
+
+ if (task->nr_reqs >= batch_limit) {
+ z_erofs_mt_read_enqueue(ctx);
+ }
return 0;
+
+err_out:
+ if (ctx && ctx->free_out)
+ free(buffer);
+ free(raw);
+ return ret;
}
static int z_erofs_read_data(struct erofs_inode *inode, char *buffer,
- erofs_off_t size, erofs_off_t offset)
+ erofs_off_t size, erofs_off_t offset)
{
erofs_off_t end, length, skip;
struct erofs_map_blocks map = {
.buf = __EROFS_BUF_INITIALIZER,
};
bool trimmed;
- unsigned int bufsize = 0;
- char *raw = NULL;
- int ret = 0;
+ int ret = 0, wait_err;
+ struct z_erofs_mt_read_ctx *ctx;
+
+ ctx = z_erofs_mt_read_ctx_alloc(-1, false);
+ if (!ctx)
+ return -ENOMEM;
end = offset + size;
while (end > offset) {
@@ -347,15 +501,10 @@ static int z_erofs_read_data(struct erofs_inode *inode, char *buffer,
if (ret)
break;
- /*
- * trim to the needed size if the returned extent is quite
- * larger than requested, and set up partial flag as well.
- */
if (end < map.m_la + map.m_llen) {
length = end - map.m_la;
trimmed = true;
} else {
- DBG_BUGON(end != map.m_la + map.m_llen);
length = map.m_llen;
trimmed = false;
}
@@ -374,25 +523,26 @@ static int z_erofs_read_data(struct erofs_inode *inode, char *buffer,
continue;
}
- if (map.m_plen > bufsize) {
- char *newraw;
-
- bufsize = map.m_plen;
- newraw = realloc(raw, bufsize);
- if (!newraw) {
- ret = -ENOMEM;
- break;
- }
- raw = newraw;
+ char *raw = malloc(map.m_plen);
+ if (!raw) {
+ ret = -ENOMEM;
+ break;
}
ret = z_erofs_read_one_data(inode, &map, raw,
- buffer + end - offset, skip, length, trimmed);
- if (ret < 0)
+ buffer + end - offset, skip, length, trimmed, 0, ctx);
+ if (ret < 0) {
break;
+ }
}
- if (raw)
- free(raw);
+ z_erofs_mt_read_enqueue(ctx);
+
+ wait_err = z_erofs_mt_read_ctx_wait(ctx);
+ if (wait_err < 0 && !ret)
+ ret = wait_err;
+
+ z_erofs_mt_read_ctx_free(ctx);
+
return ret < 0 ? ret : 0;
}
diff --git a/lib/global.c b/lib/global.c
index c3d8aec..4ef424a 100644
--- a/lib/global.c
+++ b/lib/global.c
@@ -12,6 +12,11 @@
#include "erofs/err.h"
#include "erofs/config.h"
#include "liberofs_compress.h"
+#include "erofs/internal.h"
+#ifdef EROFS_MT_ENABLED
+#include "erofs/workqueue.h"
+extern struct erofs_workqueue z_erofs_mt_wq;
+#endif
static EROFS_DEFINE_MUTEX(erofs_global_mutex);
#ifdef HAVE_LIBCURL
@@ -22,6 +27,13 @@ int liberofs_global_init(void)
{
int err = 0;
+#ifdef EROFS_MT_ENABLED
+ int workers = erofs_get_available_processors();
+ if (workers < 1)
+ workers = 1;
+ erofs_alloc_workqueue(&z_erofs_mt_wq, workers, 256, NULL, NULL);
+#endif
+
erofs_mutex_lock(&erofs_global_mutex);
erofs_init_configure();
#ifdef S3EROFS_ENABLED
@@ -45,6 +57,9 @@ void liberofs_global_exit(void)
{
erofs_mutex_lock(&erofs_global_mutex);
z_erofs_mt_global_exit();
+#ifdef EROFS_MT_ENABLED
+ erofs_destroy_workqueue(&z_erofs_mt_wq);
+#endif
#ifdef HAVE_LIBCURL
if (erofs_global_curl_initialized) {
curl_global_cleanup();
--
2.52.0
^ permalink raw reply related [flat|nested] 13+ messages in thread
* Re: [PATCH 1/2 v2] fsck.erofs: add multi-threaded decompression
2026-06-29 9:55 ` [PATCH 1/2 v2] " Nithurshen
@ 2026-07-05 14:00 ` Gao Xiang
0 siblings, 0 replies; 13+ messages in thread
From: Gao Xiang @ 2026-07-05 14:00 UTC (permalink / raw)
To: Nithurshen; +Cc: hsiangkao, linux-erofs, xiang
On Mon, Jun 29, 2026 at 03:25:29PM +0530, Nithurshen wrote:
> Currently, fsck.erofs extracts files synchronously. When decompressing
> heavily packed images, the main thread spends the majority of its time
> blocked on a combination of synchronous vfs_write() syscalls and
> decompression routines, bottlenecking overall extraction speed.
>
> This patch introduces a scalable, multi-threaded decompression framework
> using the existing erofs_workqueue infrastructure to decouple compute
> from the main thread's I/O.
>
> To prevent massive scheduling overhead (futex contention) where worker
> threads spend more CPU time waking up than actually decompressing small
> clusters, this implementation introduces a batching context. Because
> different compression algorithms exhibit vastly different scheduling
> thresholds, the batch size is algorithm-aware:
> - Fast algorithms like LZ4 utilize a larger batch limit (up to 32
> pclusters) to effectively hide synchronization overhead.
> - Compute-heavy algorithms like LZMA or ZSTD trigger at a lower
> threshold (8 pclusters) to prevent memory bloat and thread starvation.
>
> Key details of this implementation:
> - The worker pool is dynamically sized based on available system CPUs.
> - Decompression tasks take strict ownership of the raw and output
> buffers (safely tracking memory via a `free_out` flag) to prevent
> data races and memory leaks.
> - Output buffers are explicitly zero-initialized via calloc() to
> prevent trailing garbage bytes from leaking into extracted files.
> - Tail-end packed fragments are processed synchronously by the main
> thread, as their minimal overhead does not benefit from asynchronous
> offloading.
>
> Signed-off-by: Nithurshen <nithurshen.dev@gmail.com>
> ---
> fsck/main.c | 122 +++++++------------
> include/erofs/cond.h | 31 +++++
> include/erofs/internal.h | 15 ++-
> include/erofs/lock.h | 3 +
> lib/data.c | 254 +++++++++++++++++++++++++++++++--------
> lib/global.c | 15 +++
> 6 files changed, 307 insertions(+), 133 deletions(-)
> create mode 100644 include/erofs/cond.h
>
> diff --git a/fsck/main.c b/fsck/main.c
> index 16cc627..c427a70 100644
> --- a/fsck/main.c
> +++ b/fsck/main.c
> @@ -8,11 +8,13 @@
> #include <time.h>
> #include <utime.h>
> #include <unistd.h>
> +#include "erofs/lock.h"
> #include <sys/stat.h>
> #include "erofs/print.h"
> #include "erofs/decompress.h"
> #include "erofs/dir.h"
> #include "erofs/xattr.h"
> +#include "erofs/workqueue.h"
> #include "../lib/compressor.h"
> #include "../lib/liberofs_compress.h"
>
> @@ -509,40 +511,31 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd)
> .buf = __EROFS_BUF_INITIALIZER,
> };
> bool needdecode = fsckcfg.check_decomp && !erofs_is_packed_inode(inode);
> - int ret = 0;
> - bool compressed;
> + int ret = 0, wait_err;
> + bool compressed = erofs_inode_is_data_compressed(inode->datalayout);
> erofs_off_t pos = 0;
> u64 pchunk_len = 0;
> - unsigned int raw_size = 0, buffer_size = 0;
> - char *raw = NULL, *buffer = NULL;
> + struct z_erofs_mt_read_ctx *ctx;
> +
> + ctx = z_erofs_mt_read_ctx_alloc(outfd, true);
> + if (!ctx)
> + return -ENOMEM;
>
> erofs_dbg("verify data chunk of nid(%llu): type(%d)",
> - inode->nid | 0ULL, inode->datalayout);
> + inode->nid | 0ULL, inode->datalayout);
I'm still not sure why this line needs to be changed.
>
> - compressed = erofs_inode_is_data_compressed(inode->datalayout);
> while (pos < inode->i_size) {
> - unsigned int alloc_rawsize;
> -
> map.m_la = pos;
> ret = erofs_map_blocks(inode, &map, EROFS_GET_BLOCKS_FIEMAP);
> if (ret)
> goto out;
>
> - if (!compressed && map.m_llen != map.m_plen) {
> - erofs_err("broken chunk length m_la %" PRIu64 " m_llen %" PRIu64 " m_plen %" PRIu64,
> - map.m_la, map.m_llen, map.m_plen);
> - ret = -EFSCORRUPTED;
> - goto out;
Where is the check now?
> - }
> -
> - /* the last lcluster can be divided into 3 parts */
> if (map.m_la + map.m_llen > inode->i_size)
> map.m_llen = inode->i_size - map.m_la;
>
> pchunk_len += map.m_plen;
> pos += map.m_llen;
>
> - /* should skip decomp? */
> if (map.m_la >= inode->i_size || !needdecode)
> continue;
>
> @@ -555,85 +548,53 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd)
> continue;
> }
>
> - if (map.m_plen > Z_EROFS_PCLUSTER_MAX_SIZE) {
> - if (compressed && !(map.m_flags & __EROFS_MAP_FRAGMENT)) {
> - erofs_err("invalid pcluster size %" PRIu64 " @ offset %" PRIu64 " of nid %" PRIu64,
> - map.m_plen, map.m_la,
> - inode->nid | 0ULL);
> - ret = -EFSCORRUPTED;
> - goto out;
> - }
Where is the check now?
> - alloc_rawsize = Z_EROFS_PCLUSTER_MAX_SIZE;
> - } else {
> - alloc_rawsize = map.m_plen;
> - }
> -
> - if (alloc_rawsize > raw_size) {
> - char *newraw = realloc(raw, alloc_rawsize);
> -
> - if (!newraw) {
> + if (compressed) {
> + size_t buffer_size = map.m_llen > erofs_blksiz(inode->sbi) ?
> + map.m_llen : erofs_blksiz(inode->sbi);
Can you explain why?
> + char *raw, *buffer;
> + raw = malloc(map.m_plen);
> + buffer = calloc(1, buffer_size);
> +
> + if (!raw || !buffer) {
> + free(raw);
> + free(buffer);
> ret = -ENOMEM;
> goto out;
> }
> - raw = newraw;
> - raw_size = alloc_rawsize;
> - }
>
> - if (compressed) {
> - if (map.m_llen > buffer_size) {
> - char *newbuffer;
> -
> - buffer_size = map.m_llen;
> - newbuffer = realloc(buffer, buffer_size);
> - if (!newbuffer) {
> - ret = -ENOMEM;
> - goto out;
> - }
> - buffer = newbuffer;
> + ret = z_erofs_read_one_data(inode, &map, raw, buffer, 0, map.m_llen,
> + false, map.m_la, ctx);
> + if (ret) {
> + goto out;
> }
> - ret = z_erofs_read_one_data(inode, &map, raw, buffer,
> - 0, map.m_llen, false);
> + } else {
> + char *raw = calloc(1, map.m_llen);
> + ret = erofs_read_one_data(inode, &map, raw, 0, map.m_llen);
> + if (ret >= 0 && outfd >= 0)
> + pwrite(outfd, raw, map.m_llen, map.m_la);
> + free(raw);
> if (ret)
> goto out;
> -
> - if (outfd >= 0 && write(outfd, buffer, map.m_llen) < 0)
> - goto fail_eio;
> - } else {
> - u64 p = 0;
> -
> - do {
> - u64 count = min_t(u64, alloc_rawsize,
> - map.m_llen);
> -
> - ret = erofs_read_one_data(inode, &map, raw, p, count);
> - if (ret)
> - goto out;
> -
> - if (outfd >= 0 && write(outfd, raw, count) < 0)
> - goto fail_eio;
> - map.m_llen -= count;
> - p += count;
> - } while (map.m_llen);
> }
> }
> + z_erofs_mt_read_enqueue(ctx);
> +
> +out:
> + wait_err = z_erofs_mt_read_ctx_wait(ctx);
> + if (wait_err < 0 && ret >= 0)
> + ret = wait_err;
>
> if (fsckcfg.print_comp_ratio) {
> if (!erofs_is_packed_inode(inode))
> fsckcfg.logical_blocks += BLK_ROUND_UP(inode->sbi, inode->i_size);
> fsckcfg.physical_blocks += BLK_ROUND_UP(inode->sbi, pchunk_len);
> }
> -out:
> - if (raw)
> - free(raw);
> - if (buffer)
> - free(buffer);
> - return ret < 0 ? ret : 0;
>
> -fail_eio:
> - erofs_err("I/O error occurred when verifying data chunk @ nid %llu",
> - inode->nid | 0ULL);
> - ret = -EIO;
> - goto out;
> + if (outfd >= 0 && ret >= 0)
> + ftruncate(outfd, inode->i_size);
Why ftruncate here?
In short, it seems that you remove a lot sanity checks, if you
touch the existing logic, please explain why it's safe to remove
them.
Also please show valid benchmark results (at least with enough
sizes, and the decompression should take many seconds), so I can
imagine it's a good improvement.
Thanks,
Gao Xiang
^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [PATCH 2/2] fsck.erofs: implement concurrent directory traversal
2026-06-21 12:01 ` [PATCH 2/2] fsck.erofs: implement concurrent directory traversal Nithurshen
@ 2026-07-05 14:05 ` Gao Xiang
2026-07-14 1:10 ` [PATCH v5] fsck.erofs: add multi-threaded decompression Nithurshen
1 sibling, 0 replies; 13+ messages in thread
From: Gao Xiang @ 2026-07-05 14:05 UTC (permalink / raw)
To: Nithurshen; +Cc: linux-erofs, hsiangkao, xiang
On Sun, Jun 21, 2026 at 05:31:21PM +0530, Nithurshen wrote:
> Currently, fsck.erofs traverses the filesystem tree and verifies
> inodes synchronously on the main thread. While decompression
> compute is offloaded, the main thread remains a bottleneck
> during the I/O-heavy directory walk.
>
> This patch parallelizes the directory traversal and inode
> extraction processes. To achieve this safely, the globally shared
> fsckcfg.extract_path and fsckcfg.dirstack states are decoupled
> and localized into individual struct erofsfsck_inode_task
> payloads, which are deep-copied and handed off to the worker
> pool. Global statistics and hardlink tables are secured using
> native erofs_mutex_t primitives.
>
> To prevent thread pool exhaustion deadlocks—where workers
> processing a deep directory tree occupy all available execution
> slots and block on erofs_cond_wait, starving their own spawned
> decompression tasks—this patch introduces a dedicated
> erofs_traverse_wq. By isolating the producers (traversal and
> verification) from the consumers (pcluster decompression), the
> pipeline avoids gridlock.
>
> Signed-off-by: Nithurshen <nithurshen.dev@gmail.com>
I really hope you could parallelize the inodes in a single directory
first, and get a minimal commit and show the improvement.
And try to improve parallelization between directories.
The priciple is to keep each patch small so that it's more reviewable
and has low risky.
Also it seems that you could make mutex protection in seperate commits
too so it's also easier to review.
Thanks,
Gao Xiang
^ permalink raw reply [flat|nested] 13+ messages in thread
* [PATCH 1/2 v3] fsck.erofs: add multi-threaded decompression
2026-06-21 12:01 ` [PATCH 1/2] fsck.erofs: add multi-threaded decompression Nithurshen
2026-06-22 2:08 ` Gao Xiang
2026-06-29 9:55 ` [PATCH 1/2 v2] " Nithurshen
@ 2026-07-06 6:05 ` Nithurshen
2026-07-06 6:10 ` Nithurshen
2026-07-13 7:02 ` [PATCH 1/2 v4] " Nithurshen
2 siblings, 2 replies; 13+ messages in thread
From: Nithurshen @ 2026-07-06 6:05 UTC (permalink / raw)
To: nithurshen.dev; +Cc: hsiangkao, linux-erofs, xiang
Currently, fsck.erofs extracts files synchronously. When decompressing
heavily packed images, the main thread spends the majority of its time
blocked on a combination of synchronous I/O syscalls and CPU-heavy
decompression routines, bottlenecking overall extraction speed.
This patch introduces a scalable, multi-threaded decompression framework
using the existing erofs_workqueue infrastructure to decouple compute
from the main thread's I/O.
To prevent massive scheduling overhead (futex contention) where worker
threads spend more CPU time waking up than actually decompressing small
clusters, this implementation introduces a batching context. Because
different compression algorithms exhibit vastly different scheduling
thresholds, the batch limit is algorithm-aware.
Signed-off-by: Nithurshen <nithurshen.dev@gmail.com>
---
fsck/main.c | 127 ++++++++++++++++-----
include/erofs/cond.h | 31 ++++++
include/erofs/internal.h | 17 ++-
include/erofs/lock.h | 2 +
lib/data.c | 230 ++++++++++++++++++++++++++++++++++-----
lib/global.c | 17 +++
6 files changed, 373 insertions(+), 51 deletions(-)
create mode 100644 include/erofs/cond.h
diff --git a/fsck/main.c b/fsck/main.c
index b2d8f1a..b74ae68 100644
--- a/fsck/main.c
+++ b/fsck/main.c
@@ -16,6 +16,7 @@
#include "../lib/compressor.h"
#include "../lib/liberofs_compress.h"
#include "../lib/liberofs_sha256.h"
+#include "erofs/internal.h"
static int erofsfsck_check_inode(erofs_nid_t pnid, erofs_nid_t nid);
@@ -526,6 +527,13 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
u64 pchunk_len = 0;
u64 raw_size = 0, buffer_size = 0;
char *raw = NULL, *buffer = NULL;
+ struct z_erofs_mt_read_ctx *ctx = NULL;
+
+ if (!digest) {
+ ctx = z_erofs_mt_read_ctx_alloc(outfd, true);
+ if (!ctx)
+ return -ENOMEM;
+ }
erofs_dbg("verify data chunk of nid(%llu): type(%d)",
inode->nid | 0ULL, inode->datalayout);
@@ -569,7 +577,7 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
(const u8 *)zeros, chunk);
remain -= chunk;
}
- } else if (outfd >= 0) {
+ } else if (!ctx && outfd >= 0) {
ret = lseek(outfd, map.m_llen, SEEK_CUR);
if (ret < 0) {
ret = -errno;
@@ -592,8 +600,9 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
alloc_rawsize = map.m_plen;
}
- if (alloc_rawsize > raw_size) {
- char *newraw = realloc(raw, alloc_rawsize);
+ if (!ctx) {
+ if (alloc_rawsize > raw_size) {
+ char *newraw = realloc(raw, alloc_rawsize);
if (!newraw) {
ret = -ENOMEM;
@@ -602,9 +611,22 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
raw = newraw;
raw_size = alloc_rawsize;
}
+ }
if (compressed) {
- if (map.m_llen > buffer_size) {
+ char *c_raw = raw;
+ char *c_buf = buffer;
+
+ if (ctx) {
+ c_raw = malloc(alloc_rawsize);
+ c_buf = calloc(1, map.m_llen);
+ if (!c_raw || !c_buf) {
+ free(c_raw);
+ free(c_buf);
+ ret = -ENOMEM;
+ goto out;
+ }
+ } else if (map.m_llen > buffer_size) {
char *newbuffer;
buffer_size = map.m_llen;
@@ -614,45 +636,98 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
goto out;
}
buffer = newbuffer;
+ c_buf = buffer;
}
- ret = z_erofs_read_one_data(inode, &map, raw, buffer,
- 0, map.m_llen, false);
- if (ret)
+
+ ret = z_erofs_read_one_data(inode, &map, c_raw, c_buf,
+ 0, map.m_llen, false,
+ map.m_la, ctx);
+ if (ret) {
+ if (ctx) {
+ free(c_raw);
+ free(c_buf);
+ }
goto out;
+ }
- if (digest)
- erofs_sha256_process(digest,
- (const u8 *)buffer, map.m_llen);
- if (outfd >= 0 && write(outfd, buffer, map.m_llen) < 0)
- goto fail_eio;
+ if (!ctx) {
+ if (digest)
+ erofs_sha256_process(digest,
+ (const u8 *)c_buf, map.m_llen);
+ if (outfd >= 0 && write(outfd, c_buf, map.m_llen) < 0)
+ goto fail_eio;
+ }
} else {
+ char *c_raw = raw;
u64 p = 0;
+ erofs_off_t m_llen = map.m_llen;
+
+ if (ctx) {
+ c_raw = calloc(1, alloc_rawsize);
+ if (!c_raw) {
+ ret = -ENOMEM;
+ goto out;
+ }
+ }
do {
- u64 count = min_t(u64, alloc_rawsize,
- map.m_llen);
+ u64 count = min_t(u64, alloc_rawsize, m_llen);
- ret = erofs_read_one_data(inode, &map, raw, p, count);
- if (ret)
+ ret = erofs_read_one_data(inode, &map, c_raw, p, count);
+ if (ret) {
+ if (ctx)
+ free(c_raw);
goto out;
+ }
- if (digest)
- erofs_sha256_process(digest,
- (const u8 *)raw, count);
- if (outfd >= 0 && write(outfd, raw, count) < 0)
- goto fail_eio;
- map.m_llen -= count;
+ if (!ctx) {
+ if (digest)
+ erofs_sha256_process(digest,
+ (const u8 *)c_raw, count);
+ if (outfd >= 0 && write(outfd, c_raw, count) < 0)
+ goto fail_eio;
+ } else if (outfd >= 0) {
+ if (pwrite(outfd, c_raw, count, map.m_la + p) < 0) {
+ free(c_raw);
+ goto fail_eio;
+ }
+ }
+ m_llen -= count;
p += count;
- } while (map.m_llen);
+ } while (m_llen);
+
+ if (ctx)
+ free(c_raw);
}
}
+ if (ctx) {
+ int wait_err;
+
+ z_erofs_mt_read_enqueue(ctx);
+ wait_err = z_erofs_mt_read_ctx_wait(ctx);
+ if (wait_err < 0 && ret == 0)
+ ret = wait_err;
+ }
+
if (fsckcfg.print_comp_ratio) {
if (!erofs_is_packed_inode(inode))
fsckcfg.logical_blocks += BLK_ROUND_UP(inode->sbi, inode->i_size);
fsckcfg.physical_blocks += BLK_ROUND_UP(inode->sbi, pchunk_len);
}
out:
+ if (ctx) {
+ /* use ftruncate to handle trailing unmapped blocks (holes) */
+ if (outfd >= 0 && ret == 0) {
+ if (ftruncate(outfd, inode->i_size) < 0) {
+ erofs_err("failed to truncate file to %llu: %d",
+ (unsigned long long)inode->i_size, errno);
+ if (ret == 0)
+ ret = -errno;
+ }
+ }
+ z_erofs_mt_read_ctx_free(ctx);
+ }
if (raw)
free(raw);
if (buffer)
@@ -1127,7 +1202,9 @@ int main(int argc, char *argv[])
{
int err;
- erofs_init_configure();
+ err = liberofs_global_init();
+ if (err)
+ return 1;
fsckcfg.physical_blocks = 0;
fsckcfg.logical_blocks = 0;
@@ -1283,7 +1360,7 @@ exit_dev_close:
erofs_dev_close(&g_sbi);
exit:
erofs_blob_closeall(&g_sbi);
- erofs_exit_configure();
+ liberofs_global_exit();
return err ? 1 : 0;
}
diff --git a/include/erofs/cond.h b/include/erofs/cond.h
new file mode 100644
index 0000000..90ec838
--- /dev/null
+++ b/include/erofs/cond.h
@@ -0,0 +1,31 @@
+/* SPDX-License-Identifier: GPL-2.0+ OR Apache-2.0 */
+#ifndef __EROFS_COND_H
+#define __EROFS_COND_H
+
+#include "lock.h"
+
+#if defined(HAVE_PTHREAD_H) && defined(EROFS_MT_ENABLED)
+#include <pthread.h>
+
+typedef pthread_cond_t erofs_cond_t;
+
+static inline void erofs_cond_init(erofs_cond_t *cond)
+{
+ pthread_cond_init(cond, NULL);
+}
+#define erofs_cond_wait pthread_cond_wait
+#define erofs_cond_signal pthread_cond_signal
+#define erofs_cond_broadcast pthread_cond_broadcast
+#define erofs_cond_destroy pthread_cond_destroy
+
+#else
+typedef struct {} erofs_cond_t;
+
+static inline void erofs_cond_init(erofs_cond_t *cond) {}
+static inline int erofs_cond_wait(erofs_cond_t *cond, erofs_mutex_t *mutex) { return 0; }
+static inline int erofs_cond_signal(erofs_cond_t *cond) { return 0; }
+static inline int erofs_cond_broadcast(erofs_cond_t *cond) { return 0; }
+static inline int erofs_cond_destroy(erofs_cond_t *cond) { return 0; }
+#endif
+
+#endif
\ No newline at end of file
diff --git a/include/erofs/internal.h b/include/erofs/internal.h
index 2cc9cc8..5e9d8c1 100644
--- a/include/erofs/internal.h
+++ b/include/erofs/internal.h
@@ -25,6 +25,11 @@ typedef unsigned short umode_t;
#ifdef HAVE_PTHREAD_H
#include <pthread.h>
#endif
+#ifdef HAVE_PTHREAD_H
+#include <pthread.h>
+#endif
+#include <erofs/lock.h>
+#include "erofs/cond.h"
#include <stdlib.h>
#include <string.h>
#include "atomic.h"
@@ -62,6 +67,7 @@ struct erofs_buf {
#define erofs_pos(sbi, nr) ((erofs_off_t)(nr) << (sbi)->blkszbits)
#define BLK_ROUND_UP(sbi, addr) \
(roundup(addr, erofs_blksiz(sbi)) >> (sbi)->blkszbits)
+#define Z_EROFS_PCLUSTER_MAX_BATCH_SIZE 32
struct erofs_buffer_head;
struct erofs_bufmgr;
@@ -451,6 +457,14 @@ struct z_erofs_paramset {
char *extraopts;
};
+struct z_erofs_decompress_task;
+struct z_erofs_mt_read_ctx;
+
+struct z_erofs_mt_read_ctx *z_erofs_mt_read_ctx_alloc(int outfd, bool free_out);
+int z_erofs_mt_read_ctx_wait(struct z_erofs_mt_read_ctx *ctx);
+void z_erofs_mt_read_ctx_free(struct z_erofs_mt_read_ctx *ctx);
+void z_erofs_mt_read_enqueue(struct z_erofs_mt_read_ctx *ctx);
+
int liberofs_global_init(void);
void liberofs_global_exit(void);
@@ -487,7 +501,8 @@ int erofs_read_one_data(struct erofs_inode *inode, struct erofs_map_blocks *map,
char *buffer, u64 offset, size_t len);
int z_erofs_read_one_data(struct erofs_inode *inode,
struct erofs_map_blocks *map, char *raw, char *buffer,
- erofs_off_t skip, erofs_off_t length, bool trimmed);
+ erofs_off_t skip, erofs_off_t length, bool trimmed,
+ erofs_off_t out_offset, struct z_erofs_mt_read_ctx *ctx);
void *erofs_read_metadata(struct erofs_sb_info *sbi, erofs_nid_t nid,
erofs_off_t *offset, int *lengthp);
int z_erofs_parse_cfgs(struct erofs_sb_info *sbi, struct erofs_super_block *dsb);
diff --git a/include/erofs/lock.h b/include/erofs/lock.h
index 884f23e..e08b76e 100644
--- a/include/erofs/lock.h
+++ b/include/erofs/lock.h
@@ -15,6 +15,7 @@ static inline void erofs_mutex_init(erofs_mutex_t *lock)
}
#define erofs_mutex_lock pthread_mutex_lock
#define erofs_mutex_unlock pthread_mutex_unlock
+#define erofs_mutex_destroy pthread_mutex_destroy
#define EROFS_DEFINE_MUTEX(lock) \
erofs_mutex_t lock = PTHREAD_MUTEX_INITIALIZER
@@ -35,6 +36,7 @@ typedef struct {} erofs_mutex_t;
static inline void erofs_mutex_init(erofs_mutex_t *lock) {}
static inline void erofs_mutex_lock(erofs_mutex_t *lock) {}
static inline void erofs_mutex_unlock(erofs_mutex_t *lock) {}
+static inline void erofs_mutex_destroy(erofs_mutex_t *lock) {}
#define EROFS_DEFINE_MUTEX(lock) \
erofs_mutex_t lock = {}
diff --git a/lib/data.c b/lib/data.c
index 1bb9269..edb262d 100644
--- a/lib/data.c
+++ b/lib/data.c
@@ -9,6 +9,123 @@
#include "erofs/trace.h"
#include "erofs/decompress.h"
#include "liberofs_fragments.h"
+#include "erofs/lock.h"
+
+#ifdef EROFS_MT_ENABLED
+#include "erofs/workqueue.h"
+struct erofs_workqueue z_erofs_mt_wq;
+#endif
+
+struct z_erofs_mt_read_ctx {
+ erofs_mutex_t lock;
+ erofs_cond_t cond;
+ int pending_tasks;
+ int final_err;
+ int outfd;
+ bool free_out;
+ struct z_erofs_decompress_task *current_task;
+};
+
+struct z_erofs_mt_read_ctx *z_erofs_mt_read_ctx_alloc(int outfd, bool free_out)
+{
+ struct z_erofs_mt_read_ctx *ctx = calloc(1, sizeof(*ctx));
+ if (!ctx)
+ return NULL;
+
+ erofs_mutex_init(&ctx->lock);
+ erofs_cond_init(&ctx->cond);
+ ctx->outfd = outfd;
+ ctx->free_out = free_out;
+ return ctx;
+}
+
+int z_erofs_mt_read_ctx_wait(struct z_erofs_mt_read_ctx *ctx)
+{
+ int err;
+ if (!ctx)
+ return 0;
+
+ erofs_mutex_lock(&ctx->lock);
+ while (ctx->pending_tasks > 0)
+ erofs_cond_wait(&ctx->cond, &ctx->lock);
+ err = ctx->final_err;
+ erofs_mutex_unlock(&ctx->lock);
+
+ return err;
+}
+
+void z_erofs_mt_read_ctx_free(struct z_erofs_mt_read_ctx *ctx)
+{
+ if (!ctx)
+ return;
+
+ erofs_mutex_destroy(&ctx->lock);
+ erofs_cond_destroy(&ctx->cond);
+ free(ctx);
+}
+
+struct z_erofs_decompress_item {
+ struct z_erofs_decompress_req req;
+ char *raw_buf;
+ char *out_buf;
+ erofs_off_t out_offset;
+ unsigned int out_length;
+};
+
+struct z_erofs_decompress_task {
+ struct erofs_work work;
+ struct z_erofs_mt_read_ctx *ctx;
+ struct z_erofs_decompress_item items[Z_EROFS_PCLUSTER_MAX_BATCH_SIZE];
+ unsigned int nr_reqs;
+};
+
+static void z_erofs_decompress_worker(struct erofs_work *work, void *tlsp)
+{
+ struct z_erofs_decompress_task *task = (struct z_erofs_decompress_task *)work;
+ struct z_erofs_mt_read_ctx *ctx = task->ctx;
+ int i, ret = 0, first_err = 0;
+
+ for (i = 0; i < task->nr_reqs; ++i) {
+ struct z_erofs_decompress_item *item = &task->items[i];
+ ret = z_erofs_decompress(&item->req);
+
+ if (ret >= 0 && ctx && ctx->outfd >= 0) {
+ if (pwrite(ctx->outfd, item->out_buf,
+ item->out_length, item->out_offset) < 0)
+ ret = -errno;
+ }
+
+ if (ret < 0 && !first_err)
+ first_err = ret;
+
+ free(item->raw_buf);
+ if (ctx && ctx->free_out)
+ free(item->out_buf);
+ }
+
+ if (ctx) {
+ erofs_mutex_lock(&ctx->lock);
+ if (first_err < 0 && !ctx->final_err)
+ ctx->final_err = first_err;
+ ctx->pending_tasks--;
+ if (!ctx->pending_tasks)
+ erofs_cond_signal(&ctx->cond);
+ erofs_mutex_unlock(&ctx->lock);
+ }
+ free(task);
+}
+
+void z_erofs_mt_read_enqueue(struct z_erofs_mt_read_ctx *ctx)
+{
+ if (ctx && ctx->current_task) {
+#ifdef EROFS_MT_ENABLED
+ erofs_queue_work(&z_erofs_mt_wq, &ctx->current_task->work);
+#else
+ z_erofs_decompress_worker(&ctx->current_task->work, NULL);
+#endif
+ ctx->current_task = NULL;
+ }
+}
void *erofs_bread(struct erofs_buf *buf, erofs_off_t offset, bool need_kmap)
{
@@ -277,20 +394,29 @@ static int erofs_read_raw_data(struct erofs_inode *inode, char *buffer,
int z_erofs_read_one_data(struct erofs_inode *inode,
struct erofs_map_blocks *map, char *raw, char *buffer,
- erofs_off_t skip, erofs_off_t length, bool trimmed)
+ erofs_off_t skip, erofs_off_t length, bool trimmed,
+ erofs_off_t out_offset, struct z_erofs_mt_read_ctx *ctx)
{
struct erofs_sb_info *sbi = inode->sbi;
struct erofs_map_dev mdev;
- int ret = 0;
+ struct z_erofs_decompress_task *task;
+ struct z_erofs_decompress_item *item;
+ int ret = 0, idx, batch_limit;
if (map->m_flags & __EROFS_MAP_FRAGMENT) {
if (__erofs_unlikely(inode->nid == sbi->packed_nid)) {
erofs_err("fragment should not exist in the packed inode %llu",
- sbi->packed_nid | 0ULL);
- return -EFSCORRUPTED;
+ sbi->packed_nid | 0ULL);
+ ret = -EFSCORRUPTED;
+ goto err_out;
}
- return erofs_packedfile_read(sbi, buffer, length - skip,
- inode->fragmentoff + skip);
+ ret = erofs_packedfile_read(sbi, buffer, length - skip,
+ inode->fragmentoff + skip);
+ if (ret >= 0 && ctx && ctx->outfd >= 0) {
+ if (pwrite(ctx->outfd, buffer, length - skip, out_offset) < 0)
+ ret = -errno;
+ }
+ goto err_out;
}
/* no device id here, thus it will always succeed */
@@ -300,31 +426,85 @@ int z_erofs_read_one_data(struct erofs_inode *inode,
ret = erofs_map_dev(sbi, &mdev);
if (ret) {
DBG_BUGON(1);
- return ret;
+ goto err_out;
}
ret = erofs_dev_read(sbi, mdev.m_deviceid, raw, mdev.m_pa, map->m_plen);
if (ret < 0)
+ goto err_out;
+
+ if (!ctx) {
+ ret = z_erofs_decompress(&(struct z_erofs_decompress_req) {
+ .sbi = sbi,
+ .in = raw,
+ .out = buffer,
+ .decodedskip = skip,
+ .interlaced_offset =
+ map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
+ erofs_blkoff(sbi, map->m_la) : 0,
+ .inputsize = map->m_plen,
+ .decodedlength = length,
+ .alg = map->m_algorithmformat,
+ .partial_decoding = trimmed ? true :
+ !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
+ (map->m_flags & EROFS_MAP_PARTIAL_REF),
+ });
return ret;
+ }
- ret = z_erofs_decompress(&(struct z_erofs_decompress_req) {
- .sbi = sbi,
- .in = raw,
- .out = buffer,
- .decodedskip = skip,
- .interlaced_offset =
- map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
- erofs_blkoff(sbi, map->m_la) : 0,
- .inputsize = map->m_plen,
- .decodedlength = length,
- .alg = map->m_algorithmformat,
- .partial_decoding = trimmed ? true :
- !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
- (map->m_flags & EROFS_MAP_PARTIAL_REF),
- });
- if (ret < 0)
- return ret;
+ task = ctx->current_task;
+ if (!task) {
+ task = calloc(1, sizeof(*task));
+ if (!task) {
+ ret = -ENOMEM;
+ goto err_out;
+ }
+ task->ctx = ctx;
+ task->work.fn = z_erofs_decompress_worker;
+ ctx->current_task = task;
+
+ erofs_mutex_lock(&ctx->lock);
+ ctx->pending_tasks++;
+ erofs_mutex_unlock(&ctx->lock);
+ }
+
+ idx = task->nr_reqs++;
+ item = &task->items[idx];
+
+ item->req = (struct z_erofs_decompress_req) {
+ .sbi = sbi,
+ .in = raw,
+ .out = buffer,
+ .decodedskip = skip,
+ .interlaced_offset =
+ map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
+ erofs_blkoff(sbi, map->m_la) : 0,
+ .inputsize = map->m_plen,
+ .decodedlength = length,
+ .alg = map->m_algorithmformat,
+ .partial_decoding = trimmed ? true :
+ !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
+ (map->m_flags & EROFS_MAP_PARTIAL_REF),
+ };
+ item->raw_buf = raw;
+ item->out_buf = buffer;
+ item->out_offset = out_offset;
+ item->out_length = length;
+
+ batch_limit = (map->m_algorithmformat == Z_EROFS_COMPRESSION_LZ4) ?
+ Z_EROFS_PCLUSTER_MAX_BATCH_SIZE : 8;
+
+ if (task->nr_reqs >= batch_limit)
+ z_erofs_mt_read_enqueue(ctx);
return 0;
+
+err_out:
+ if (ctx) {
+ if (ctx->free_out)
+ free(buffer);
+ free(raw);
+ }
+ return ret;
}
static int z_erofs_read_data(struct erofs_inode *inode, char *buffer,
@@ -387,7 +567,7 @@ static int z_erofs_read_data(struct erofs_inode *inode, char *buffer,
}
ret = z_erofs_read_one_data(inode, &map, raw,
- buffer + end - offset, skip, length, trimmed);
+ buffer + end - offset, skip, length, trimmed, 0, NULL);
if (ret < 0)
break;
}
diff --git a/lib/global.c b/lib/global.c
index 938aa0a..af2093b 100644
--- a/lib/global.c
+++ b/lib/global.c
@@ -12,6 +12,12 @@
#include "erofs/err.h"
#include "erofs/config.h"
#include "liberofs_compress.h"
+#include "erofs/internal.h"
+
+#ifdef EROFS_MT_ENABLED
+#include "erofs/workqueue.h"
+extern struct erofs_workqueue z_erofs_mt_wq;
+#endif
static EROFS_DEFINE_MUTEX(erofs_global_mutex);
#ifdef HAVE_LIBCURL
@@ -27,6 +33,14 @@ int liberofs_global_init(void)
#ifdef S3EROFS_ENABLED
xmlInitParser();
#endif
+#ifdef EROFS_MT_ENABLED
+ {
+ int workers = erofs_get_available_processors();
+ if (workers < 1)
+ workers = 1;
+ erofs_alloc_workqueue(&z_erofs_mt_wq, workers, 256, NULL, NULL);
+ }
+#endif
#ifdef HAVE_LIBCURL
if (!erofs_global_curl_initialized) {
if (curl_global_init(CURL_GLOBAL_DEFAULT) != CURLE_OK) {
@@ -45,6 +59,9 @@ void liberofs_global_exit(void)
{
erofs_mutex_lock(&erofs_global_mutex);
z_erofs_mt_global_exit();
+#ifdef EROFS_MT_ENABLED
+ erofs_destroy_workqueue(&z_erofs_mt_wq);
+#endif
#ifdef HAVE_LIBCURL
if (erofs_global_curl_initialized) {
curl_global_cleanup();
--
2.53.0
^ permalink raw reply related [flat|nested] 13+ messages in thread
* [PATCH 1/2 v3] fsck.erofs: add multi-threaded decompression
2026-07-06 6:05 ` [PATCH 1/2 v3] " Nithurshen
@ 2026-07-06 6:10 ` Nithurshen
2026-07-08 2:31 ` Gao Xiang
2026-07-13 7:02 ` [PATCH 1/2 v4] " Nithurshen
1 sibling, 1 reply; 13+ messages in thread
From: Nithurshen @ 2026-07-06 6:10 UTC (permalink / raw)
To: nithurshen.dev; +Cc: hsiangkao, linux-erofs, xiang
Hi Xiang,
Please find the updated multi-threaded decompression implementation for
fsck.erofs. This version introduces an algorithm-aware asynchronous
worker pool, dynamically sized based on system CPUs, to significantly
accelerate the extraction of computationally expensive images.
Benchmarks were performed on an ARM64 environment, extracting the 8.2 GB
Linux repository downloaded from Github.
Extraction Time (Seconds):
| Algorithm | 4k | 8k | 16k | 32k | 64k |
| --- | --- | --- | --- | --- | --- |
| lz4hc(MT) | 9.36 | 8.31 | 8.27 | 8.54 | 6.94 |
| lz4hc(ST) | 4.40 | 5.77 | 3.65 | 5.45 | 3.36 |
| zstd(MT) | 9.26 | 8.68 | 8.89 | 8.11 | 7.94 |
| zstd(ST) | 5.23 | 4.52 | 4.62 | 4.11 | 4.02 |
| lzma(MT) | 23.72 | 24.79 | 25.90 | 26.92 | 27.77 |
| lzma(ST) | 56.37 | 65.06 | 71.07 | 74.69 | 81.63 |
Performance Analysis:
The implementation provides a significant speedup for LZMA (up to 2.9x)
as the heavy decompression workload effectively amortizes the thread
synchronization overhead.
However, for fast algorithms like LZ4 and ZSTD, the current MT overhead
(futex contention and scheduling) leads to slower performance compared
to the synchronous baseline.
I have tried various batch sizes for fast algorithms, but the time did
not improve. (I tried from 32 to 256 batch sizes)
Can we fall back to synchronous extraction here?
Verification:
* Deadlocks/Race Conditions: Verified via GDB backtrace analysis and
stress testing.
* Memory Leaks: Verified via rigorous buffer ownership tracking and
ensuring all task resources are cleaned up upon worker completion.
* Integrity: All configurations passed bit-for-bit integrity checks
between the extracted and original directory for all algorithms and
chunk sizes.
All concurrency primitives and memory paths have been verified to the
best of my knowledge.
Thanks,
Nithurshen
^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [PATCH 1/2 v3] fsck.erofs: add multi-threaded decompression
2026-07-06 6:10 ` Nithurshen
@ 2026-07-08 2:31 ` Gao Xiang
0 siblings, 0 replies; 13+ messages in thread
From: Gao Xiang @ 2026-07-08 2:31 UTC (permalink / raw)
To: Nithurshen; +Cc: linux-erofs, xiang
On 2026/7/6 14:10, Nithurshen wrote:
> Hi Xiang,
>
> Please find the updated multi-threaded decompression implementation for
> fsck.erofs. This version introduces an algorithm-aware asynchronous
> worker pool, dynamically sized based on system CPUs, to significantly
> accelerate the extraction of computationally expensive images.
>
> Benchmarks were performed on an ARM64 environment, extracting the 8.2 GB
> Linux repository downloaded from Github.
>
> Extraction Time (Seconds):
>
> | Algorithm | 4k | 8k | 16k | 32k | 64k |
> | --- | --- | --- | --- | --- | --- |
> | lz4hc(MT) | 9.36 | 8.31 | 8.27 | 8.54 | 6.94 |
> | lz4hc(ST) | 4.40 | 5.77 | 3.65 | 5.45 | 3.36 |
> | zstd(MT) | 9.26 | 8.68 | 8.89 | 8.11 | 7.94 |
> | zstd(ST) | 5.23 | 4.52 | 4.62 | 4.11 | 4.02 |
> | lzma(MT) | 23.72 | 24.79 | 25.90 | 26.92 | 27.77 |
> | lzma(ST) | 56.37 | 65.06 | 71.07 | 74.69 | 81.63 |
>
> Performance Analysis:
> The implementation provides a significant speedup for LZMA (up to 2.9x)
> as the heavy decompression workload effectively amortizes the thread
> synchronization overhead.
>
> However, for fast algorithms like LZ4 and ZSTD, the current MT overhead
> (futex contention and scheduling) leads to slower performance compared
> to the synchronous baseline.
>
> I have tried various batch sizes for fast algorithms, but the time did
> not improve. (I tried from 32 to 256 batch sizes)
>
> Can we fall back to synchronous extraction here?
Sorry for delay.
We cannot, that is why we need to find a proper way to batch
the pclusters and reschedule.
And that is why I asked you to benchmark each commit.
Thanks,
Gao Xiang
>
> Verification:
>
> * Deadlocks/Race Conditions: Verified via GDB backtrace analysis and
> stress testing.
> * Memory Leaks: Verified via rigorous buffer ownership tracking and
> ensuring all task resources are cleaned up upon worker completion.
> * Integrity: All configurations passed bit-for-bit integrity checks
> between the extracted and original directory for all algorithms and
> chunk sizes.
>
> All concurrency primitives and memory paths have been verified to the
> best of my knowledge.
>
> Thanks,
> Nithurshen
^ permalink raw reply [flat|nested] 13+ messages in thread
* [PATCH 1/2 v4] fsck.erofs: add multi-threaded decompression
2026-07-06 6:05 ` [PATCH 1/2 v3] " Nithurshen
2026-07-06 6:10 ` Nithurshen
@ 2026-07-13 7:02 ` Nithurshen
1 sibling, 0 replies; 13+ messages in thread
From: Nithurshen @ 2026-07-13 7:02 UTC (permalink / raw)
To: nithurshen.dev; +Cc: hsiangkao, linux-erofs, xiang
Currently, fsck.erofs extracts files synchronously. When decompressing
heavily packed images, the main thread spends the majority of its time
blocked on a combination of synchronous I/O syscalls and CPU-heavy
decompression routines, bottlenecking overall extraction speed.
This patch introduces a scalable, multi-threaded decompression framework
using the existing erofs_workqueue infrastructure to decouple compute
from the main thread's I/O.
To prevent massive scheduling overhead (futex contention) where worker
threads spend more CPU time waking up than actually decompressing small
clusters, this implementation introduces a batching context. Because
different compression algorithms exhibit vastly different scheduling
thresholds, the batch limit is algorithm-aware.
Signed-off-by: Nithurshen <nithurshen.dev@gmail.com>
---
fsck/main.c | 127 ++++++++++++---
include/erofs/cond.h | 31 ++++
include/erofs/internal.h | 20 ++-
include/erofs/lock.h | 2 +
lib/data.c | 334 ++++++++++++++++++++++++++++++++++++---
lib/global.c | 11 +-
6 files changed, 473 insertions(+), 52 deletions(-)
create mode 100644 include/erofs/cond.h
diff --git a/fsck/main.c b/fsck/main.c
index b2d8f1a..b74ae68 100644
--- a/fsck/main.c
+++ b/fsck/main.c
@@ -16,6 +16,7 @@
#include "../lib/compressor.h"
#include "../lib/liberofs_compress.h"
#include "../lib/liberofs_sha256.h"
+#include "erofs/internal.h"
static int erofsfsck_check_inode(erofs_nid_t pnid, erofs_nid_t nid);
@@ -526,6 +527,13 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
u64 pchunk_len = 0;
u64 raw_size = 0, buffer_size = 0;
char *raw = NULL, *buffer = NULL;
+ struct z_erofs_mt_read_ctx *ctx = NULL;
+
+ if (!digest) {
+ ctx = z_erofs_mt_read_ctx_alloc(outfd, true);
+ if (!ctx)
+ return -ENOMEM;
+ }
erofs_dbg("verify data chunk of nid(%llu): type(%d)",
inode->nid | 0ULL, inode->datalayout);
@@ -569,7 +577,7 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
(const u8 *)zeros, chunk);
remain -= chunk;
}
- } else if (outfd >= 0) {
+ } else if (!ctx && outfd >= 0) {
ret = lseek(outfd, map.m_llen, SEEK_CUR);
if (ret < 0) {
ret = -errno;
@@ -592,8 +600,9 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
alloc_rawsize = map.m_plen;
}
- if (alloc_rawsize > raw_size) {
- char *newraw = realloc(raw, alloc_rawsize);
+ if (!ctx) {
+ if (alloc_rawsize > raw_size) {
+ char *newraw = realloc(raw, alloc_rawsize);
if (!newraw) {
ret = -ENOMEM;
@@ -602,9 +611,22 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
raw = newraw;
raw_size = alloc_rawsize;
}
+ }
if (compressed) {
- if (map.m_llen > buffer_size) {
+ char *c_raw = raw;
+ char *c_buf = buffer;
+
+ if (ctx) {
+ c_raw = malloc(alloc_rawsize);
+ c_buf = calloc(1, map.m_llen);
+ if (!c_raw || !c_buf) {
+ free(c_raw);
+ free(c_buf);
+ ret = -ENOMEM;
+ goto out;
+ }
+ } else if (map.m_llen > buffer_size) {
char *newbuffer;
buffer_size = map.m_llen;
@@ -614,45 +636,98 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
goto out;
}
buffer = newbuffer;
+ c_buf = buffer;
}
- ret = z_erofs_read_one_data(inode, &map, raw, buffer,
- 0, map.m_llen, false);
- if (ret)
+
+ ret = z_erofs_read_one_data(inode, &map, c_raw, c_buf,
+ 0, map.m_llen, false,
+ map.m_la, ctx);
+ if (ret) {
+ if (ctx) {
+ free(c_raw);
+ free(c_buf);
+ }
goto out;
+ }
- if (digest)
- erofs_sha256_process(digest,
- (const u8 *)buffer, map.m_llen);
- if (outfd >= 0 && write(outfd, buffer, map.m_llen) < 0)
- goto fail_eio;
+ if (!ctx) {
+ if (digest)
+ erofs_sha256_process(digest,
+ (const u8 *)c_buf, map.m_llen);
+ if (outfd >= 0 && write(outfd, c_buf, map.m_llen) < 0)
+ goto fail_eio;
+ }
} else {
+ char *c_raw = raw;
u64 p = 0;
+ erofs_off_t m_llen = map.m_llen;
+
+ if (ctx) {
+ c_raw = calloc(1, alloc_rawsize);
+ if (!c_raw) {
+ ret = -ENOMEM;
+ goto out;
+ }
+ }
do {
- u64 count = min_t(u64, alloc_rawsize,
- map.m_llen);
+ u64 count = min_t(u64, alloc_rawsize, m_llen);
- ret = erofs_read_one_data(inode, &map, raw, p, count);
- if (ret)
+ ret = erofs_read_one_data(inode, &map, c_raw, p, count);
+ if (ret) {
+ if (ctx)
+ free(c_raw);
goto out;
+ }
- if (digest)
- erofs_sha256_process(digest,
- (const u8 *)raw, count);
- if (outfd >= 0 && write(outfd, raw, count) < 0)
- goto fail_eio;
- map.m_llen -= count;
+ if (!ctx) {
+ if (digest)
+ erofs_sha256_process(digest,
+ (const u8 *)c_raw, count);
+ if (outfd >= 0 && write(outfd, c_raw, count) < 0)
+ goto fail_eio;
+ } else if (outfd >= 0) {
+ if (pwrite(outfd, c_raw, count, map.m_la + p) < 0) {
+ free(c_raw);
+ goto fail_eio;
+ }
+ }
+ m_llen -= count;
p += count;
- } while (map.m_llen);
+ } while (m_llen);
+
+ if (ctx)
+ free(c_raw);
}
}
+ if (ctx) {
+ int wait_err;
+
+ z_erofs_mt_read_enqueue(ctx);
+ wait_err = z_erofs_mt_read_ctx_wait(ctx);
+ if (wait_err < 0 && ret == 0)
+ ret = wait_err;
+ }
+
if (fsckcfg.print_comp_ratio) {
if (!erofs_is_packed_inode(inode))
fsckcfg.logical_blocks += BLK_ROUND_UP(inode->sbi, inode->i_size);
fsckcfg.physical_blocks += BLK_ROUND_UP(inode->sbi, pchunk_len);
}
out:
+ if (ctx) {
+ /* use ftruncate to handle trailing unmapped blocks (holes) */
+ if (outfd >= 0 && ret == 0) {
+ if (ftruncate(outfd, inode->i_size) < 0) {
+ erofs_err("failed to truncate file to %llu: %d",
+ (unsigned long long)inode->i_size, errno);
+ if (ret == 0)
+ ret = -errno;
+ }
+ }
+ z_erofs_mt_read_ctx_free(ctx);
+ }
if (raw)
free(raw);
if (buffer)
@@ -1127,7 +1202,9 @@ int main(int argc, char *argv[])
{
int err;
- erofs_init_configure();
+ err = liberofs_global_init();
+ if (err)
+ return 1;
fsckcfg.physical_blocks = 0;
fsckcfg.logical_blocks = 0;
@@ -1283,7 +1360,7 @@ exit_dev_close:
erofs_dev_close(&g_sbi);
exit:
erofs_blob_closeall(&g_sbi);
- erofs_exit_configure();
+ liberofs_global_exit();
return err ? 1 : 0;
}
diff --git a/include/erofs/cond.h b/include/erofs/cond.h
new file mode 100644
index 0000000..90ec838
--- /dev/null
+++ b/include/erofs/cond.h
@@ -0,0 +1,31 @@
+/* SPDX-License-Identifier: GPL-2.0+ OR Apache-2.0 */
+#ifndef __EROFS_COND_H
+#define __EROFS_COND_H
+
+#include "lock.h"
+
+#if defined(HAVE_PTHREAD_H) && defined(EROFS_MT_ENABLED)
+#include <pthread.h>
+
+typedef pthread_cond_t erofs_cond_t;
+
+static inline void erofs_cond_init(erofs_cond_t *cond)
+{
+ pthread_cond_init(cond, NULL);
+}
+#define erofs_cond_wait pthread_cond_wait
+#define erofs_cond_signal pthread_cond_signal
+#define erofs_cond_broadcast pthread_cond_broadcast
+#define erofs_cond_destroy pthread_cond_destroy
+
+#else
+typedef struct {} erofs_cond_t;
+
+static inline void erofs_cond_init(erofs_cond_t *cond) {}
+static inline int erofs_cond_wait(erofs_cond_t *cond, erofs_mutex_t *mutex) { return 0; }
+static inline int erofs_cond_signal(erofs_cond_t *cond) { return 0; }
+static inline int erofs_cond_broadcast(erofs_cond_t *cond) { return 0; }
+static inline int erofs_cond_destroy(erofs_cond_t *cond) { return 0; }
+#endif
+
+#endif
\ No newline at end of file
diff --git a/include/erofs/internal.h b/include/erofs/internal.h
index 2cc9cc8..6d3d407 100644
--- a/include/erofs/internal.h
+++ b/include/erofs/internal.h
@@ -25,6 +25,11 @@ typedef unsigned short umode_t;
#ifdef HAVE_PTHREAD_H
#include <pthread.h>
#endif
+#ifdef HAVE_PTHREAD_H
+#include <pthread.h>
+#endif
+#include <erofs/lock.h>
+#include "erofs/cond.h"
#include <stdlib.h>
#include <string.h>
#include "atomic.h"
@@ -62,6 +67,7 @@ struct erofs_buf {
#define erofs_pos(sbi, nr) ((erofs_off_t)(nr) << (sbi)->blkszbits)
#define BLK_ROUND_UP(sbi, addr) \
(roundup(addr, erofs_blksiz(sbi)) >> (sbi)->blkszbits)
+#define Z_EROFS_PCLUSTER_MAX_BATCH_SIZE 32
struct erofs_buffer_head;
struct erofs_bufmgr;
@@ -451,6 +457,17 @@ struct z_erofs_paramset {
char *extraopts;
};
+struct z_erofs_decompress_task;
+struct z_erofs_mt_read_ctx;
+
+struct z_erofs_mt_read_ctx *z_erofs_mt_read_ctx_alloc(int outfd, bool free_out);
+int z_erofs_mt_read_ctx_wait(struct z_erofs_mt_read_ctx *ctx);
+void z_erofs_mt_read_ctx_free(struct z_erofs_mt_read_ctx *ctx);
+void z_erofs_mt_read_enqueue(struct z_erofs_mt_read_ctx *ctx);
+
+int z_erofs_mt_workers_init(void);
+void z_erofs_mt_workers_exit(void);
+
int liberofs_global_init(void);
void liberofs_global_exit(void);
@@ -487,7 +504,8 @@ int erofs_read_one_data(struct erofs_inode *inode, struct erofs_map_blocks *map,
char *buffer, u64 offset, size_t len);
int z_erofs_read_one_data(struct erofs_inode *inode,
struct erofs_map_blocks *map, char *raw, char *buffer,
- erofs_off_t skip, erofs_off_t length, bool trimmed);
+ erofs_off_t skip, erofs_off_t length, bool trimmed,
+ erofs_off_t out_offset, struct z_erofs_mt_read_ctx *ctx);
void *erofs_read_metadata(struct erofs_sb_info *sbi, erofs_nid_t nid,
erofs_off_t *offset, int *lengthp);
int z_erofs_parse_cfgs(struct erofs_sb_info *sbi, struct erofs_super_block *dsb);
diff --git a/include/erofs/lock.h b/include/erofs/lock.h
index 884f23e..e08b76e 100644
--- a/include/erofs/lock.h
+++ b/include/erofs/lock.h
@@ -15,6 +15,7 @@ static inline void erofs_mutex_init(erofs_mutex_t *lock)
}
#define erofs_mutex_lock pthread_mutex_lock
#define erofs_mutex_unlock pthread_mutex_unlock
+#define erofs_mutex_destroy pthread_mutex_destroy
#define EROFS_DEFINE_MUTEX(lock) \
erofs_mutex_t lock = PTHREAD_MUTEX_INITIALIZER
@@ -35,6 +36,7 @@ typedef struct {} erofs_mutex_t;
static inline void erofs_mutex_init(erofs_mutex_t *lock) {}
static inline void erofs_mutex_lock(erofs_mutex_t *lock) {}
static inline void erofs_mutex_unlock(erofs_mutex_t *lock) {}
+static inline void erofs_mutex_destroy(erofs_mutex_t *lock) {}
#define EROFS_DEFINE_MUTEX(lock) \
erofs_mutex_t lock = {}
diff --git a/lib/data.c b/lib/data.c
index 1bb9269..6bcd5b1 100644
--- a/lib/data.c
+++ b/lib/data.c
@@ -9,6 +9,228 @@
#include "erofs/trace.h"
#include "erofs/decompress.h"
#include "liberofs_fragments.h"
+#include "erofs/lock.h"
+
+#ifdef EROFS_MT_ENABLED
+#include "erofs/config.h"
+
+struct z_erofs_percpu_queue {
+ erofs_mutex_t lock;
+ erofs_cond_t cond;
+ struct z_erofs_decompress_task *head;
+ struct z_erofs_decompress_task *tail;
+ int pending_tasks;
+ bool shutdown;
+ pthread_t thread;
+} __attribute__((aligned(64)));
+
+static struct z_erofs_percpu_queue *worker_queues;
+static int num_workers;
+#endif
+
+struct z_erofs_mt_read_ctx {
+ erofs_mutex_t lock;
+ erofs_cond_t cond;
+ int pending_tasks;
+ int final_err;
+ int outfd;
+ bool free_out;
+ struct z_erofs_decompress_task *current_task;
+};
+
+struct z_erofs_mt_read_ctx *z_erofs_mt_read_ctx_alloc(int outfd, bool free_out)
+{
+ struct z_erofs_mt_read_ctx *ctx = calloc(1, sizeof(*ctx));
+ if (!ctx)
+ return NULL;
+
+ erofs_mutex_init(&ctx->lock);
+ erofs_cond_init(&ctx->cond);
+ ctx->outfd = outfd;
+ ctx->free_out = free_out;
+ return ctx;
+}
+
+int z_erofs_mt_read_ctx_wait(struct z_erofs_mt_read_ctx *ctx)
+{
+ int err;
+ if (!ctx)
+ return 0;
+
+ erofs_mutex_lock(&ctx->lock);
+ while (ctx->pending_tasks > 0)
+ erofs_cond_wait(&ctx->cond, &ctx->lock);
+ err = ctx->final_err;
+ erofs_mutex_unlock(&ctx->lock);
+
+ return err;
+}
+
+void z_erofs_mt_read_ctx_free(struct z_erofs_mt_read_ctx *ctx)
+{
+ if (!ctx)
+ return;
+
+ erofs_mutex_destroy(&ctx->lock);
+ erofs_cond_destroy(&ctx->cond);
+ free(ctx);
+}
+
+struct z_erofs_decompress_item {
+ struct z_erofs_decompress_req req;
+ char *raw_buf;
+ char *out_buf;
+ erofs_off_t out_offset;
+ unsigned int out_length;
+};
+
+struct z_erofs_decompress_task {
+ struct z_erofs_decompress_task *next;
+ struct z_erofs_mt_read_ctx *ctx;
+ struct z_erofs_decompress_item items[Z_EROFS_PCLUSTER_MAX_BATCH_SIZE];
+ unsigned int nr_reqs;
+};
+
+static void z_erofs_process_task(struct z_erofs_decompress_task *task)
+{
+ struct z_erofs_mt_read_ctx *ctx = task->ctx;
+ int i, ret = 0, first_err = 0;
+
+ for (i = 0; i < task->nr_reqs; ++i) {
+ struct z_erofs_decompress_item *item = &task->items[i];
+ ret = z_erofs_decompress(&item->req);
+
+ if (ret >= 0 && ctx && ctx->outfd >= 0) {
+ if (pwrite(ctx->outfd, item->out_buf,
+ item->out_length, item->out_offset) < 0)
+ ret = -errno;
+ }
+
+ if (ret < 0 && !first_err)
+ first_err = ret;
+
+ free(item->raw_buf);
+ if (ctx && ctx->free_out)
+ free(item->out_buf);
+ }
+
+ if (ctx) {
+ erofs_mutex_lock(&ctx->lock);
+ if (first_err < 0 && !ctx->final_err)
+ ctx->final_err = first_err;
+ ctx->pending_tasks--;
+ if (!ctx->pending_tasks)
+ erofs_cond_signal(&ctx->cond);
+ erofs_mutex_unlock(&ctx->lock);
+ }
+ free(task);
+}
+
+#ifdef EROFS_MT_ENABLED
+static void *z_erofs_worker_thread(void *arg)
+{
+ struct z_erofs_percpu_queue *q = arg;
+
+ while (1) {
+ erofs_mutex_lock(&q->lock);
+ while (!q->head && !q->shutdown)
+ erofs_cond_wait(&q->cond, &q->lock);
+
+ if (q->shutdown && !q->head) {
+ erofs_mutex_unlock(&q->lock);
+ break;
+ }
+
+ struct z_erofs_decompress_task *tasks = q->head;
+ q->head = q->tail = NULL;
+ q->pending_tasks = 0;
+ erofs_mutex_unlock(&q->lock);
+
+ while (tasks) {
+ struct z_erofs_decompress_task *task = tasks;
+ tasks = tasks->next;
+ z_erofs_process_task(task);
+ }
+ }
+ return NULL;
+}
+
+int z_erofs_mt_workers_init(void)
+{
+ int i;
+ num_workers = erofs_get_available_processors();
+ if (num_workers < 1)
+ num_workers = 1;
+
+ worker_queues = calloc(num_workers, sizeof(*worker_queues));
+ if (!worker_queues)
+ return -ENOMEM;
+
+ for (i = 0; i < num_workers; ++i) {
+ erofs_mutex_init(&worker_queues[i].lock);
+ erofs_cond_init(&worker_queues[i].cond);
+ worker_queues[i].head = worker_queues[i].tail = NULL;
+ worker_queues[i].pending_tasks = 0;
+ worker_queues[i].shutdown = false;
+ pthread_create(&worker_queues[i].thread, NULL, z_erofs_worker_thread, &worker_queues[i]);
+ }
+ return 0;
+}
+
+void z_erofs_mt_workers_exit(void)
+{
+ int i;
+ if (!worker_queues) return;
+ for (i = 0; i < num_workers; ++i) {
+ erofs_mutex_lock(&worker_queues[i].lock);
+ worker_queues[i].shutdown = true;
+ erofs_cond_signal(&worker_queues[i].cond);
+ erofs_mutex_unlock(&worker_queues[i].lock);
+
+ pthread_join(worker_queues[i].thread, NULL);
+ erofs_cond_destroy(&worker_queues[i].cond);
+ erofs_mutex_destroy(&worker_queues[i].lock);
+ }
+ free(worker_queues);
+ worker_queues = NULL;
+}
+#endif
+
+void z_erofs_mt_read_enqueue(struct z_erofs_mt_read_ctx *ctx)
+{
+#ifdef EROFS_MT_ENABLED
+ static int next_worker = 0;
+#endif
+
+ if (!ctx || !ctx->current_task)
+ return;
+
+ struct z_erofs_decompress_task *task = ctx->current_task;
+ ctx->current_task = NULL;
+
+#ifdef EROFS_MT_ENABLED
+ if (num_workers > 0) {
+ int target = next_worker;
+ next_worker = (next_worker + 1) % num_workers;
+ struct z_erofs_percpu_queue *q = &worker_queues[target];
+
+ erofs_mutex_lock(&q->lock);
+ task->next = NULL;
+ if (!q->tail) {
+ q->head = q->tail = task;
+ } else {
+ q->tail->next = task;
+ q->tail = task;
+ }
+ q->pending_tasks++;
+ erofs_cond_signal(&q->cond);
+ erofs_mutex_unlock(&q->lock);
+ return;
+ }
+#endif
+ task->next = NULL;
+ z_erofs_process_task(task);
+}
void *erofs_bread(struct erofs_buf *buf, erofs_off_t offset, bool need_kmap)
{
@@ -277,20 +499,29 @@ static int erofs_read_raw_data(struct erofs_inode *inode, char *buffer,
int z_erofs_read_one_data(struct erofs_inode *inode,
struct erofs_map_blocks *map, char *raw, char *buffer,
- erofs_off_t skip, erofs_off_t length, bool trimmed)
+ erofs_off_t skip, erofs_off_t length, bool trimmed,
+ erofs_off_t out_offset, struct z_erofs_mt_read_ctx *ctx)
{
struct erofs_sb_info *sbi = inode->sbi;
struct erofs_map_dev mdev;
- int ret = 0;
+ struct z_erofs_decompress_task *task;
+ struct z_erofs_decompress_item *item;
+ int ret = 0, idx, batch_limit;
if (map->m_flags & __EROFS_MAP_FRAGMENT) {
if (__erofs_unlikely(inode->nid == sbi->packed_nid)) {
erofs_err("fragment should not exist in the packed inode %llu",
- sbi->packed_nid | 0ULL);
- return -EFSCORRUPTED;
+ sbi->packed_nid | 0ULL);
+ ret = -EFSCORRUPTED;
+ goto err_out;
+ }
+ ret = erofs_packedfile_read(sbi, buffer, length - skip,
+ inode->fragmentoff + skip);
+ if (ret >= 0 && ctx && ctx->outfd >= 0) {
+ if (pwrite(ctx->outfd, buffer, length - skip, out_offset) < 0)
+ ret = -errno;
}
- return erofs_packedfile_read(sbi, buffer, length - skip,
- inode->fragmentoff + skip);
+ goto err_out;
}
/* no device id here, thus it will always succeed */
@@ -300,31 +531,84 @@ int z_erofs_read_one_data(struct erofs_inode *inode,
ret = erofs_map_dev(sbi, &mdev);
if (ret) {
DBG_BUGON(1);
- return ret;
+ goto err_out;
}
ret = erofs_dev_read(sbi, mdev.m_deviceid, raw, mdev.m_pa, map->m_plen);
if (ret < 0)
+ goto err_out;
+
+ if (!ctx) {
+ ret = z_erofs_decompress(&(struct z_erofs_decompress_req) {
+ .sbi = sbi,
+ .in = raw,
+ .out = buffer,
+ .decodedskip = skip,
+ .interlaced_offset =
+ map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
+ erofs_blkoff(sbi, map->m_la) : 0,
+ .inputsize = map->m_plen,
+ .decodedlength = length,
+ .alg = map->m_algorithmformat,
+ .partial_decoding = trimmed ? true :
+ !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
+ (map->m_flags & EROFS_MAP_PARTIAL_REF),
+ });
return ret;
+ }
- ret = z_erofs_decompress(&(struct z_erofs_decompress_req) {
- .sbi = sbi,
- .in = raw,
- .out = buffer,
- .decodedskip = skip,
- .interlaced_offset =
- map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
- erofs_blkoff(sbi, map->m_la) : 0,
- .inputsize = map->m_plen,
- .decodedlength = length,
- .alg = map->m_algorithmformat,
- .partial_decoding = trimmed ? true :
- !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
- (map->m_flags & EROFS_MAP_PARTIAL_REF),
- });
- if (ret < 0)
- return ret;
+ task = ctx->current_task;
+ if (!task) {
+ task = calloc(1, sizeof(*task));
+ if (!task) {
+ ret = -ENOMEM;
+ goto err_out;
+ }
+ task->ctx = ctx;
+ ctx->current_task = task;
+
+ erofs_mutex_lock(&ctx->lock);
+ ctx->pending_tasks++;
+ erofs_mutex_unlock(&ctx->lock);
+ }
+
+ idx = task->nr_reqs++;
+ item = &task->items[idx];
+
+ item->req = (struct z_erofs_decompress_req) {
+ .sbi = sbi,
+ .in = raw,
+ .out = buffer,
+ .decodedskip = skip,
+ .interlaced_offset =
+ map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
+ erofs_blkoff(sbi, map->m_la) : 0,
+ .inputsize = map->m_plen,
+ .decodedlength = length,
+ .alg = map->m_algorithmformat,
+ .partial_decoding = trimmed ? true :
+ !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
+ (map->m_flags & EROFS_MAP_PARTIAL_REF),
+ };
+ item->raw_buf = raw;
+ item->out_buf = buffer;
+ item->out_offset = out_offset;
+ item->out_length = length;
+
+ batch_limit = (map->m_algorithmformat == Z_EROFS_COMPRESSION_LZ4) ?
+ Z_EROFS_PCLUSTER_MAX_BATCH_SIZE : 8;
+
+ if (task->nr_reqs >= batch_limit)
+ z_erofs_mt_read_enqueue(ctx);
return 0;
+
+err_out:
+ if (ctx) {
+ if (ctx->free_out)
+ free(buffer);
+ free(raw);
+ }
+ return ret;
}
static int z_erofs_read_data(struct erofs_inode *inode, char *buffer,
@@ -387,7 +671,7 @@ static int z_erofs_read_data(struct erofs_inode *inode, char *buffer,
}
ret = z_erofs_read_one_data(inode, &map, raw,
- buffer + end - offset, skip, length, trimmed);
+ buffer + end - offset, skip, length, trimmed, 0, NULL);
if (ret < 0)
break;
}
diff --git a/lib/global.c b/lib/global.c
index 938aa0a..8bb9da0 100644
--- a/lib/global.c
+++ b/lib/global.c
@@ -12,6 +12,7 @@
#include "erofs/err.h"
#include "erofs/config.h"
#include "liberofs_compress.h"
+#include "erofs/internal.h"
static EROFS_DEFINE_MUTEX(erofs_global_mutex);
#ifdef HAVE_LIBCURL
@@ -27,6 +28,11 @@ int liberofs_global_init(void)
#ifdef S3EROFS_ENABLED
xmlInitParser();
#endif
+#ifdef EROFS_MT_ENABLED
+ err = z_erofs_mt_workers_init();
+ if (err)
+ goto out_unlock;
+#endif
#ifdef HAVE_LIBCURL
if (!erofs_global_curl_initialized) {
if (curl_global_init(CURL_GLOBAL_DEFAULT) != CURLE_OK) {
@@ -35,8 +41,8 @@ int liberofs_global_init(void)
}
erofs_global_curl_initialized = true;
}
-out_unlock:
#endif
+out_unlock:
erofs_mutex_unlock(&erofs_global_mutex);
return err;
}
@@ -45,6 +51,9 @@ void liberofs_global_exit(void)
{
erofs_mutex_lock(&erofs_global_mutex);
z_erofs_mt_global_exit();
+#ifdef EROFS_MT_ENABLED
+ z_erofs_mt_workers_exit();
+#endif
#ifdef HAVE_LIBCURL
if (erofs_global_curl_initialized) {
curl_global_cleanup();
--
2.53.0
^ permalink raw reply related [flat|nested] 13+ messages in thread
* [PATCH v5] fsck.erofs: add multi-threaded decompression
2026-06-21 12:01 ` [PATCH 2/2] fsck.erofs: implement concurrent directory traversal Nithurshen
2026-07-05 14:05 ` Gao Xiang
@ 2026-07-14 1:10 ` Nithurshen
2026-07-14 1:18 ` Nithurshen Karthikeyan
1 sibling, 1 reply; 13+ messages in thread
From: Nithurshen @ 2026-07-14 1:10 UTC (permalink / raw)
To: nithurshen.dev; +Cc: hsiangkao, linux-erofs, xiang
Currently, fsck.erofs extracts files synchronously. When decompressing
heavily packed images, the main thread spends the majority of its time
blocked on a combination of synchronous I/O syscalls and CPU-heavy
decompression routines, bottlenecking overall extraction speed.
This patch introduces a scalable, multi-threaded decompression framework
to decouple compute operations from the main thread's I/O. This is
achieved by implementing a dedicated, per-CPU worker queue system
alongside custom condition variable wrappers to handle decompression
tasks asynchronously.
To prevent massive scheduling overhead (futex contention) where worker
threads spend more CPU time waking up than actually decompressing small
clusters, this implementation introduces a batching context
(`z_erofs_mt_read_ctx`). Because different compression algorithms exhibit
vastly different scheduling thresholds, the batch limit is algorithm-aware,
and multi-threading is bypassed entirely for files under a 128KB threshold.
Signed-off-by: Nithurshen <nithurshen.dev@gmail.com>
---
fsck/main.c | 136 +++++++++++++---
include/erofs/cond.h | 31 ++++
include/erofs/internal.h | 20 ++-
include/erofs/lock.h | 2 +
lib/data.c | 336 ++++++++++++++++++++++++++++++++++++---
lib/global.c | 11 +-
6 files changed, 484 insertions(+), 52 deletions(-)
create mode 100644 include/erofs/cond.h
diff --git a/fsck/main.c b/fsck/main.c
index b2d8f1a..be1d3c5 100644
--- a/fsck/main.c
+++ b/fsck/main.c
@@ -16,6 +16,7 @@
#include "../lib/compressor.h"
#include "../lib/liberofs_compress.h"
#include "../lib/liberofs_sha256.h"
+#include "erofs/internal.h"
static int erofsfsck_check_inode(erofs_nid_t pnid, erofs_nid_t nid);
@@ -513,6 +514,8 @@ out:
return ret;
}
+#define EROFS_MT_THRESHOLD (128 * 1024)
+
static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
struct sha256_state *digest)
{
@@ -526,6 +529,19 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
u64 pchunk_len = 0;
u64 raw_size = 0, buffer_size = 0;
char *raw = NULL, *buffer = NULL;
+ struct z_erofs_mt_read_ctx *ctx = NULL;
+
+ if (!digest && inode->i_size > EROFS_MT_THRESHOLD) {
+ ctx = z_erofs_mt_read_ctx_alloc(outfd, true);
+ if (!ctx)
+ return -ENOMEM;
+ }
+ if (outfd >= 0) {
+ if (ftruncate(outfd, inode->i_size) < 0) {
+ erofs_warn("failed to pre-allocate file to %llu",
+ (unsigned long long)inode->i_size);
+ }
+ }
erofs_dbg("verify data chunk of nid(%llu): type(%d)",
inode->nid | 0ULL, inode->datalayout);
@@ -569,7 +585,7 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
(const u8 *)zeros, chunk);
remain -= chunk;
}
- } else if (outfd >= 0) {
+ } else if (!ctx && outfd >= 0) {
ret = lseek(outfd, map.m_llen, SEEK_CUR);
if (ret < 0) {
ret = -errno;
@@ -592,8 +608,9 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
alloc_rawsize = map.m_plen;
}
- if (alloc_rawsize > raw_size) {
- char *newraw = realloc(raw, alloc_rawsize);
+ if (!ctx) {
+ if (alloc_rawsize > raw_size) {
+ char *newraw = realloc(raw, alloc_rawsize);
if (!newraw) {
ret = -ENOMEM;
@@ -602,9 +619,22 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
raw = newraw;
raw_size = alloc_rawsize;
}
+ }
if (compressed) {
- if (map.m_llen > buffer_size) {
+ char *c_raw = raw;
+ char *c_buf = buffer;
+
+ if (ctx) {
+ c_raw = malloc(alloc_rawsize);
+ c_buf = malloc(map.m_llen);
+ if (!c_raw || !c_buf) {
+ free(c_raw);
+ free(c_buf);
+ ret = -ENOMEM;
+ goto out;
+ }
+ } else if (map.m_llen > buffer_size) {
char *newbuffer;
buffer_size = map.m_llen;
@@ -614,45 +644,99 @@ static int erofs_verify_inode_data(struct erofs_inode *inode, int outfd,
goto out;
}
buffer = newbuffer;
+ c_buf = buffer;
}
- ret = z_erofs_read_one_data(inode, &map, raw, buffer,
- 0, map.m_llen, false);
- if (ret)
+
+ ret = z_erofs_read_one_data(inode, &map, c_raw, c_buf,
+ 0, map.m_llen, false,
+ map.m_la, ctx);
+ if (ret < 0) {
+ if (ctx) {
+ free(c_raw);
+ free(c_buf);
+ }
goto out;
+ }
- if (digest)
- erofs_sha256_process(digest,
- (const u8 *)buffer, map.m_llen);
- if (outfd >= 0 && write(outfd, buffer, map.m_llen) < 0)
- goto fail_eio;
+ if (!ctx) {
+ if (digest)
+ erofs_sha256_process(digest,
+ (const u8 *)c_buf, map.m_llen);
+ if (outfd >= 0 && write(outfd, c_buf, map.m_llen) < 0)
+ goto fail_eio;
+ }
} else {
+ char *c_raw = raw;
u64 p = 0;
+ erofs_off_t m_llen = map.m_llen;
+
+ if (ctx) {
+ c_raw = malloc(alloc_rawsize);
+ if (!c_raw) {
+ ret = -ENOMEM;
+ goto out;
+ }
+ }
do {
- u64 count = min_t(u64, alloc_rawsize,
- map.m_llen);
+ u64 count = min_t(u64, alloc_rawsize, m_llen);
- ret = erofs_read_one_data(inode, &map, raw, p, count);
- if (ret)
+ ret = erofs_read_one_data(inode, &map, c_raw, p, count);
+ if (ret < 0) {
+ if (ctx)
+ free(c_raw);
goto out;
+ }
- if (digest)
- erofs_sha256_process(digest,
- (const u8 *)raw, count);
- if (outfd >= 0 && write(outfd, raw, count) < 0)
- goto fail_eio;
- map.m_llen -= count;
+ if (!ctx) {
+ if (digest)
+ erofs_sha256_process(digest,
+ (const u8 *)c_raw, count);
+ if (outfd >= 0 && write(outfd, c_raw, count) < 0)
+ goto fail_eio;
+ } else if (outfd >= 0) {
+ if (pwrite(outfd, c_raw, count, map.m_la + p) < 0) {
+ free(c_raw);
+ goto fail_eio;
+ }
+ }
+ m_llen -= count;
p += count;
- } while (map.m_llen);
+ } while (m_llen);
+
+ if (ctx)
+ free(c_raw);
}
}
+ if (ctx) {
+ int wait_err;
+
+ z_erofs_mt_read_enqueue(ctx);
+ wait_err = z_erofs_mt_read_ctx_wait(ctx);
+ if (wait_err < 0 && ret == 0)
+ ret = wait_err;
+ }
+
if (fsckcfg.print_comp_ratio) {
if (!erofs_is_packed_inode(inode))
fsckcfg.logical_blocks += BLK_ROUND_UP(inode->sbi, inode->i_size);
fsckcfg.physical_blocks += BLK_ROUND_UP(inode->sbi, pchunk_len);
}
out:
+ /* use ftruncate to handle trailing unmapped blocks (holes) */
+ if (outfd >= 0 && ret == 0) {
+ if (ftruncate(outfd, inode->i_size) < 0) {
+ erofs_err("failed to truncate file to %llu: %d",
+ (unsigned long long)inode->i_size, errno);
+ if (ret == 0)
+ ret = -errno;
+ }
+ }
+
+ if (ctx) {
+ z_erofs_mt_read_ctx_free(ctx);
+ }
if (raw)
free(raw);
if (buffer)
@@ -1127,7 +1211,9 @@ int main(int argc, char *argv[])
{
int err;
- erofs_init_configure();
+ err = liberofs_global_init();
+ if (err)
+ return 1;
fsckcfg.physical_blocks = 0;
fsckcfg.logical_blocks = 0;
@@ -1283,7 +1369,7 @@ exit_dev_close:
erofs_dev_close(&g_sbi);
exit:
erofs_blob_closeall(&g_sbi);
- erofs_exit_configure();
+ liberofs_global_exit();
return err ? 1 : 0;
}
diff --git a/include/erofs/cond.h b/include/erofs/cond.h
new file mode 100644
index 0000000..90ec838
--- /dev/null
+++ b/include/erofs/cond.h
@@ -0,0 +1,31 @@
+/* SPDX-License-Identifier: GPL-2.0+ OR Apache-2.0 */
+#ifndef __EROFS_COND_H
+#define __EROFS_COND_H
+
+#include "lock.h"
+
+#if defined(HAVE_PTHREAD_H) && defined(EROFS_MT_ENABLED)
+#include <pthread.h>
+
+typedef pthread_cond_t erofs_cond_t;
+
+static inline void erofs_cond_init(erofs_cond_t *cond)
+{
+ pthread_cond_init(cond, NULL);
+}
+#define erofs_cond_wait pthread_cond_wait
+#define erofs_cond_signal pthread_cond_signal
+#define erofs_cond_broadcast pthread_cond_broadcast
+#define erofs_cond_destroy pthread_cond_destroy
+
+#else
+typedef struct {} erofs_cond_t;
+
+static inline void erofs_cond_init(erofs_cond_t *cond) {}
+static inline int erofs_cond_wait(erofs_cond_t *cond, erofs_mutex_t *mutex) { return 0; }
+static inline int erofs_cond_signal(erofs_cond_t *cond) { return 0; }
+static inline int erofs_cond_broadcast(erofs_cond_t *cond) { return 0; }
+static inline int erofs_cond_destroy(erofs_cond_t *cond) { return 0; }
+#endif
+
+#endif
\ No newline at end of file
diff --git a/include/erofs/internal.h b/include/erofs/internal.h
index 2cc9cc8..6d3d407 100644
--- a/include/erofs/internal.h
+++ b/include/erofs/internal.h
@@ -25,6 +25,11 @@ typedef unsigned short umode_t;
#ifdef HAVE_PTHREAD_H
#include <pthread.h>
#endif
+#ifdef HAVE_PTHREAD_H
+#include <pthread.h>
+#endif
+#include <erofs/lock.h>
+#include "erofs/cond.h"
#include <stdlib.h>
#include <string.h>
#include "atomic.h"
@@ -62,6 +67,7 @@ struct erofs_buf {
#define erofs_pos(sbi, nr) ((erofs_off_t)(nr) << (sbi)->blkszbits)
#define BLK_ROUND_UP(sbi, addr) \
(roundup(addr, erofs_blksiz(sbi)) >> (sbi)->blkszbits)
+#define Z_EROFS_PCLUSTER_MAX_BATCH_SIZE 32
struct erofs_buffer_head;
struct erofs_bufmgr;
@@ -451,6 +457,17 @@ struct z_erofs_paramset {
char *extraopts;
};
+struct z_erofs_decompress_task;
+struct z_erofs_mt_read_ctx;
+
+struct z_erofs_mt_read_ctx *z_erofs_mt_read_ctx_alloc(int outfd, bool free_out);
+int z_erofs_mt_read_ctx_wait(struct z_erofs_mt_read_ctx *ctx);
+void z_erofs_mt_read_ctx_free(struct z_erofs_mt_read_ctx *ctx);
+void z_erofs_mt_read_enqueue(struct z_erofs_mt_read_ctx *ctx);
+
+int z_erofs_mt_workers_init(void);
+void z_erofs_mt_workers_exit(void);
+
int liberofs_global_init(void);
void liberofs_global_exit(void);
@@ -487,7 +504,8 @@ int erofs_read_one_data(struct erofs_inode *inode, struct erofs_map_blocks *map,
char *buffer, u64 offset, size_t len);
int z_erofs_read_one_data(struct erofs_inode *inode,
struct erofs_map_blocks *map, char *raw, char *buffer,
- erofs_off_t skip, erofs_off_t length, bool trimmed);
+ erofs_off_t skip, erofs_off_t length, bool trimmed,
+ erofs_off_t out_offset, struct z_erofs_mt_read_ctx *ctx);
void *erofs_read_metadata(struct erofs_sb_info *sbi, erofs_nid_t nid,
erofs_off_t *offset, int *lengthp);
int z_erofs_parse_cfgs(struct erofs_sb_info *sbi, struct erofs_super_block *dsb);
diff --git a/include/erofs/lock.h b/include/erofs/lock.h
index 884f23e..e08b76e 100644
--- a/include/erofs/lock.h
+++ b/include/erofs/lock.h
@@ -15,6 +15,7 @@ static inline void erofs_mutex_init(erofs_mutex_t *lock)
}
#define erofs_mutex_lock pthread_mutex_lock
#define erofs_mutex_unlock pthread_mutex_unlock
+#define erofs_mutex_destroy pthread_mutex_destroy
#define EROFS_DEFINE_MUTEX(lock) \
erofs_mutex_t lock = PTHREAD_MUTEX_INITIALIZER
@@ -35,6 +36,7 @@ typedef struct {} erofs_mutex_t;
static inline void erofs_mutex_init(erofs_mutex_t *lock) {}
static inline void erofs_mutex_lock(erofs_mutex_t *lock) {}
static inline void erofs_mutex_unlock(erofs_mutex_t *lock) {}
+static inline void erofs_mutex_destroy(erofs_mutex_t *lock) {}
#define EROFS_DEFINE_MUTEX(lock) \
erofs_mutex_t lock = {}
diff --git a/lib/data.c b/lib/data.c
index 1bb9269..ef9f29b 100644
--- a/lib/data.c
+++ b/lib/data.c
@@ -9,6 +9,228 @@
#include "erofs/trace.h"
#include "erofs/decompress.h"
#include "liberofs_fragments.h"
+#include "erofs/lock.h"
+
+#ifdef EROFS_MT_ENABLED
+#include "erofs/config.h"
+
+struct z_erofs_percpu_queue {
+ erofs_mutex_t lock;
+ erofs_cond_t cond;
+ struct z_erofs_decompress_task *head;
+ struct z_erofs_decompress_task *tail;
+ int pending_tasks;
+ bool shutdown;
+ pthread_t thread;
+} __attribute__((aligned(64)));
+
+static struct z_erofs_percpu_queue *worker_queues;
+static int num_workers;
+#endif
+
+struct z_erofs_mt_read_ctx {
+ erofs_mutex_t lock;
+ erofs_cond_t cond;
+ int pending_tasks;
+ int final_err;
+ int outfd;
+ bool free_out;
+ struct z_erofs_decompress_task *current_task;
+};
+
+struct z_erofs_mt_read_ctx *z_erofs_mt_read_ctx_alloc(int outfd, bool free_out)
+{
+ struct z_erofs_mt_read_ctx *ctx = calloc(1, sizeof(*ctx));
+ if (!ctx)
+ return NULL;
+
+ erofs_mutex_init(&ctx->lock);
+ erofs_cond_init(&ctx->cond);
+ ctx->outfd = outfd;
+ ctx->free_out = free_out;
+ return ctx;
+}
+
+int z_erofs_mt_read_ctx_wait(struct z_erofs_mt_read_ctx *ctx)
+{
+ int err;
+ if (!ctx)
+ return 0;
+
+ erofs_mutex_lock(&ctx->lock);
+ while (ctx->pending_tasks > 0)
+ erofs_cond_wait(&ctx->cond, &ctx->lock);
+ err = ctx->final_err;
+ erofs_mutex_unlock(&ctx->lock);
+
+ return err;
+}
+
+void z_erofs_mt_read_ctx_free(struct z_erofs_mt_read_ctx *ctx)
+{
+ if (!ctx)
+ return;
+
+ erofs_mutex_destroy(&ctx->lock);
+ erofs_cond_destroy(&ctx->cond);
+ free(ctx);
+}
+
+struct z_erofs_decompress_item {
+ struct z_erofs_decompress_req req;
+ char *raw_buf;
+ char *out_buf;
+ erofs_off_t out_offset;
+ unsigned int out_length;
+};
+
+struct z_erofs_decompress_task {
+ struct z_erofs_decompress_task *next;
+ struct z_erofs_mt_read_ctx *ctx;
+ struct z_erofs_decompress_item items[Z_EROFS_PCLUSTER_MAX_BATCH_SIZE];
+ unsigned int nr_reqs;
+};
+
+static void z_erofs_process_task(struct z_erofs_decompress_task *task)
+{
+ struct z_erofs_mt_read_ctx *ctx = task->ctx;
+ int i, ret = 0, first_err = 0;
+
+ for (i = 0; i < task->nr_reqs; ++i) {
+ struct z_erofs_decompress_item *item = &task->items[i];
+ ret = z_erofs_decompress(&item->req);
+
+ if (ret >= 0 && ctx && ctx->outfd >= 0) {
+ if (pwrite(ctx->outfd, item->out_buf,
+ item->out_length, item->out_offset) < 0)
+ ret = -errno;
+ }
+
+ if (ret < 0 && !first_err)
+ first_err = ret;
+
+ free(item->raw_buf);
+ if (ctx && ctx->free_out)
+ free(item->out_buf);
+ }
+
+ if (ctx) {
+ erofs_mutex_lock(&ctx->lock);
+ if (first_err < 0 && !ctx->final_err)
+ ctx->final_err = first_err;
+ ctx->pending_tasks--;
+ if (!ctx->pending_tasks)
+ erofs_cond_signal(&ctx->cond);
+ erofs_mutex_unlock(&ctx->lock);
+ }
+ free(task);
+}
+
+#ifdef EROFS_MT_ENABLED
+static void *z_erofs_worker_thread(void *arg)
+{
+ struct z_erofs_percpu_queue *q = arg;
+
+ while (1) {
+ erofs_mutex_lock(&q->lock);
+ while (!q->head && !q->shutdown)
+ erofs_cond_wait(&q->cond, &q->lock);
+
+ if (q->shutdown && !q->head) {
+ erofs_mutex_unlock(&q->lock);
+ break;
+ }
+
+ struct z_erofs_decompress_task *tasks = q->head;
+ q->head = q->tail = NULL;
+ q->pending_tasks = 0;
+ erofs_mutex_unlock(&q->lock);
+
+ while (tasks) {
+ struct z_erofs_decompress_task *task = tasks;
+ tasks = tasks->next;
+ z_erofs_process_task(task);
+ }
+ }
+ return NULL;
+}
+
+int z_erofs_mt_workers_init(void)
+{
+ int i;
+ num_workers = erofs_get_available_processors();
+ if (num_workers < 1)
+ num_workers = 1;
+
+ worker_queues = calloc(num_workers, sizeof(*worker_queues));
+ if (!worker_queues)
+ return -ENOMEM;
+
+ for (i = 0; i < num_workers; ++i) {
+ erofs_mutex_init(&worker_queues[i].lock);
+ erofs_cond_init(&worker_queues[i].cond);
+ worker_queues[i].head = worker_queues[i].tail = NULL;
+ worker_queues[i].pending_tasks = 0;
+ worker_queues[i].shutdown = false;
+ pthread_create(&worker_queues[i].thread, NULL, z_erofs_worker_thread, &worker_queues[i]);
+ }
+ return 0;
+}
+
+void z_erofs_mt_workers_exit(void)
+{
+ int i;
+ if (!worker_queues) return;
+ for (i = 0; i < num_workers; ++i) {
+ erofs_mutex_lock(&worker_queues[i].lock);
+ worker_queues[i].shutdown = true;
+ erofs_cond_signal(&worker_queues[i].cond);
+ erofs_mutex_unlock(&worker_queues[i].lock);
+
+ pthread_join(worker_queues[i].thread, NULL);
+ erofs_cond_destroy(&worker_queues[i].cond);
+ erofs_mutex_destroy(&worker_queues[i].lock);
+ }
+ free(worker_queues);
+ worker_queues = NULL;
+}
+#endif
+
+void z_erofs_mt_read_enqueue(struct z_erofs_mt_read_ctx *ctx)
+{
+#ifdef EROFS_MT_ENABLED
+ static int next_worker = 0;
+#endif
+
+ if (!ctx || !ctx->current_task)
+ return;
+
+ struct z_erofs_decompress_task *task = ctx->current_task;
+ ctx->current_task = NULL;
+
+#ifdef EROFS_MT_ENABLED
+ if (num_workers > 0) {
+ int target = next_worker;
+ next_worker = (next_worker + 1) % num_workers;
+ struct z_erofs_percpu_queue *q = &worker_queues[target];
+
+ erofs_mutex_lock(&q->lock);
+ task->next = NULL;
+ if (!q->tail) {
+ q->head = q->tail = task;
+ } else {
+ q->tail->next = task;
+ q->tail = task;
+ }
+ q->pending_tasks++;
+ erofs_cond_signal(&q->cond);
+ erofs_mutex_unlock(&q->lock);
+ return;
+ }
+#endif
+ task->next = NULL;
+ z_erofs_process_task(task);
+}
void *erofs_bread(struct erofs_buf *buf, erofs_off_t offset, bool need_kmap)
{
@@ -277,20 +499,29 @@ static int erofs_read_raw_data(struct erofs_inode *inode, char *buffer,
int z_erofs_read_one_data(struct erofs_inode *inode,
struct erofs_map_blocks *map, char *raw, char *buffer,
- erofs_off_t skip, erofs_off_t length, bool trimmed)
+ erofs_off_t skip, erofs_off_t length, bool trimmed,
+ erofs_off_t out_offset, struct z_erofs_mt_read_ctx *ctx)
{
struct erofs_sb_info *sbi = inode->sbi;
struct erofs_map_dev mdev;
- int ret = 0;
+ struct z_erofs_decompress_task *task;
+ struct z_erofs_decompress_item *item;
+ int ret = 0, idx, batch_limit;
if (map->m_flags & __EROFS_MAP_FRAGMENT) {
if (__erofs_unlikely(inode->nid == sbi->packed_nid)) {
erofs_err("fragment should not exist in the packed inode %llu",
- sbi->packed_nid | 0ULL);
- return -EFSCORRUPTED;
+ sbi->packed_nid | 0ULL);
+ ret = -EFSCORRUPTED;
+ goto err_out;
+ }
+ ret = erofs_packedfile_read(sbi, buffer, length - skip,
+ inode->fragmentoff + skip);
+ if (ret >= 0 && ctx && ctx->outfd >= 0) {
+ if (pwrite(ctx->outfd, buffer, length - skip, out_offset) < 0)
+ ret = -errno;
}
- return erofs_packedfile_read(sbi, buffer, length - skip,
- inode->fragmentoff + skip);
+ goto err_out;
}
/* no device id here, thus it will always succeed */
@@ -300,31 +531,86 @@ int z_erofs_read_one_data(struct erofs_inode *inode,
ret = erofs_map_dev(sbi, &mdev);
if (ret) {
DBG_BUGON(1);
- return ret;
+ goto err_out;
}
ret = erofs_dev_read(sbi, mdev.m_deviceid, raw, mdev.m_pa, map->m_plen);
if (ret < 0)
+ goto err_out;
+
+ if (!ctx) {
+ ret = z_erofs_decompress(&(struct z_erofs_decompress_req) {
+ .sbi = sbi,
+ .in = raw,
+ .out = buffer,
+ .decodedskip = skip,
+ .interlaced_offset =
+ map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
+ erofs_blkoff(sbi, map->m_la) : 0,
+ .inputsize = map->m_plen,
+ .decodedlength = length,
+ .alg = map->m_algorithmformat,
+ .partial_decoding = trimmed ? true :
+ !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
+ (map->m_flags & EROFS_MAP_PARTIAL_REF),
+ });
return ret;
+ }
- ret = z_erofs_decompress(&(struct z_erofs_decompress_req) {
- .sbi = sbi,
- .in = raw,
- .out = buffer,
- .decodedskip = skip,
- .interlaced_offset =
- map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
- erofs_blkoff(sbi, map->m_la) : 0,
- .inputsize = map->m_plen,
- .decodedlength = length,
- .alg = map->m_algorithmformat,
- .partial_decoding = trimmed ? true :
- !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
- (map->m_flags & EROFS_MAP_PARTIAL_REF),
- });
- if (ret < 0)
- return ret;
+ task = ctx->current_task;
+ if (!task) {
+ task = malloc(sizeof(*task));
+ if (!task) {
+ ret = -ENOMEM;
+ goto err_out;
+ }
+ task->nr_reqs = 0;
+ task->ctx = ctx;
+ ctx->current_task = task;
+
+ erofs_mutex_lock(&ctx->lock);
+ ctx->pending_tasks++;
+ erofs_mutex_unlock(&ctx->lock);
+ }
+
+ idx = task->nr_reqs++;
+ item = &task->items[idx];
+
+ item->req = (struct z_erofs_decompress_req) {
+ .sbi = sbi,
+ .in = raw,
+ .out = buffer,
+ .decodedskip = skip,
+ .interlaced_offset =
+ map->m_algorithmformat == Z_EROFS_COMPRESSION_INTERLACED ?
+ erofs_blkoff(sbi, map->m_la) : 0,
+ .inputsize = map->m_plen,
+ .decodedlength = length,
+ .alg = map->m_algorithmformat,
+ .partial_decoding = trimmed ? true :
+ !(map->m_flags & EROFS_MAP_FULL_MAPPED) ||
+ (map->m_flags & EROFS_MAP_PARTIAL_REF),
+ };
+ item->raw_buf = raw;
+ item->out_buf = buffer;
+ item->out_offset = out_offset;
+ item->out_length = length;
+
+ batch_limit = (map->m_algorithmformat == Z_EROFS_COMPRESSION_LZ4 ||
+ map->m_algorithmformat == Z_EROFS_COMPRESSION_LZ4) ?
+ Z_EROFS_PCLUSTER_MAX_BATCH_SIZE : 8;
+
+ if (task->nr_reqs >= batch_limit)
+ z_erofs_mt_read_enqueue(ctx);
return 0;
+
+err_out:
+ if (ctx) {
+ if (ctx->free_out)
+ free(buffer);
+ free(raw);
+ }
+ return ret;
}
static int z_erofs_read_data(struct erofs_inode *inode, char *buffer,
@@ -387,7 +673,7 @@ static int z_erofs_read_data(struct erofs_inode *inode, char *buffer,
}
ret = z_erofs_read_one_data(inode, &map, raw,
- buffer + end - offset, skip, length, trimmed);
+ buffer + end - offset, skip, length, trimmed, 0, NULL);
if (ret < 0)
break;
}
diff --git a/lib/global.c b/lib/global.c
index 938aa0a..8bb9da0 100644
--- a/lib/global.c
+++ b/lib/global.c
@@ -12,6 +12,7 @@
#include "erofs/err.h"
#include "erofs/config.h"
#include "liberofs_compress.h"
+#include "erofs/internal.h"
static EROFS_DEFINE_MUTEX(erofs_global_mutex);
#ifdef HAVE_LIBCURL
@@ -27,6 +28,11 @@ int liberofs_global_init(void)
#ifdef S3EROFS_ENABLED
xmlInitParser();
#endif
+#ifdef EROFS_MT_ENABLED
+ err = z_erofs_mt_workers_init();
+ if (err)
+ goto out_unlock;
+#endif
#ifdef HAVE_LIBCURL
if (!erofs_global_curl_initialized) {
if (curl_global_init(CURL_GLOBAL_DEFAULT) != CURLE_OK) {
@@ -35,8 +41,8 @@ int liberofs_global_init(void)
}
erofs_global_curl_initialized = true;
}
-out_unlock:
#endif
+out_unlock:
erofs_mutex_unlock(&erofs_global_mutex);
return err;
}
@@ -45,6 +51,9 @@ void liberofs_global_exit(void)
{
erofs_mutex_lock(&erofs_global_mutex);
z_erofs_mt_global_exit();
+#ifdef EROFS_MT_ENABLED
+ z_erofs_mt_workers_exit();
+#endif
#ifdef HAVE_LIBCURL
if (erofs_global_curl_initialized) {
curl_global_cleanup();
--
2.53.0
^ permalink raw reply related [flat|nested] 13+ messages in thread
* Re: [PATCH v5] fsck.erofs: add multi-threaded decompression
2026-07-14 1:10 ` [PATCH v5] fsck.erofs: add multi-threaded decompression Nithurshen
@ 2026-07-14 1:18 ` Nithurshen Karthikeyan
0 siblings, 0 replies; 13+ messages in thread
From: Nithurshen Karthikeyan @ 2026-07-14 1:18 UTC (permalink / raw)
To: nithurshen.dev; +Cc: hsiangkao, linux-erofs, xiang
Hi Xiang,
| Algorithm | 4k | 8k | 16k | 32k | 64k |
| --- | --- | --- | --- | --- | --- |
| lz4hc(MTv5) | 6.44 | 6.71 | 5.64 | 5.18 | 4.71 |
| lz4hc(MTv3) | 9.36 | 8.31 | 8.27 | 8.54 | 6.94 |
| lz4hc(ST) | 4.40 | 5.77 | 3.65 | 5.45 | 3.36 |
| zstd(MTv5) | 6.85 | 6.47 | 5.99 | 6.11 | 6.16 |
| zstd(MTv3) | 9.26 | 8.68 | 8.89 | 8.11 | 7.94 |
| zstd(ST) | 5.23 | 4.52 | 4.62 | 4.11 | 4.02 |
| lzma(MTv5) | 21.92 | 23.43 | 23.82 | 24.92 | 26.74 |
| lzma(MTv3) | 23.72 | 24.79 | 25.90 | 26.92 | 27.77 |
| lzma(ST) | 56.37 | 65.06 | 71.07 | 74.69 | 81.63 |
Please find the new benchmark times for this patch.
I profiled the extraction process, and found out there was some
bottleneck for using `calloc` calls. I have switched them to
`malloc`.
And with some optimisation, I am seeing a ~2 sec improvement
from our previous baseline.
I will share profiling data shortly.
Please let me know your thoughts on this.
Thanks,
Nithurshen
^ permalink raw reply [flat|nested] 13+ messages in thread
end of thread, other threads:[~2026-07-14 1:18 UTC | newest]
Thread overview: 13+ messages (download: mbox.gz follow: Atom feed
-- links below jump to the message on this page --
2026-06-21 12:01 [PATCH 0/2] fsck.erofs: implement multi-threaded extraction Nithurshen
2026-06-21 12:01 ` [PATCH 1/2] fsck.erofs: add multi-threaded decompression Nithurshen
2026-06-22 2:08 ` Gao Xiang
2026-06-29 9:55 ` [PATCH 1/2 v2] " Nithurshen
2026-07-05 14:00 ` Gao Xiang
2026-07-06 6:05 ` [PATCH 1/2 v3] " Nithurshen
2026-07-06 6:10 ` Nithurshen
2026-07-08 2:31 ` Gao Xiang
2026-07-13 7:02 ` [PATCH 1/2 v4] " Nithurshen
2026-06-21 12:01 ` [PATCH 2/2] fsck.erofs: implement concurrent directory traversal Nithurshen
2026-07-05 14:05 ` Gao Xiang
2026-07-14 1:10 ` [PATCH v5] fsck.erofs: add multi-threaded decompression Nithurshen
2026-07-14 1:18 ` Nithurshen Karthikeyan
This is an external index of several public inboxes,
see mirroring instructions on how to clone and mirror
all data and code used by this external index.