RCBC-551: FIT performer - add support for CounterEq bounds type

fit-performer/lib/fit/performer/bounds.rb

@@ -0,0 +1,51 @@
+# frozen_string_literal: true
+
+#  Copyright 2026-Present Couchbase, Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+require_relative 'bounds/simple'
+require_relative 'bounds/for_time'
+require_relative 'bounds/counter_equality'
+require_relative 'bounds/counter'
+
+module FIT
+  module Performer
+    module Bounds
+      def self.create_bounds(raw_workload, global_counters)
+        return Simple.new(raw_workload.command.size) unless raw_workload.has_bounds?
+
+        raw_bounds = raw_workload.bounds
+        case raw_bounds.bounds
+        when :counter
+          counter_id = raw_bounds.counter.counter_id
+          initial_value = raw_bounds.counter.global.count
+          global_counter = global_counters.get_counter(counter_id, initial_value)
+          Counter.new(global_counter)
+
+        when :for_time
+          ForTime.new(raw_bounds.for_time.seconds)
+
+        when :counter_eq
+          counter_id = raw_bounds.counter_eq.counter_id
+          initial_value = raw_bounds.counter_eq.global.count
+          global_counter = global_counters.get_counter(counter_id, initial_value)
+          CounterEquality.new(global_counter, initial_value)
+
+        else
+          raise PerformerError, "Bounds type `#{raw_bounds.bounds}` not recognised"
+        end
+      end
+    end
+  end
+end

fit-performer/lib/fit/performer/bounds/counter.rb

@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+#  Copyright 2026-Present Couchbase, Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+module FIT
+  module Performer
+    module Bounds
+      # A bounds implementation that executes commands for a specified number of times, based on a global counter.
+      # The global counter may be shared with other workloads.
+      class Counter
+        # Initializes a new instance of the Counter bounds implementation.
+        #
+        # @param global_counter [Concurrent::AtomicFixnum] The global counter to use for determining if the workload
+        # can execute.
+        def initialize(global_counter)
+          @global_counter = global_counter
+        end
+
+        def can_execute?
+          @global_counter.decrement >= 0
+        end
+      end
+    end
+  end
+end

fit-performer/lib/fit/performer/bounds/counter_equality.rb

@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+#  Copyright 2026-Present Couchbase, Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+module FIT
+  module Performer
+    module Bounds
+      # A bounds implementation that executes commands while a global counter remains unchanged.
+      class CounterEquality
+        # Initializes a new instance of the CounterEquality bounds implementation.
+        #
+        # @param global_counter [Concurrent::AtomicFixnum] The global counter to use for determining if the workload
+        # can execute.
+        # @param initial_value [Integer] The initial value of the counter to compare against
+        def initialize(global_counter, initial_value)
+          @global_counter = global_counter
+          @initial_value = initial_value
+        end
+
+        def can_execute?
+          @global_counter.value == @initial_value
+        end
+      end
+    end
+  end
+end

fit-performer/lib/fit/performer/bounds/for_time.rb

@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+#  Copyright 2026-Present Couchbase, Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+module FIT
+  module Performer
+    module Bounds
+      # A bounds implementation that executes commands for a specified duration of time.
+      class ForTime
+        def initialize(duration_seconds)
+          @deadline = Time.now + duration_seconds
+        end
+
+        def can_execute?
+          Time.now < @deadline
+        end
+      end
+    end
+  end
+end

fit-performer/lib/fit/performer/bounds/simple.rb

@@ -0,0 +1,33 @@
+# frozen_string_literal: true
+
+#  Copyright 2026-Present Couchbase, Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+module FIT
+  module Performer
+    module Bounds
+      # A simple bounds implementation that executes every command in the workload once.
+      class Simple
+        def initialize(command_count)
+          @remaining_command_count = command_count
+        end
+
+        def can_execute?
+          @remaining_command_count -= 1
+          @remaining_command_count >= 0
+        end
+      end
+    end
+  end
+end

fit-performer/lib/fit/performer/counters.rb

@@ -0,0 +1,54 @@
+# frozen_string_literal: true
+
+#  Copyright 2026-Present Couchbase, Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+require 'concurrent/atomic/atomic_fixnum'
+
+module FIT
+  module Performer
+    class Counters
+      def initialize
+        @counters = {}
+        @mutex = Mutex.new
+      end
+
+      def clear
+        @mutex.synchronize do
+          @counters.clear
+        end
+      end
+
+      def set_counter_value(counter_id, value)
+        @mutex.synchronize do
+          if @counters.key?(counter_id)
+            @counters[counter_id].value = value
+          else
+            @counters[counter_id] = Concurrent::AtomicFixnum.new(value)
+          end
+        end
+      end
+
+      def get_counter(counter_id, initial_value = nil)
+        @mutex.synchronize do
+          return @counters[counter_id] if @counters.key?(counter_id)
+
+          raise PerformerError, "Counter #{counter_id} does not exist and no initial value was provided" if initial_value.nil?
+
+          @counters[counter_id] = Concurrent::AtomicFixnum.new(initial_value)
+        end
+      end
+    end
+  end
+end

fit-performer/lib/fit/performer/multi_thread_executor.rb

@@ -22,30 +22,28 @@ module Performer
     class MultiThreadExecutor
       attr_reader :results
 
-      def initialize(connection, run_id, stream_owner, span_owner)
+      def initialize(connection, run_id, stream_owner, span_owner, global_counters)
         @logger = Logger.new($stdout)
         @results = Queue.new
         @connection = connection
         @run_id = run_id
         @stream_owner = stream_owner
         @span_owner = span_owner
-        @global_counters = {}
-        @global_semaphore = Mutex.new
+        @global_counters = global_counters
       end
 
       def build_workloads(request)
         @workloads = request.workloads.horizontal_scaling.map do |h|
           h.workloads.map do |w|
-            Workloads.build_workload(w, @connection, @run_id, @stream_owner, @span_owner)
+            Workloads.build_workload(w, @connection, @run_id, @stream_owner, @span_owner, @global_counters)
           end
         end
       end
 
       def start_workload_thread(workloads)
         Thread.new do # rubocop:disable ThreadSafety/NewThread
           workloads.each do |w|
-            w.set_bounds(@global_counters, @global_semaphore)
-            w.execute(@results, @global_counters, @global_semaphore)
+            w.execute(@results)
           end
         end
       end
@@ -69,8 +67,8 @@ def execute_workloads
         end
       end
 
-      def self.build_executor(request, connection, run_id, stream_owner, span_owner)
-        executor = MultiThreadExecutor.new(connection, run_id, stream_owner, span_owner)
+      def self.build_executor(request, connection, run_id, stream_owner, span_owner, global_counters)
+        executor = MultiThreadExecutor.new(connection, run_id, stream_owner, span_owner, global_counters)
         executor.build_workloads(request)
         executor
       end

fit-performer/lib/fit/performer/service.rb

@@ -27,6 +27,7 @@
 require 'fit/performer/request_executor'
 require 'fit/performer/streaming'
 require 'fit/performer/observability/span_owner'
+require 'fit/performer/counters'
 
 require 'fit/protocol/performer_services_pb'
 require 'fit/protocol/performer.caps_pb'
@@ -95,6 +96,7 @@ def initialize
         @connections = Connections.new
         @stream_owner = Streaming::StreamOwner.new
         @span_owner = Observability::SpanOwner.new
+        @global_counters = Counters.new
         @logger.info("Using Ruby SDK version #{SDK_VERSION}")
       end
 
@@ -144,7 +146,7 @@ def run(request, _call)
           begin
             connection = @connections[request.workloads.cluster_connection_id]
             workload_executor = MultiThreadExecutor.build_executor(
-              request, connection, run_id, @stream_owner, @span_owner
+              request, connection, run_id, @stream_owner, @span_owner, @global_counters
             )
             @logger.info("Created the workload executor")
             req_executor = RequestExecutor.build_request(workload_executor, request)
@@ -188,6 +190,16 @@ def span_finish(request, _call)
       rescue PerformerError => e
         raise GRPC::FailedPrecondition, e.message
       end
+
+      def set_counter(request, _call)
+        @global_counters.set_counter_value(request.counter_id, request.global.count)
+        FIT::Protocol::Shared::SetCounterResponse.new
+      end
+
+      def clear_all_counters(_request, _call)
+        @global_counters.clear
+        FIT::Protocol::Shared::ClearAllCountersResponse.new
+      end
     end
   end
 end

fit-performer/lib/fit/performer/workloads.rb

@@ -20,11 +20,11 @@
 module FIT
   module Performer
     module Workloads
-      def self.build_workload(raw_workload, connection, run_id, stream_owner, span_owner)
+      def self.build_workload(raw_workload, connection, run_id, stream_owner, span_owner, global_counters)
         workload_type = raw_workload.workload
         case workload_type
         when :sdk
-          SdkWorkload.new(raw_workload.sdk, connection, run_id, stream_owner, span_owner)
+          SdkWorkload.new(raw_workload.sdk, connection, run_id, stream_owner, span_owner, global_counters)
         else
           raise PerformerError, "Workload type `#{workload_type}` not supported"
         end