Add stress testing framework, with basic metrics example to demonstrate.

CMakeLists.txt

@@ -170,6 +170,8 @@ option(BUILD_TESTING "Whether to enable tests" ON)
 
 option(WITH_BENCHMARK "Whether to build benchmark program" ON)
 
+option(WITH_STRESS_TESTS "Whether to build stress test programs" OFF)
+
 option(BUILD_W3CTRACECONTEXT_TEST "Whether to build w3c trace context" OFF)
 
 option(OTELCPP_MAINTAINER_MODE "Build in maintainer mode (-Wall -Werror)" OFF)
@@ -531,6 +533,7 @@ message(STATUS "---------------------------------------------")
 message(STATUS "examples and test options")
 message(STATUS "---------------------------------------------")
 message(STATUS "WITH_BENCHMARK: ${WITH_BENCHMARK}")
+message(STATUS "WITH_STRESS_TESTS: ${WITH_STRESS_TESTS}")
 message(STATUS "WITH_EXAMPLES: ${WITH_EXAMPLES}")
 message(STATUS "WITH_EXAMPLES_HTTP: ${WITH_EXAMPLES_HTTP}")
 message(STATUS "WITH_FUNC_TESTS: ${WITH_FUNC_TESTS}")
@@ -633,6 +636,9 @@ if(NOT WITH_API_ONLY)
   if(WITH_FUNC_TESTS)
     add_subdirectory(functional)
   endif()
+  if(WITH_STRESS_TESTS)
+    add_subdirectory(stress)
+  endif()
 endif()
 
 include(

stress/CMakeLists.txt

@@ -0,0 +1,6 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+# Add subdirectories for common and metrics components
+add_subdirectory(common)
+add_subdirectory(metrics)

stress/common/CMakeLists.txt

@@ -0,0 +1,14 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+add_library(stress STATIC stress.cc)
+
+# Include directory for the throughput library
+target_include_directories(stress PUBLIC ${CMAKE_CURRENT_SOURCE_DIR})
+
+# Set C++ standard
+set_target_properties(
+  stress
+  PROPERTIES CXX_STANDARD 17
+             CXX_STANDARD_REQUIRED YES
+             CXX_EXTENSIONS NO)

stress/common/stress.cc

@@ -0,0 +1,153 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include "stress.h"
+
+#include <algorithm>
+#include <limits>
+#include <locale>
+#include <utility>
+
+#ifdef __linux__
+#  include <pthread.h>
+#  include <sched.h>
+#endif
+
+// StressTest constructor
+Stress::Stress(std::function<void()> func, size_t numThreads)
+    : func_(std::move(func)), stats_(numThreads), numThreads_(numThreads)
+{}
+
+// Main function to start the stress test
+void Stress::run()
+{
+  std::cout << "Starting stress test with " << numThreads_ << " threads...\n";
+  stopFlag_.store(false, std::memory_order_release);
+  readyFlag_.store(false, std::memory_order_release);
+
+  auto startTime = std::chrono::steady_clock::now();
+
+  std::thread controllerThread(&Stress::monitorThroughput, this);
+
+  threads_.reserve(numThreads_);
+  for (size_t i = 0; i < numThreads_; ++i)
+  {
+    threads_.emplace_back(&Stress::workerThread, this, i);
+  }
+
+  readyFlag_.store(true, std::memory_order_release);
+
+  for (auto &thread : threads_)
+  {
+    if (thread.joinable())
+    {
+      thread.join();
+    }
+  }
+
+  if (controllerThread.joinable())
+  {
+    controllerThread.join();
+  }
+
+  auto endTime  = std::chrono::steady_clock::now();
+  auto duration = std::chrono::duration_cast<std::chrono::seconds>(endTime - startTime);
+
+  uint64_t totalCount = 0;
+  for (const auto &stat : stats_)
+  {
+    totalCount += stat.count.load(std::memory_order_relaxed);
+  }
+
+  const auto duration_seconds = duration.count();
+  const auto average_throughput =
+      duration_seconds > 0 ? totalCount / static_cast<uint64_t>(duration_seconds) : 0;
+
+  std::cout << "\nTest completed:\n"
+            << "Total iterations: " << formatNumber(totalCount) << "\n"
+            << "Duration: " << duration_seconds << " seconds\n"
+            << "Average throughput: " << formatNumber(average_throughput) << " iterations/sec\n";
+}
+
+// Worker thread function
+void Stress::workerThread(size_t threadIndex)
+{
+#ifdef __linux__
+  cpu_set_t cpuset;
+  CPU_ZERO(&cpuset);
+  CPU_SET(threadIndex % std::thread::hardware_concurrency(), &cpuset);
+  pthread_setaffinity_np(pthread_self(), sizeof(cpuset), &cpuset);
+#endif
+
+  while (!readyFlag_.load(std::memory_order_acquire))
+  {
+    std::this_thread::yield();
+  }
+
+  while (!stopFlag_.load(std::memory_order_acquire))
+  {
+    func_();
+    stats_[threadIndex].count.fetch_add(1, std::memory_order_relaxed);
+  }
+}
+
+// Monitoring thread function
+void Stress::monitorThroughput()
+{
+  uint64_t lastTotalCount = 0;
+  auto lastTime           = std::chrono::steady_clock::now();
+  uint64_t sampleCount    = 0;
+  uint64_t throughputSum  = 0;
+  uint64_t minThroughput  = std::numeric_limits<uint64_t>::max();
+  uint64_t maxThroughput  = 0;
+
+  while (!stopFlag_.load(std::memory_order_acquire))
+  {
+    std::this_thread::sleep_for(std::chrono::seconds(SLIDING_WINDOW_SIZE));
+
+    auto currentTime = std::chrono::steady_clock::now();
+    auto elapsed =
+        std::chrono::duration_cast<std::chrono::duration<double>>(currentTime - lastTime).count();
+
+    uint64_t totalCount = 0;
+    for (const auto &stat : stats_)
+    {
+      totalCount += stat.count.load(std::memory_order_relaxed);
+    }
+
+    uint64_t currentCount = totalCount - lastTotalCount;
+    lastTotalCount        = totalCount;
+    lastTime              = currentTime;
+
+    if (elapsed > 0.0)
+    {
+      uint64_t throughput = static_cast<uint64_t>(static_cast<double>(currentCount) / elapsed);
+      ++sampleCount;
+      throughputSum += throughput;
+      minThroughput                = std::min(minThroughput, throughput);
+      maxThroughput                = std::max(maxThroughput, throughput);
+      const uint64_t avgThroughput = throughputSum / sampleCount;
+
+      std::cout << "\rThroughput: " << formatNumber(throughput)
+                << " it/s | Avg: " << formatNumber(avgThroughput)
+                << " | Min: " << formatNumber(minThroughput)
+                << " | Max: " << formatNumber(maxThroughput) << std::flush;
+    }
+  }
+  std::cout << std::endl;
+}
+
+// Helper function to format numbers with commas
+std::string Stress::formatNumber(uint64_t num)
+{
+  std::ostringstream oss;
+  oss.imbue(std::locale(""));
+  oss << std::fixed << num;
+  return oss.str();
+}
+
+// Signal handler to set the STOP flag when receiving a termination signal
+void Stress::stop()
+{
+  stopFlag_.store(true, std::memory_order_release);
+}

stress/common/stress.h

@@ -0,0 +1,100 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+/**
+ * A multi-threaded stress test framework to measure throughput and performance of a given workload.
+ *
+ * ## Overview
+ * Multi-threaded stress test framework designed to execute a specified function
+ * in parallel across multiple threads and measure its throughput. The results are displayed
+ * dynamically, including current throughput, average throughput, and minimum/maximum throughput
+ * during the test.
+ *
+ * ## Key Features
+ * - **Multi-threading**: Uses std::thread to execute the workload in parallel across a user-defined
+ * number of threads.
+ * - **Thread Safety**: Tracks iteration counts per thread using an aligned and padded structure
+ * (WorkerStats) to avoid false sharing and ensure efficient thread-safe updates.
+ * - **Dynamic Metrics**: Continuously calculates and displays throughput (iterations/sec) over
+ * sliding time windows.
+ * - **Graceful Termination**: Captures signals (e.g., Ctrl+C) to cleanly stop all threads and
+ * summarize the results.
+ * - **Thread Affinity (Linux-only)**: Optionally binds threads to specific CPU cores for consistent
+ * performance.
+ *
+ * ## Implementation Details
+ * - **Worker Threads**:
+ *   - Each worker thread executes the workload function (func) in a loop until a global STOP flag
+ * is set.
+ *   - Each thread maintains its own iteration count to minimize contention.
+ *
+ * - **Throughput Monitoring**:
+ *   - A separate controller thread monitors throughput by periodically summing up iteration counts
+ * across threads.
+ *   - Throughput is calculated over a sliding window (SLIDING_WINDOW_SIZE) and displayed
+ * dynamically.
+ *
+ * - **Thread Synchronization**:
+ *   - The STOP flag, an std::atomic<bool>, ensures all threads stop gracefully when signaled.
+ *   - Memory ordering (e.g., std::memory_order_relaxed, std::memory_order_acquire/release) is used
+ * to optimize performance while maintaining correctness.
+ *
+ * - **Final Summary**:
+ *   - At the end of the test, the program calculates and prints the total iterations, duration, and
+ * average throughput.
+ */
+
+#pragma once
+
+#include <atomic>
+#include <chrono>
+#include <csignal>
+#include <cstdint>
+#include <functional>
+#include <iostream>
+#include <sstream>
+#include <thread>
+#include <vector>
+
+// Configuration constants
+constexpr uint64_t SLIDING_WINDOW_SIZE = 2;   // Time window for throughput calculation (in seconds)
+constexpr size_t CACHE_LINE_SIZE       = 64;  // Typical CPU cache line size for alignment
+
+// WorkerStats structure for tracking iteration counts per thread
+struct alignas(CACHE_LINE_SIZE) WorkerStats
+{
+  std::atomic<uint64_t> count{0};  // Count of iterations for a specific thread
+  char padding[CACHE_LINE_SIZE -
+               sizeof(std::atomic<uint64_t>)];  // Padding to ensure proper alignment
+};
+
+// StressTest class
+class Stress
+{
+public:
+  // Constructor
+  Stress(std::function<void()> func, size_t numThreads = std::thread::hardware_concurrency());
+
+  // Main function to start the stress test
+  void run();
+
+  // function to stop the test
+  void stop();
+
+private:
+  std::function<void()> func_;          // Function to be executed by each thread
+  std::vector<std::thread> threads_;    // Vector to hold worker threads
+  std::vector<WorkerStats> stats_;      // Vector to hold statistics for each thread
+  const size_t numThreads_;             // Number of threads to run
+  std::atomic<bool> stopFlag_{false};   // signal to stop the test
+  std::atomic<bool> readyFlag_{false};  // signal to start worker loops
+
+  // Worker thread function
+  void workerThread(size_t threadIndex);
+
+  // Monitoring thread function to calculate and display throughput
+  void monitorThroughput();
+
+  // Helper function to format numbers with commas for readability
+  static std::string formatNumber(uint64_t num);
+};

stress/metrics/CMakeLists.txt

@@ -0,0 +1,30 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+# Define the metrics executable
+add_executable(stress_metrics metrics.cc)
+
+# Link throughput library and OpenTelemetry Metrics API
+target_link_libraries(
+  stress_metrics PRIVATE stress opentelemetry_metrics # OpenTelemetry Metrics
+                                                      # SDK
+)
+
+# Include directories for throughput
+target_include_directories(stress_metrics
+                           PRIVATE ${CMAKE_SOURCE_DIR}/stress/common)
+
+# Set properties
+set_target_properties(
+  stress_metrics
+  PROPERTIES CXX_STANDARD 17
+             CXX_STANDARD_REQUIRED YES
+             CXX_EXTENSIONS NO)
+
+# Optional: Installation
+if(OPENTELEMETRY_INSTALL)
+  install(
+    TARGETS stress_metrics
+    EXPORT "${PROJECT_NAME}-target"
+    RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR})
+endif()