From 09b4a40f40a326246b421191ced38366ac951493 Mon Sep 17 00:00:00 2001 From: Dimitris Christodoulou Date: Fri, 3 Jul 2026 16:44:31 +0100 Subject: [PATCH] RCBC-551: FIT performer - add support for CounterEq bounds type --- fit-performer/lib/fit/performer/bounds.rb | 51 +++++++++++++++++ .../lib/fit/performer/bounds/counter.rb | 37 ++++++++++++ .../fit/performer/bounds/counter_equality.rb | 38 +++++++++++++ .../lib/fit/performer/bounds/for_time.rb | 32 +++++++++++ .../lib/fit/performer/bounds/simple.rb | 33 +++++++++++ fit-performer/lib/fit/performer/counters.rb | 54 ++++++++++++++++++ .../fit/performer/multi_thread_executor.rb | 14 ++--- fit-performer/lib/fit/performer/service.rb | 14 ++++- fit-performer/lib/fit/performer/workloads.rb | 4 +- .../fit/performer/workloads/base_workload.rb | 57 ++----------------- .../fit/performer/workloads/sdk_workload.rb | 6 +- 11 files changed, 273 insertions(+), 67 deletions(-) create mode 100644 fit-performer/lib/fit/performer/bounds.rb create mode 100644 fit-performer/lib/fit/performer/bounds/counter.rb create mode 100644 fit-performer/lib/fit/performer/bounds/counter_equality.rb create mode 100644 fit-performer/lib/fit/performer/bounds/for_time.rb create mode 100644 fit-performer/lib/fit/performer/bounds/simple.rb create mode 100644 fit-performer/lib/fit/performer/counters.rb diff --git a/fit-performer/lib/fit/performer/bounds.rb b/fit-performer/lib/fit/performer/bounds.rb new file mode 100644 index 00000000..629ee422 --- /dev/null +++ b/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 diff --git a/fit-performer/lib/fit/performer/bounds/counter.rb b/fit-performer/lib/fit/performer/bounds/counter.rb new file mode 100644 index 00000000..d282ad40 --- /dev/null +++ b/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 diff --git a/fit-performer/lib/fit/performer/bounds/counter_equality.rb b/fit-performer/lib/fit/performer/bounds/counter_equality.rb new file mode 100644 index 00000000..df5166b2 --- /dev/null +++ b/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 diff --git a/fit-performer/lib/fit/performer/bounds/for_time.rb b/fit-performer/lib/fit/performer/bounds/for_time.rb new file mode 100644 index 00000000..40fba8e2 --- /dev/null +++ b/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 diff --git a/fit-performer/lib/fit/performer/bounds/simple.rb b/fit-performer/lib/fit/performer/bounds/simple.rb new file mode 100644 index 00000000..c7fc24b7 --- /dev/null +++ b/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 diff --git a/fit-performer/lib/fit/performer/counters.rb b/fit-performer/lib/fit/performer/counters.rb new file mode 100644 index 00000000..77e464d9 --- /dev/null +++ b/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 diff --git a/fit-performer/lib/fit/performer/multi_thread_executor.rb b/fit-performer/lib/fit/performer/multi_thread_executor.rb index 9964f0c7..521fadf1 100644 --- a/fit-performer/lib/fit/performer/multi_thread_executor.rb +++ b/fit-performer/lib/fit/performer/multi_thread_executor.rb @@ -22,21 +22,20 @@ 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 @@ -44,8 +43,7 @@ def build_workloads(request) 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 diff --git a/fit-performer/lib/fit/performer/service.rb b/fit-performer/lib/fit/performer/service.rb index 7394b34c..ba9f2029 100644 --- a/fit-performer/lib/fit/performer/service.rb +++ b/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 diff --git a/fit-performer/lib/fit/performer/workloads.rb b/fit-performer/lib/fit/performer/workloads.rb index 78106d2f..6e95ec6e 100644 --- a/fit-performer/lib/fit/performer/workloads.rb +++ b/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 diff --git a/fit-performer/lib/fit/performer/workloads/base_workload.rb b/fit-performer/lib/fit/performer/workloads/base_workload.rb index 334ba754..559a2498 100644 --- a/fit-performer/lib/fit/performer/workloads/base_workload.rb +++ b/fit-performer/lib/fit/performer/workloads/base_workload.rb @@ -15,12 +15,13 @@ # limitations under the License. require 'fit/performer/performer_error' +require 'fit/performer/bounds' module FIT module Performer module Workloads class BaseWorkload - def initialize(raw_workload, connection, run_id, stream_owner, span_owner) + def initialize(raw_workload, connection, run_id, stream_owner, span_owner, global_counters) @logger = Logger.new($stdout) @raw_workload = raw_workload @connection = connection @@ -28,63 +29,13 @@ def initialize(raw_workload, connection, run_id, stream_owner, span_owner) @stream_owner = stream_owner @span_owner = span_owner @command_count = @raw_workload.command.size - - # Bounds-related attributes - @counter_id = nil - @deadline = nil - @remaining_command_count = nil + @bounds = Bounds.create_bounds(@raw_workload, global_counters) end - def execute(_results, _global_counters, _global_semaphore) + def execute(_results) raise "`execute` method must be implemented" end - def set_bounds(global_counters, global_semaphore) - unless @raw_workload.has_bounds? - @remaining_command_count = @command_count - return - end - - raw_bounds = @raw_workload.bounds - case raw_bounds.bounds - when :counter - raw_counter = raw_bounds.counter - @counter_id = raw_counter.counter_id - - case raw_counter.counter - when :global - global_semaphore.synchronize do - unless global_counters.include?(@counter_id) - initial_count = raw_counter.global.count - global_counters[@counter_id] = initial_count - end - end - else - raise PerformerError, "Counter type `#{counter.counter}` not recognised" - end - when :for_time - @deadline = Time.now + raw_bounds.for_time.seconds - else - raise PerformerError, "Bounds type `#{@raw_workload.bounds}` not recognised" - end - end - - def within_bounds(global_counters, global_semaphore) - if !@counter_id.nil? - global_semaphore.synchronize do - global_counters[@counter_id] -= 1 - global_counters[@counter_id] >= 0 - end - elsif !@deadline.nil? - Time.now < @deadline - elsif !@remaining_command_count.nil? - @remaining_command_count -= 1 - @remaining_command_count >= 0 - else - raise PerformerError, "Bounds have not been set" - end - end - private def execute_command(_raw_command, _executed_cmd_count, _results) diff --git a/fit-performer/lib/fit/performer/workloads/sdk_workload.rb b/fit-performer/lib/fit/performer/workloads/sdk_workload.rb index d8ae199f..fbc2a942 100644 --- a/fit-performer/lib/fit/performer/workloads/sdk_workload.rb +++ b/fit-performer/lib/fit/performer/workloads/sdk_workload.rb @@ -22,14 +22,14 @@ module FIT module Performer module Workloads class SdkWorkload < BaseWorkload - def initialize(raw_workload, connection, run_id, stream_owner, span_owner) + def initialize(raw_workload, connection, run_id, stream_owner, span_owner, global_counters) super @logger = Logger.new($stdout) end - def execute(results, global_counters, global_semaphore) + def execute(results) executed_cmd_count = 0 - while within_bounds(global_counters, global_semaphore) + while @bounds.can_execute? raw_cmd = @raw_workload.command[executed_cmd_count % @command_count] execute_command(raw_cmd, executed_cmd_count, results) executed_cmd_count += 1