unlock the GVL for compression/decompression operations

This allows to run them in parallel (i.e. multi-threaded)

Author: doudou

ext/zstdruby/streaming_compress.c

@@ -1,5 +1,6 @@
 #include <common.h>
 #include <streaming_compress.h>
+#include <ruby/thread.h>
 
 struct streaming_compress_t {
   ZSTD_CCtx* ctx;
@@ -76,6 +77,30 @@ rb_streaming_compress_initialize(int argc, VALUE *argv, VALUE obj)
     : (FIX2INT((val))))
 #define ARG_CONTINUE(val)     FIXNUMARG((val), ZSTD_e_continue)
 
+struct compress_stream_nogvl_t {
+  ZSTD_CCtx* ctx;
+  ZSTD_outBuffer* output;
+  ZSTD_inBuffer* input;
+  ZSTD_EndDirective endOp;
+  size_t ret;
+};
+
+static void*
+compressStream2_nogvl(void* arg)
+{
+  struct compress_stream_nogvl_t* params = arg;
+  params->ret = ZSTD_compressStream2(params->ctx, params->output, params->input, params->endOp);
+  return NULL;
+}
+
+static size_t
+compressStream2(ZSTD_CCtx* ctx, ZSTD_outBuffer* output, ZSTD_inBuffer* input, ZSTD_EndDirective endOp)
+{
+  struct compress_stream_nogvl_t params = { ctx, output, input, endOp, 0 };
+  rb_thread_call_without_gvl(compressStream2_nogvl, &params, NULL, NULL);
+  return params.ret;
+}
+
 static VALUE
 no_compress(struct streaming_compress_t* sc, ZSTD_EndDirective endOp)
 {
@@ -86,7 +111,7 @@ no_compress(struct streaming_compress_t* sc, ZSTD_EndDirective endOp)
   do {
     ZSTD_outBuffer output = { (void*)output_data, sc->buf_size, 0 };
 
-    size_t const ret = ZSTD_compressStream2(sc->ctx, &output, &input, endOp);
+    size_t const ret = compressStream2(sc->ctx, &output, &input, endOp);
     if (ZSTD_isError(ret)) {
       rb_raise(rb_eRuntimeError, "flush error error code: %s", ZSTD_getErrorName(ret));
     }
@@ -109,7 +134,7 @@ rb_streaming_compress_compress(VALUE obj, VALUE src)
   VALUE result = rb_str_new(0, 0);
   while (input.pos < input.size) {
     ZSTD_outBuffer output = { (void*)output_data, sc->buf_size, 0 };
-    size_t const ret = ZSTD_compressStream2(sc->ctx, &output, &input, ZSTD_e_continue);
+    size_t const ret = compressStream2(sc->ctx, &output, &input, ZSTD_e_continue);
     if (ZSTD_isError(ret)) {
       rb_raise(rb_eRuntimeError, "compress error error code: %s", ZSTD_getErrorName(ret));
     }
@@ -132,7 +157,7 @@ rb_streaming_compress_addstr(VALUE obj, VALUE src)
 
   while (input.pos < input.size) {
     ZSTD_outBuffer output = { (void*)output_data, sc->buf_size, 0 };
-    size_t const result = ZSTD_compressStream2(sc->ctx, &output, &input, ZSTD_e_continue);
+    size_t const result = compressStream2(sc->ctx, &output, &input, ZSTD_e_continue);
     if (ZSTD_isError(result)) {
       rb_raise(rb_eRuntimeError, "compress error error code: %s", ZSTD_getErrorName(result));
     }

ext/zstdruby/streaming_decompress.c

@@ -1,4 +1,5 @@
 #include <common.h>
+#include <ruby/thread.h>
 
 struct streaming_decompress_t {
   ZSTD_DCtx* ctx;
@@ -65,6 +66,29 @@ rb_streaming_decompress_initialize(VALUE obj)
   return obj;
 }
 
+struct decompress_stream_nogvl_t {
+  ZSTD_DCtx* ctx;
+  ZSTD_outBuffer* output;
+  ZSTD_inBuffer* input;
+  size_t ret;
+};
+
+static void*
+decompressStream_nogvl(void* args)
+{
+  struct decompress_stream_nogvl_t* params = args;
+  params->ret = ZSTD_decompressStream(params->ctx, params->output, params->input);
+  return NULL;
+}
+
+static size_t
+decompressStream(ZSTD_DCtx* ctx, ZSTD_outBuffer* output, ZSTD_inBuffer* input)
+{
+  struct decompress_stream_nogvl_t params = { ctx, output, input, 0 };
+  rb_thread_call_without_gvl(decompressStream_nogvl, &params, NULL, NULL);
+  return params.ret;
+}
+
 static VALUE
 rb_streaming_decompress_decompress(VALUE obj, VALUE src)
 {
@@ -79,7 +103,7 @@ rb_streaming_decompress_decompress(VALUE obj, VALUE src)
   VALUE result = rb_str_new(0, 0);
   while (input.pos < input.size) {
     ZSTD_outBuffer output = { (void*)output_data, sd->buf_size, 0 };
-    size_t const ret = ZSTD_decompressStream(sd->ctx, &output, &input);
+    size_t const ret = decompressStream(sd->ctx, &output, &input);
     if (ZSTD_isError(ret)) {
       rb_raise(rb_eRuntimeError, "compress error error code: %s", ZSTD_getErrorName(ret));
     }
@@ -102,7 +126,7 @@ rb_streaming_decompress_addstr(VALUE obj, VALUE src)
 
   while (input.pos < input.size) {
     ZSTD_outBuffer output = { (void*)output_data, sd->buf_size, 0 };
-    size_t const result = ZSTD_decompressStream(sd->ctx, &output, &input);
+    size_t const result = decompressStream(sd->ctx, &output, &input);
     if (ZSTD_isError(result)) {
       rb_raise(rb_eRuntimeError, "compress error error code: %s", ZSTD_getErrorName(result));
     }