Add request muxer remove jnats

.circleci/config.yml

@@ -13,10 +13,9 @@ jobs:
         environment:
           JRUBY_OPTS: "-J-Xmx1024m"
           RAILS_ENV: test
-          NATS_URL: "changeme"
 
       # 2. The Service Container (runs in the background)
-      - image: nats:2.14.1-linux
+      - image: nats:2.14-linux
 
     working_directory: ~/project
 
@@ -82,4 +81,4 @@ workflows:
                 - "cimg/ruby:3.1"
                 - "cimg/ruby:3.4"
                 - "jruby:9.4"
-                - "jruby:10.0"
+                - "jruby:10.0"

CHANGELOG.md

@@ -0,0 +1,7 @@
+## Changelog
+
+### 0.13.0 - WIP
+- Removed JNats (`nats-pure` is fast enough for JRuby and CRuby parallel work)
+- Added ResponseMuxer (similar to Golang)
+- Added instrumentation when encountering unexpected messages.
+

README.md

@@ -148,13 +148,20 @@ And we can see the message was sent to the server and the server replied with a
 If we were to add another service endpoint called `search` to the `UserService` but fail to define an instance method
 `search`, then `protobuf-nats` will not subscribe to that route.
 
+## Future Improvements (locked behind ruby version)
+- Migrate to native `Random.new.uuid_v7`
+```ruby
+@prng_lock.synchronize { @prng.uuid_v7(extra_timestamp_bits: 12) }
+```
+- Change ResponseMuxer to use `.pop()` with a timeout.
+
+
 ## Development
 
 After checking out the repo, run `bin/setup` to install dependencies. Then, run `rake test` to run the tests. You can also run `bin/console` for an interactive prompt that will allow you to experiment.
 
 To install this gem onto your local machine, run `bundle exec rake install`. To release a new version, update the version number in `version.rb`, and then run `bundle exec rake release`, which will create a git tag for the version, push git commits and tags, and push the `.gem` file to [rubygems.org](https://rubygems.org).
 
-The java-nats client is temporarily forked to support jruby > 9.2.10.0. The living branch for that is here: https://github.com/film42/java-nats/tree/jruby-compat. This will be removed when we upgrade to the new nats.java client.
 
 ## Contributing
 

bench/bench.md

@@ -0,0 +1,16 @@
+
+Notes:
+`-Xjit.threshold=0` - Setting the threshold to 0 forces JRuby to compile every method into Java bytecode immediately before its very first execution. This is particularly useful for debugging or bypassing warm-up times during profiling
+
+
+`-Xjit.threshold=10 -J-XX:CompileThreshold=10` - If you are running benchmarks and want both JRuby and the JVM to aggressively optimize early, you can lower both thresholds simultaneously
+
+`bundle;  bx ruby -I lib bench/real_client.rb`
+
+Start local nats server so details can be monitored.
+`/opt/homebrew/opt/nats-server/bin/nats-server -DV -m 8222 -p 4222`
+
+
+```
+export JRUBY_OPTS="--disable:did_you_mean -J-Djava.security.egd=file:/dev/./urandom -J-Xmx2g -J-Xms1024m -J-Xmn512m -Xjit.threshold=10 -J-XX:CompileThreshold=10"
+```

bench/console.rb

@@ -0,0 +1,17 @@
+#!/usr/bin/env ruby
+
+require "bundler/setup"
+require "protobuf/nats"
+
+# You can add fixtures and/or initialization code here to make experimenting
+# with your gem easier. You can also use a different console, if you like.
+
+# (If you use this, don't forget to add pry to your Gemfile!)
+# require "pry"
+# Pry.start
+
+# ENV["PB_CLIENT_TYPE"] = "protobuf/nats/client"
+# ENV["PB_SERVER_TYPE"] = "protobuf/nats/runner"
+
+require "irb"
+IRB.start(__FILE__)

bench/real_client.rb

@@ -7,8 +7,8 @@
 Protobuf::Logging.logger = ::Logger.new(nil)
 
 Benchmark.ips do |config|
-  config.warmup = 10
-  config.time = 10
+  config.warmup = 30
+  config.time = 30
 
   config.report("single threaded performance") do
     req = Warehouse::Shipment.new(:guid => SecureRandom.uuid)

bench/real_client.sh

@@ -0,0 +1,13 @@
+#!/bin/bash
+
+export JRUBY_OPTS="-J-server --disable:did_you_mean -Xcompile.invokedynamic=true -Xjit.threshold=10 -J-Djava.security.egd=file:/dev/./urandom -J-Xms2g -J-Xmx2g -J-XX:+UseG1GC -J-XX:MaxGCPauseMillis=100"
+
+# export JRUBY_OPTS="-J-server --disable:did_you_mean -Xcompile.invokedynamic=true -Xjit.threshold=0 -J-Djruby.jit.max=0 -J-Djruby.jit.background=false -J-Djava.security.egd=file:/dev/./urandom -J-Xms2g -J-Xmx2g -J-XX:+UseG1GC -J-XX:MaxGCPauseMillis=100"
+
+
+export PB_SERVER_TYPE="protobuf/nats/runner"
+export PB_CLIENT_TYPE="protobuf/nats/client"
+
+echo "$PWD"
+
+bundle exec ruby -I lib bench/real_client.rb

bench/real_client_threaded.rb

@@ -0,0 +1,19 @@
+ENV["PB_CLIENT_TYPE"] = "protobuf/nats/client"
+ENV["PB_SERVER_TYPE"] = "protobuf/nats/runner"
+
+require "./examples/warehouse/app"
+
+THREAD_COUNT = ENV.fetch("CLIENT_THREADS",4).to_i
+
+puts "THREAD_COUNT = #{THREAD_COUNT}"
+
+::Protobuf::Logging.logger = ::Logger.new(nil)
+
+while true
+  THREAD_COUNT.times.map do |i|
+    Thread.new do
+      req = Warehouse::Shipment.new(:guid => SecureRandom.uuid, :sleep_time_ms => 5)
+      Warehouse::ShipmentService.client.create(req)
+    end
+  end.each(&:join)
+end

bench/real_client_threaded.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+
+export JRUBY_OPTS="-J-server --disable:did_you_mean -Xcompile.invokedynamic=true -Xjit.threshold=10 -J-Djava.security.egd=file:/dev/./urandom -J-Xms2g -J-Xmx2g -J-XX:+UseG1GC -J-XX:MaxGCPauseMillis=100"
+
+export PB_SERVER_TYPE="protobuf/nats/runner"
+export PB_CLIENT_TYPE="protobuf/nats/client"
+
+export THREAD_COUNT=4
+
+bundle exec ruby -I lib bench/real_client_threaded.rb

bench/real_server.sh

@@ -1 +1,16 @@
-PB_SERVER_TYPE="protobuf/nats/runner" PB_CLIENT_TYPE="protobuf/nats/client" bundle exec rpc_server start --threads=20 ./examples/warehouse/app.rb > /dev/null
+#!/bin/bash
+
+# export JRUBY_OPTS="-J-server --disable:did_you_mean -Xcompile.invokedynamic=true -Xjit.threshold=0 -J-Djruby.jit.max=0 -J-Djruby.jit.background=false -J-Djava.security.egd=file:/dev/./urandom -J-Xms2g -J-Xmx2g -J-XX:+UseG1GC -J-XX:MaxGCPauseMillis=100"
+
+export JRUBY_OPTS="-J-server -J-Xms4g -J-Xmx4g -J-XX:+AlwaysPreTouch -J-XX:+UseParallelGC -J-XX:ReservedCodeCacheSize=768m -J-XX:MaxInlineLevel=18 -J-XX:MaxInlineSize=100 -J-XX:FreqInlineSize=500 -J-XX:LoopUnrollLimit=250 -J-XX:+UseSuperWord -J-Djruby.jit.threshold=0 -J-Djruby.jit.max=0 -J-Djruby.jit.background=false  -Xcompile.invokedynamic=true"
+
+
+export PB_SERVER_TYPE="protobuf/nats/runner"
+export PB_CLIENT_TYPE="protobuf/nats/client"
+
+export PB_NATS_SERVER_SLOW_START_DELAY=1
+
+export PB_NATS_SERVER_MAX_QUEUE_SIZE=6
+
+bundle exec rpc_server start --threads=10 ./examples/warehouse/app.rb
+

examples/warehouse/app.rb

@@ -12,7 +12,6 @@ class Shipment < ::Protobuf::Message; end
   class ShipmentRequest < ::Protobuf::Message; end
   class Shipments < ::Protobuf::Message; end
 
-
   ##
   # Message Fields
   #
@@ -21,6 +20,8 @@ class Shipment
     optional :string, :address, 2
     optional :double, :price, 3
     optional :string, :package_guid, 4
+
+    optional :int64, :sleep_time_ms, 100
   end
 
   class ShipmentRequest
@@ -33,7 +34,6 @@ class Shipments
     repeated ::Warehouse::Shipment, :records, 1
   end
 
-
   ##
   # Service Classes
   #
@@ -43,6 +43,12 @@ class ShipmentService < ::Protobuf::Rpc::Service
     rpc :search, ::Warehouse::ShipmentRequest, ::Warehouse::Shipments
 
     def create
+      # Allows for easier testing of multiple threads
+      if request.sleep_time_ms > 0
+        sleep(request.sleep_time_ms / 1000.0)
+        puts "sleep_time:#{request.sleep_time_ms}"
+      end
+
       respond_with request
     end
 

lib/protobuf/nats.rb

@@ -6,11 +6,17 @@
 
 require "nats/io/client"
 
-require "protobuf/nats/errors"
+
+
 require "protobuf/nats/client"
-require "protobuf/nats/server"
-require "protobuf/nats/runner"
 require "protobuf/nats/config"
+require "protobuf/nats/errors"
+require "protobuf/nats/runner"
+require "protobuf/nats/server"
+
+require "protobuf/nats/response_muxer"
+require "protobuf/nats/response_muxer_request"
+require "protobuf/nats/super_subscription_manager"
 
 module Protobuf
   module Nats
@@ -23,12 +29,7 @@ module Messages
       NACK = "\2".freeze
     end
 
-    NatsClient = if defined? JRUBY_VERSION
-                   require "protobuf/nats/jnats"
-                   ::Protobuf::Nats::JNats
-                 else
-                   ::NATS::IO::Client
-                 end
+    NatsClient = ::NATS::IO::Client
 
     GET_CONNECTED_MUTEX = ::Mutex.new
 
@@ -114,7 +115,6 @@ def self.start_client_nats_connection
       end
     end
 
-    # This will work with both ruby and java errors
     def self.log_error(error)
       logger.error error.to_s
       logger.error error.class.to_s