Skip to content

Commit cc225b4

Browse files
SpringMTSpringMT
authored
Merge pull request #79 from SpringMT/feature/unlock-gvl
Feature/unlock gvl for streaming compression/decompression
2 parents 788f4f5 + 04d74fc commit cc225b4

13 files changed

+142
-58
lines changed

‎.github/workflows/ruby.yml‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,11 +7,20 @@
77

88
name: Ruby
99

10-
on: [push, pull_request]
10+
on:
11+
push:
12+
branches:
13+
- main
14+
paths-ignore:
15+
- 'README.md'
16+
pull_request:
17+
18+
concurrency:
19+
group: ${{ github.workflow }}-${{ github.ref }}
20+
cancel-in-progress: true
1121

1222
jobs:
1323
test:
14-
1524
runs-on: ubuntu-latest
1625
strategy:
1726
matrix:

‎benchmarks/multi_thread_comporess.rb‎ renamed to ‎benchmarks/multi_thread_streaming_comporess.rb‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,4 @@
1-
require 'benchmark/ips'
21
$LOAD_PATH.unshift '../lib'
3-
require 'json'
4-
require 'objspace'
52
require 'zstd-ruby'
63
require 'thread'
74

@@ -19,7 +16,11 @@
1916
THREADS.times.map {
2017
Thread.new {
2118
while str = queue.pop
22-
Zstd.compress(str)
19+
stream = Zstd::StreamingCompress.new
20+
stream << str
21+
res = stream.flush
22+
stream << str
23+
res << stream.finish
2324
end
2425
}
2526
}.each(&:join)

‎benchmarks/multi_thread_streaming_decomporess.rb‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
$LOAD_PATH.unshift '../lib'
2+
require 'zstd-ruby'
3+
require 'thread'
4+
5+
GUESSES = (ENV['GUESSES'] || 1000).to_i
6+
THREADS = (ENV['THREADS'] || 1).to_i
7+
8+
p GUESSES: GUESSES, THREADS: THREADS
9+
10+
sample_file_name = ARGV[0]
11+
json_string = File.read("./samples/#{sample_file_name}")
12+
target = Zstd.compress(json_string)
13+
14+
queue = Queue.new
15+
GUESSES.times { queue << target }
16+
THREADS.times { queue << nil }
17+
THREADS.times.map {
18+
Thread.new {
19+
while str = queue.pop
20+
stream = Zstd::StreamingDecompress.new
21+
stream.decompress(str)
22+
stream.decompress(str)
23+
end
24+
}
25+
}.each(&:join)

‎benchmarks/zstd_compress_memory.rb‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010

1111
sample_file_name = ARGV[0]
1212

13-
json_data = JSON.parse(IO.read("./samples/#{sample_file_name}"), symbolize_names: true)
13+
json_data = JSON.parse(File.read("./samples/#{sample_file_name}"), symbolize_names: true)
1414
json_string = json_data.to_json
1515

1616
i = 0

‎benchmarks/zstd_decompress_memory.rb‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@
99
p "#{ObjectSpace.memsize_of_all/1000} #{ObjectSpace.count_objects} #{`ps -o rss= -p #{Process.pid}`.to_i}"
1010

1111
sample_file_name = ARGV[0]
12-
json_data = JSON.parse(IO.read("./samples/#{sample_file_name}"), symbolize_names: true)
12+
json_data = JSON.parse(File.read("./samples/#{sample_file_name}"), symbolize_names: true)
1313
json_string = json_data.to_json
1414

1515
i = 0

‎benchmarks/zstd_streaming_compress_memory.rb‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010

1111
sample_file_name = ARGV[0]
1212

13-
json_string = IO.read("./samples/#{sample_file_name}")
13+
json_string = File.read("./samples/#{sample_file_name}")
1414

1515
i = 0
1616
start_time = Time.now

‎benchmarks/zstd_streaming_decompress_memory.rb‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010

1111
sample_file_name = ARGV[0]
1212

13-
cstr = IO.read("./results/#{sample_file_name}.zstd")
13+
cstr = File.read("./results/#{sample_file_name}.zstd")
1414
i = 0
1515
start_time = Time.now
1616
while true do

‎examples/sinatra/Gemfile.lock‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
PATH
22
remote: ../..
33
specs:
4-
zstd-ruby (1.5.6.1)
4+
zstd-ruby (1.5.6.2)
55

66
GEM
77
remote: https://rubygems.org/

‎ext/zstdruby/common.h‎

Lines changed: 63 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,10 @@
22
#define ZSTD_RUBY_H 1
33

44
#include <ruby.h>
5+
#ifdef HAVE_RUBY_THREAD_H
6+
#include <ruby/thread.h>
7+
#endif
8+
#include <stdbool.h>
59
#include "./libzstd/zstd.h"
610

711
static int convert_compression_level(VALUE compression_level_value)
@@ -12,11 +16,6 @@ static int convert_compression_level(VALUE compression_level_value)
1216
return NUM2INT(compression_level_value);
1317
}
1418

15-
static size_t zstd_compress(ZSTD_CCtx* const ctx, ZSTD_outBuffer* output, ZSTD_inBuffer* input, ZSTD_EndDirective endOp)
16-
{
17-
return ZSTD_compressStream2(ctx, output, input, endOp);
18-
}
19-
2019
static void set_compress_params(ZSTD_CCtx* const ctx, VALUE level_from_args, VALUE kwargs)
2120
{
2221
ID kwargs_keys[2];
@@ -45,6 +44,36 @@ static void set_compress_params(ZSTD_CCtx* const ctx, VALUE level_from_args, VAL
4544
}
4645
}
4746

47+
struct compress_params {
48+
ZSTD_CCtx* ctx;
49+
ZSTD_outBuffer* output;
50+
ZSTD_inBuffer* input;
51+
ZSTD_EndDirective endOp;
52+
size_t ret;
53+
};
54+
55+
static void* compress_wrapper(void* args)
56+
{
57+
struct compress_params* params = args;
58+
params->ret = ZSTD_compressStream2(params->ctx, params->output, params->input, params->endOp);
59+
return NULL;
60+
}
61+
62+
static size_t zstd_compress(ZSTD_CCtx* const ctx, ZSTD_outBuffer* output, ZSTD_inBuffer* input, ZSTD_EndDirective endOp, bool gvl)
63+
{
64+
#ifdef HAVE_RUBY_THREAD_H
65+
if (gvl) {
66+
return ZSTD_compressStream2(ctx, output, input, endOp);
67+
} else {
68+
struct compress_params params = { ctx, output, input, endOp };
69+
rb_thread_call_without_gvl(compress_wrapper, &params, NULL, NULL);
70+
return params.ret;
71+
}
72+
#else
73+
return ZSTD_compressStream2(ctx, output, input, endOp);
74+
#endif
75+
}
76+
4877
static void set_decompress_params(ZSTD_DCtx* const dctx, VALUE kwargs)
4978
{
5079
ID kwargs_keys[1];
@@ -63,4 +92,33 @@ static void set_decompress_params(ZSTD_DCtx* const dctx, VALUE kwargs)
6392
}
6493
}
6594

95+
struct decompress_params {
96+
ZSTD_DCtx* dctx;
97+
ZSTD_outBuffer* output;
98+
ZSTD_inBuffer* input;
99+
size_t ret;
100+
};
101+
102+
static void* decompress_wrapper(void* args)
103+
{
104+
struct decompress_params* params = args;
105+
params->ret = ZSTD_decompressStream(params->dctx, params->output, params->input);
106+
return NULL;
107+
}
108+
109+
static size_t zstd_decompress(ZSTD_DCtx* const dctx, ZSTD_outBuffer* output, ZSTD_inBuffer* input, bool gvl)
110+
{
111+
#ifdef HAVE_RUBY_THREAD_H
112+
if (gvl) {
113+
return ZSTD_decompressStream(dctx, output, input);
114+
} else {
115+
struct decompress_params params = { dctx, output, input };
116+
rb_thread_call_without_gvl(decompress_wrapper, &params, NULL, NULL);
117+
return params.ret;
118+
}
119+
#else
120+
return ZSTD_decompressStream(dctx, output, input);
121+
#endif
122+
}
123+
66124
#endif /* ZSTD_RUBY_H */

‎ext/zstdruby/extconf.rb‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
have_func('rb_gc_mark_movable')
44

5-
$CFLAGS = '-I. -O3 -std=c99 -DZSTD_STATIC_LINKING_ONLY'
5+
$CFLAGS = '-I. -O3 -std=c99 -DZSTD_STATIC_LINKING_ONLY -DZSTD_MULTITHREAD -pthread -DDEBUGLEVEL=0'
66
$CPPFLAGS += " -fdeclspec" if CONFIG['CXX'] =~ /clang/
77

88
Dir.chdir File.expand_path('..', __FILE__) do

0 commit comments

Comments
 (0)