Use SizedQueue to limit the number of requests for asynchronous proce…

…ssing
Watson1978 · Mar 7, 2024 · 279a26e · 279a26e
1 parent 7c7ebf2
commit 279a26e
Show file tree

Hide file tree

Showing 5 changed files with 51 additions and 80 deletions.
diff --git a/example/benchmark_insert.rb b/example/benchmark_insert.rb
@@ -63,8 +63,6 @@ def run_execute(x)
   end
 
   def run_execute_async(x)
-    futures = []
-
     x.report('cassandra-driver:execute_async') do
       future = @session.execute_async(
         statement,
@@ -78,16 +76,7 @@ def run_execute_async(x)
       )
       future.on_success do |rows|
       end
-
-      futures << future
-
-      if futures.size > 100
-        tmp = futures.slice!(0..10)
-        Cassandra::Future.all(*tmp).get
-      end
     end
-
-    Cassandra::Future.all(*futures).get
   end
 
   def statement
@@ -116,8 +105,6 @@ def run_execute(x)
   end
 
   def run_execute_async(x)
-    futures = []
-
     x.report('ilios:execute_async') do
       statement.bind(
         {
@@ -129,13 +116,7 @@ def run_execute_async(x)
       future = Ilios::Cassandra.session.execute_async(statement)
       future.on_success do |results|
       end
-
-      futures << future
-
-      futures.slice!(0..10).each(&:await) if futures.size > 100
     end
-
-    futures.each(&:await)
   end
 
   def statement
@@ -149,22 +130,22 @@ def statement
   end
 end
 
-Benchmark.ips do |x|
-  x.warmup = 0
-  x.time = 10
-  BenchmarkCassandra.new.run_execute(x)
-  BenchmarkCassandra.new.run_execute_async(x)
-end
-
-GC.start
-sleep 20
+case ENV['RUN']
+when 'cassandra'
+  Benchmark.ips do |x|
+    x.warmup = 0
+    x.time = 20
+    BenchmarkCassandra.new.run_execute(x)
+    BenchmarkCassandra.new.run_execute_async(x)
+  end
 
-puts ''
-Benchmark.ips do |x|
-  x.warmup = 0
-  x.time = 10
-  BenchmarkIlios.new.run_execute(x)
-  BenchmarkIlios.new.run_execute_async(x)
+when 'ilios', nil
+  Benchmark.ips do |x|
+    x.warmup = 0
+    x.time = 20
+    BenchmarkIlios.new.run_execute(x)
+    BenchmarkIlios.new.run_execute_async(x)
+  end
 end
 
 =begin
@@ -176,13 +157,17 @@ def statement
 - Ruby: ruby 3.3.0
 
 ## Results
+### cassandra-driver
+$ RUN=cassandra ruby benchmark_insert.rb
 Calculating -------------------------------------
 cassandra-driver:execute
-                          4.121k (±19.4%) i/s -     39.035k in   9.979254s
+                          4.022k (±19.1%) i/s -     76.144k in  19.956985s
 cassandra-driver:execute_async
-                         18.461k (±20.5%) i/s -    132.226k in   9.951913s
+                         45.162k (±20.6%) i/s -    514.720k in  19.872873s
 
+### ilios
+$ RUN=ilios ruby benchmark_insert.rb
 Calculating -------------------------------------
-       ilios:execute      4.880k (±23.8%) i/s -     45.928k in   9.978697s
- ilios:execute_async    348.102k (±53.6%) i/s -    966.952k in   9.745057s
+       ilios:execute      4.857k (±24.0%) i/s -     91.235k in  19.960159s
+ ilios:execute_async    351.393k (±55.5%) i/s -      2.029M in  19.462660s
 =end
diff --git a/example/benchmark_select.rb b/example/benchmark_select.rb
@@ -74,22 +74,11 @@ def run_execute(x)
   end
 
   def run_execute_async(x)
-    futures = []
-
     x.report('cassandra-driver:execute_async') do
       future = @session.execute_async(statement)
       future.on_success do |rows|
       end
-
-      futures << future
-
-      if futures.size > 100
-        tmp = futures.slice!(0..10)
-        Cassandra::Future.all(*tmp).get
-      end
     end
-
-    Cassandra::Future.all(*futures).get
   end
 
   def statement
@@ -107,21 +96,13 @@ def run_execute(x)
   end
 
   def run_execute_async(x)
-    futures = []
-
     x.report('ilios:execute_async') do
       future = Ilios::Cassandra.session.execute_async(statement)
       future.on_success do |results|
         results.each do |row|
         end
       end
-
-      futures << future
-
-      futures.slice!(0..10).each(&:await) if futures.size > 100
     end
-
-    futures.each(&:await)
   end
 
   def statement
@@ -131,22 +112,22 @@ def statement
   end
 end
 
-Benchmark.ips do |x|
-  x.warmup = 0
-  x.time = 10
-  BenchmarkCassandra.new.run_execute(x)
-  BenchmarkCassandra.new.run_execute_async(x)
-end
-
-GC.start
-sleep 20
+case ENV['RUN']
+when 'cassandra'
+  Benchmark.ips do |x|
+    x.warmup = 0
+    x.time = 20
+    BenchmarkCassandra.new.run_execute(x)
+    BenchmarkCassandra.new.run_execute_async(x)
+  end
 
-puts ''
-Benchmark.ips do |x|
-  x.warmup = 0
-  x.time = 10
-  BenchmarkIlios.new.run_execute(x)
-  BenchmarkIlios.new.run_execute_async(x)
+when 'ilios', nil
+  Benchmark.ips do |x|
+    x.warmup = 0
+    x.time = 20
+    BenchmarkIlios.new.run_execute(x)
+    BenchmarkIlios.new.run_execute_async(x)
+  end
 end
 
 =begin
@@ -158,13 +139,17 @@ def statement
 - Ruby: ruby 3.3.0
 
 ## Results
+### cassandra-driver
+$ RUN=cassandra ruby benchmark_select.rb
 Calculating -------------------------------------
 cassandra-driver:execute
-                        139.571 (± 7.2%) i/s -      1.387k in   9.998314s
+                        140.897 (± 9.2%) i/s -      2.794k in  19.995255s
 cassandra-driver:execute_async
-                         11.554k (±86.6%) i/s -      2.467k in  10.001188s
+                         95.534k (±17.8%) i/s -      1.003M in  20.788894s
 
+### ilios
+RUN=ilios ruby benchmark_select.rb
 Calculating -------------------------------------
-       ilios:execute    327.715 (±16.5%) i/s -      3.187k in   9.994403s
- ilios:execute_async    454.625k (±78.4%) i/s -     10.161k in   9.996146s
+       ilios:execute    327.603 (±15.6%) i/s -      6.400k in  19.991734s
+ ilios:execute_async    421.849k (±80.8%) i/s -     19.441k in  19.987868s
 =end
diff --git a/ext/ilios/future.c b/ext/ilios/future.c
@@ -1,6 +1,7 @@
 #include "ilios.h"
 
 #define THREAD_MAX 5
+#define QUEUE_MAX 100
 
 typedef struct
 {
@@ -31,7 +32,7 @@ const rb_data_type_t cassandra_future_data_type = {
 
 static void future_thread_pool_init(future_thread_pool *pool)
 {
-    pool->queue = rb_funcall(cQueue, id_new, 0);
+    pool->queue = rb_funcall(cSizedQueue, id_new, 1, INT2NUM(QUEUE_MAX));
     rb_gc_register_mark_object(pool->queue);
 }
 

diff --git a/ext/ilios/ilios.c b/ext/ilios/ilios.c
@@ -11,7 +11,7 @@ VALUE eConnectError;
 VALUE eExecutionError;
 VALUE eStatementError;
 
-VALUE cQueue;
+VALUE cSizedQueue;
 
 VALUE id_to_time;
 VALUE id_new;
@@ -85,7 +85,7 @@ void Init_ilios(void)
     eExecutionError = rb_define_class_under(mCassandra, "ExecutionError", rb_eStandardError);
     eStatementError = rb_define_class_under(mCassandra, "StatementError", rb_eStandardError);
 
-    cQueue = rb_const_get(rb_cThread, rb_intern("Queue"));
+    cSizedQueue = rb_const_get(rb_cThread, rb_intern("SizedQueue"));
 
     id_to_time = rb_intern("to_time");
     id_new = rb_intern("new");

diff --git a/ext/ilios/ilios.h b/ext/ilios/ilios.h
@@ -84,7 +84,7 @@ extern VALUE eConnectError;
 extern VALUE eExecutionError;
 extern VALUE eStatementError;
 
-extern VALUE cQueue;
+extern VALUE cSizedQueue;
 
 extern VALUE id_to_time;
 extern VALUE id_new;