erofs-utils: mkfs: enable inter-file multi-threaded compression

Dispatch deferred ops in another per-sb worker thread.  Note that
deferred ops are strictly FIFOed.

Signed-off-by: Gao Xiang <hsiangkao@linux.alibaba.com>
Link: https://lore.kernel.org/r/20240422003450.19132-8-xiang@kernel.org

diff --git a/include/erofs/internal.h b/include/erofs/internal.h
index f31e548bd1d6af152ed0f885d0cbffb4bd3e12a5..ecbbdf6a659a774f920ea567751aa6f122b80bdd 100644 (file)
--- a/include/erofs/internal.h
+++ b/include/erofs/internal.h
@@ -71,6 +71,7 @@ struct erofs_xattr_prefix_item {
 
 #define EROFS_PACKED_NID_UNALLOCATED   -1
 
+struct erofs_mkfs_dfops;
 struct erofs_sb_info {
        struct erofs_device_info *devs;
        char *devname;
@@ -124,6 +125,11 @@ struct erofs_sb_info {
        struct list_head list;
 
        u64 saved_by_deduplication;
+
+#ifdef EROFS_MT_ENABLED
+       pthread_t dfops_worker;
+       struct erofs_mkfs_dfops *mkfs_dfops;
+#endif
 };
 
 /* make sure that any user of the erofs headers has atleast 64bit off_t type */

diff --git a/lib/inode.c b/lib/inode.c
index 415490d98d801dd10cb17a1dbf235a8971924730..44d684f9678a64220050fbfc458c1b63189fedeb 100644 (file)
--- a/lib/inode.c
+++ b/lib/inode.c
@@ -1167,6 +1167,7 @@ enum erofs_mkfs_jobtype { /* ordered job types */
        EROFS_MKFS_JOB_NDIR,
        EROFS_MKFS_JOB_DIR,
        EROFS_MKFS_JOB_DIR_BH,
+       EROFS_MKFS_JOB_MAX
 };
 
 struct erofs_mkfs_jobitem {
@@ -1205,6 +1206,73 @@ static int erofs_mkfs_jobfn(struct erofs_mkfs_jobitem *item)
        return -EINVAL;
 }
 
+#ifdef EROFS_MT_ENABLED
+
+struct erofs_mkfs_dfops {
+       pthread_t worker;
+       pthread_mutex_t lock;
+       pthread_cond_t full, empty;
+       struct erofs_mkfs_jobitem *queue;
+       unsigned int entries, head, tail;
+};
+
+#define EROFS_MT_QUEUE_SIZE 128
+
+void *erofs_mkfs_pop_jobitem(struct erofs_mkfs_dfops *q)
+{
+       struct erofs_mkfs_jobitem *item;
+
+       pthread_mutex_lock(&q->lock);
+       while (q->head == q->tail)
+               pthread_cond_wait(&q->empty, &q->lock);
+
+       item = q->queue + q->head;
+       q->head = (q->head + 1) & (q->entries - 1);
+
+       pthread_cond_signal(&q->full);
+       pthread_mutex_unlock(&q->lock);
+       return item;
+}
+
+void *z_erofs_mt_dfops_worker(void *arg)
+{
+       struct erofs_sb_info *sbi = arg;
+       int ret = 0;
+
+       while (1) {
+               struct erofs_mkfs_jobitem *item;
+
+               item = erofs_mkfs_pop_jobitem(sbi->mkfs_dfops);
+               if (item->type >= EROFS_MKFS_JOB_MAX)
+                       break;
+               ret = erofs_mkfs_jobfn(item);
+               if (ret)
+                       break;
+       }
+       pthread_exit((void *)(uintptr_t)ret);
+}
+
+int erofs_mkfs_go(struct erofs_sb_info *sbi,
+                 enum erofs_mkfs_jobtype type, void *elem, int size)
+{
+       struct erofs_mkfs_jobitem *item;
+       struct erofs_mkfs_dfops *q = sbi->mkfs_dfops;
+
+       pthread_mutex_lock(&q->lock);
+
+       while (((q->tail + 1) & (q->entries - 1)) == q->head)
+               pthread_cond_wait(&q->full, &q->lock);
+
+       item = q->queue + q->tail;
+       item->type = type;
+       memcpy(&item->u, elem, size);
+       q->tail = (q->tail + 1) & (q->entries - 1);
+
+       pthread_cond_signal(&q->empty);
+       pthread_mutex_unlock(&q->lock);
+       return 0;
+}
+#else
 int erofs_mkfs_go(struct erofs_sb_info *sbi,
                  enum erofs_mkfs_jobtype type, void *elem, int size)
 {
@@ -1214,6 +1282,7 @@ int erofs_mkfs_go(struct erofs_sb_info *sbi,
        memcpy(&item.u, elem, size);
        return erofs_mkfs_jobfn(&item);
 }
+#endif
 
 static int erofs_mkfs_handle_directory(struct erofs_inode *dir)
 {
@@ -1346,7 +1415,11 @@ static int erofs_mkfs_handle_inode(struct erofs_inode *inode)
        return ret;
 }
 
-struct erofs_inode *erofs_mkfs_build_tree_from_path(const char *path)
+#ifndef EROFS_MT_ENABLED
+#define __erofs_mkfs_build_tree_from_path erofs_mkfs_build_tree_from_path
+#endif
+
+struct erofs_inode *__erofs_mkfs_build_tree_from_path(const char *path)
 {
        struct erofs_inode *root, *dumpdir;
        int err;
@@ -1405,6 +1478,49 @@ struct erofs_inode *erofs_mkfs_build_tree_from_path(const char *path)
        return root;
 }
 
+#ifdef EROFS_MT_ENABLED
+struct erofs_inode *erofs_mkfs_build_tree_from_path(const char *path)
+{
+       struct erofs_mkfs_dfops *q;
+       struct erofs_inode *root;
+       int err;
+
+       q = malloc(sizeof(*q));
+       if (!q)
+               return ERR_PTR(-ENOMEM);
+
+       q->entries = EROFS_MT_QUEUE_SIZE;
+       q->queue = malloc(q->entries * sizeof(*q->queue));
+       if (!q->queue) {
+               free(q);
+               return ERR_PTR(-ENOMEM);
+       }
+       pthread_mutex_init(&q->lock, NULL);
+       pthread_cond_init(&q->empty, NULL);
+       pthread_cond_init(&q->full, NULL);
+
+       q->head = 0;
+       q->tail = 0;
+       sbi.mkfs_dfops = q;
+       err = pthread_create(&sbi.dfops_worker, NULL,
+                            z_erofs_mt_dfops_worker, &sbi);
+       if (err)
+               goto fail;
+       root = __erofs_mkfs_build_tree_from_path(path);
+
+       erofs_mkfs_go(&sbi, ~0, NULL, 0);
+       err = pthread_join(sbi.dfops_worker, NULL);
+
+fail:
+       pthread_cond_destroy(&q->empty);
+       pthread_cond_destroy(&q->full);
+       pthread_mutex_destroy(&q->lock);
+       free(q->queue);
+       free(q);
+       return err ? ERR_PTR(err) : root;
+}
+#endif
+
 struct erofs_inode *erofs_mkfs_build_special_from_fd(int fd, const char *name)
 {
        struct stat st;