Async write for decompression (#2975)

* Async IO decompression:
- Added --[no-]asyncio flag for CLI decompression.
- Replaced dstBuffer in decompression with a pool of write jobs.
- Added an ability to execute write jobs in a separate thread.
- Added an ability to wait (join) on all jobs in a thread pool (queued and running).

Author: yoniko

build/meson/programs/meson.build

@@ -20,14 +20,24 @@ zstd_programs_sources = [join_paths(zstd_rootdir, 'programs/zstdcli.c'),
   join_paths(zstd_rootdir, 'programs/dibio.c'),
   join_paths(zstd_rootdir, 'programs/zstdcli_trace.c'),
   # needed due to use of private symbol + -fvisibility=hidden
-  join_paths(zstd_rootdir, 'lib/common/xxhash.c')]
+  join_paths(zstd_rootdir, 'lib/common/xxhash.c'),
+  join_paths(zstd_rootdir, 'lib/common/pool.c'),
+  join_paths(zstd_rootdir, 'lib/common/zstd_common.c'),
+  join_paths(zstd_rootdir, 'lib/common/error_private.c')]
 
+zstd_deps = [ libzstd_dep ]
 zstd_c_args = libzstd_debug_cflags
+
+zstd_frugal_deps = [ libzstd_dep ]
+zstd_frugal_c_args = [ '-DZSTD_NOBENCH', '-DZSTD_NODICT', '-DZSTD_NOTRACE' ]
+
 if use_multi_thread
+  zstd_deps += [ thread_dep ]
   zstd_c_args += [ '-DZSTD_MULTITHREAD' ]
+  zstd_frugal_deps += [ thread_dep ]
+  zstd_frugal_c_args += [ '-DZSTD_MULTITHREAD' ]
 endif
 
-zstd_deps = [ libzstd_dep ]
 if use_zlib
   zstd_deps += [ zlib_dep ]
   zstd_c_args += [ '-DZSTD_GZCOMPRESS', '-DZSTD_GZDECOMPRESS' ]
@@ -69,14 +79,17 @@ zstd = executable('zstd',
 zstd_frugal_sources = [join_paths(zstd_rootdir, 'programs/zstdcli.c'),
   join_paths(zstd_rootdir, 'programs/timefn.c'),
   join_paths(zstd_rootdir, 'programs/util.c'),
-  join_paths(zstd_rootdir, 'programs/fileio.c')]
+  join_paths(zstd_rootdir, 'programs/fileio.c'),
+  join_paths(zstd_rootdir, 'lib/common/pool.c'),
+  join_paths(zstd_rootdir, 'lib/common/zstd_common.c'),
+  join_paths(zstd_rootdir, 'lib/common/error_private.c')]
 
 # Minimal target, with only zstd compression and decompression.
 # No bench. No legacy.
 executable('zstd-frugal',
   zstd_frugal_sources,
-  dependencies: libzstd_dep,
-  c_args: [ '-DZSTD_NOBENCH', '-DZSTD_NODICT', '-DZSTD_NOTRACE' ],
+  dependencies: zstd_frugal_deps,
+  c_args: zstd_frugal_c_args,
   install: true)
 
 install_data(join_paths(zstd_rootdir, 'programs/zstdgrep'),

lib/common/pool.c

@@ -96,9 +96,7 @@ static void* POOL_thread(void* opaque) {
             /* If the intended queue size was 0, signal after finishing job */
             ZSTD_pthread_mutex_lock(&ctx->queueMutex);
             ctx->numThreadsBusy--;
-            if (ctx->queueSize == 1) {
-                ZSTD_pthread_cond_signal(&ctx->queuePushCond);
-            }
+            ZSTD_pthread_cond_signal(&ctx->queuePushCond);
             ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
         }
     }  /* for (;;) */
@@ -190,6 +188,17 @@ void POOL_free(POOL_ctx *ctx) {
     ZSTD_customFree(ctx, ctx->customMem);
 }
 
+/*! POOL_joinJobs() :
+ *  Waits for all queued jobs to finish executing.
+ */
+void POOL_joinJobs(POOL_ctx* ctx) {
+    ZSTD_pthread_mutex_lock(&ctx->queueMutex);
+    while(!ctx->queueEmpty || ctx->numThreadsBusy > 0) {
+        ZSTD_pthread_cond_wait(&ctx->queuePushCond, &ctx->queueMutex);
+    }
+    ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
+}
+
 void ZSTD_freeThreadPool (ZSTD_threadPool* pool) {
   POOL_free (pool);
 }
@@ -330,6 +339,11 @@ void POOL_free(POOL_ctx* ctx) {
     (void)ctx;
 }
 
+void POOL_joinJobs(POOL_ctx* ctx){
+    assert(!ctx || ctx == &g_poolCtx);
+    (void)ctx;
+}
+
 int POOL_resize(POOL_ctx* ctx, size_t numThreads) {
     (void)ctx; (void)numThreads;
     return 0;

lib/common/pool.h

@@ -38,6 +38,12 @@ POOL_ctx* POOL_create_advanced(size_t numThreads, size_t queueSize,
  */
 void POOL_free(POOL_ctx* ctx);
 
+
+/*! POOL_joinJobs() :
+ *  Waits for all queued jobs to finish executing.
+ */
+void POOL_joinJobs(POOL_ctx* ctx);
+
 /*! POOL_resize() :
  *  Expands or shrinks pool's number of threads.
  *  This is more efficient than releasing + creating a new context,