New circular buffer using faa instead of cas to optimize performance

sdk/include/opentelemetry/sdk/common/circular_buffer.h

@@ -59,7 +59,9 @@ class CircularBuffer
     auto range = PeekImpl().Take(n);
     static_assert(noexcept(callback(range)), "callback not allowed to throw");
     tail_ += n;
+    // why not call back after tail_ += n
     callback(range);
+    // tail_ += n;
   }
 
   /**

sdk/include/opentelemetry/sdk/common/circular_buffer_range.h

@@ -75,7 +75,10 @@ class CircularBufferRange
    */
   CircularBufferRange Take(size_t n) const noexcept
   {
-    assert(n <= size());
+    // assert(n <= size());
+    if (n >= size()) {
+      return CircularBufferRange{first_,second_};
+    }
     if (first_.size() >= n)
     {
       return CircularBufferRange{nostd::span<T>{first_.data(), n}};

sdk/test/common/CMakeLists.txt

@@ -8,6 +8,7 @@ foreach(
   atomic_unique_ptr_test
   circular_buffer_range_test
   circular_buffer_test
+  optimize_circular_buffer_test
   attribute_utils_test
   attributemap_hash_test
   global_log_handle_test

sdk/test/common/baseline_circular_buffer.h

@@ -78,6 +78,11 @@ class BaselineCircularBuffer
     tail_ = head_;
   }
 
+  size_t size() noexcept {
+    std::lock_guard<std::mutex> lock_guard{mutex_};
+    return head_ - tail_;
+  }
+
 private:
   std::mutex mutex_;
   uint64_t head_{0};

sdk/test/common/circular_buffer_benchmark.cc

@@ -18,10 +18,12 @@
 #include "opentelemetry/sdk/common/circular_buffer.h"
 #include "opentelemetry/sdk/common/circular_buffer_range.h"
 #include "test/common/baseline_circular_buffer.h"
+#include "test/common/optmize_circular_buffer.h"
 
 using opentelemetry::sdk::common::AtomicUniquePtr;
 using opentelemetry::sdk::common::CircularBuffer;
 using opentelemetry::sdk::common::CircularBufferRange;
+using opentelemetry::sdk::common::OptimizedCircularBuffer;
 using opentelemetry::testing::BaselineCircularBuffer;
 
 const int N = 10000;
@@ -52,6 +54,20 @@ static uint64_t ConsumeBufferNumbers(CircularBuffer<uint64_t> &buffer) noexcept
   return result;
 }
 
+static uint64_t ConsumeBufferNumbers(OptimizedCircularBuffer<uint64_t> &buffer) noexcept
+{
+  uint64_t result = 0;
+  buffer.Consume(buffer.size(),
+                 [&](CircularBufferRange<AtomicUniquePtr<uint64_t>> &range) noexcept {
+                   range.ForEach([&](AtomicUniquePtr<uint64_t> &ptr) noexcept {
+                     result += *ptr;
+                     ptr.Reset();
+                     return true;
+                   });
+                 });
+  return result;
+}
+
 template <class Buffer>
 static void GenerateNumbersForThread(Buffer &buffer, int n, std::atomic<uint64_t> &sum) noexcept
 {
@@ -90,7 +106,9 @@ static void ConsumeNumbers(Buffer &buffer, uint64_t &sum, std::atomic<bool> &fin
   {
     sum += ConsumeBufferNumbers(buffer);
   }
-  sum += ConsumeBufferNumbers(buffer);
+  while (buffer.size()){
+    sum += ConsumeBufferNumbers(buffer);
+  }
 }
 
 template <class Buffer>
@@ -122,7 +140,7 @@ static void BM_BaselineBuffer(benchmark::State &state)
   }
 }
 
-BENCHMARK(BM_BaselineBuffer)->Arg(1)->Arg(2)->Arg(4);
+BENCHMARK(BM_BaselineBuffer)->Arg(1)->Arg(2)->Arg(4)->Arg(8)->Arg(16);
 
 static void BM_LockFreeBuffer(benchmark::State &state)
 {
@@ -136,6 +154,20 @@ static void BM_LockFreeBuffer(benchmark::State &state)
   }
 }
 
-BENCHMARK(BM_LockFreeBuffer)->Arg(1)->Arg(2)->Arg(4);
+BENCHMARK(BM_LockFreeBuffer)->Arg(1)->Arg(2)->Arg(4)->Arg(8)->Arg(16);
+
+static void BM_OptimizedBuffer(benchmark::State &state)
+{
+  const size_t max_elements = 500;
+  const int num_threads     = static_cast<int>(state.range(0));
+  const int n               = static_cast<int>(N / num_threads);
+  OptimizedCircularBuffer<uint64_t> buffer{max_elements};
+  for (auto _ : state)
+  {
+    RunSimulation(buffer, num_threads, n);
+  }
+}
+
+BENCHMARK(BM_OptimizedBuffer)->Arg(1)->Arg(2)->Arg(4)->Arg(8)->Arg(16);
 
 BENCHMARK_MAIN();

sdk/test/common/optimize_circular_buffer_test.cc

@@ -0,0 +1,171 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include <gtest/gtest.h>
+#include <stddef.h>
+#include <algorithm>
+#include <atomic>
+#include <cassert>
+#include <cstdint>
+#include <functional>
+#include <initializer_list>
+#include <memory>
+#include <random>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "opentelemetry/sdk/common/atomic_unique_ptr.h"
+#include "opentelemetry/sdk/common/circular_buffer.h"
+#include "test/common/optmize_circular_buffer.h"
+
+using opentelemetry::sdk::common::AtomicUniquePtr;
+using opentelemetry::sdk::common::CircularBufferRange;
+using opentelemetry::sdk::common::OptimizedCircularBuffer;
+
+static thread_local std::mt19937 RandomNumberGenerator{std::random_device{}()};
+
+static void GenerateRandomNumbers(OptimizedCircularBuffer<uint32_t> &buffer,
+                                  std::vector<uint32_t> &numbers,
+                                  int n)
+{
+  for (int i = 0; i < n; ++i)
+  {
+    auto value = static_cast<uint32_t>(RandomNumberGenerator());
+    std::unique_ptr<uint32_t> x{new uint32_t{value}};
+    if (buffer.Add(x))
+    {
+      numbers.push_back(value);
+    }
+  }
+}
+
+static void RunNumberProducers(OptimizedCircularBuffer<uint32_t> &buffer,
+                               std::vector<uint32_t> &numbers,
+                               int num_threads,
+                               int n)
+{
+  std::vector<std::vector<uint32_t>> thread_numbers(num_threads);
+  std::vector<std::thread> threads(num_threads);
+  for (int thread_index = 0; thread_index < num_threads; ++thread_index)
+  {
+    threads[thread_index] = std::thread{GenerateRandomNumbers, std::ref(buffer),
+                                        std::ref(thread_numbers[thread_index]), n};
+  }
+  for (auto &thread : threads)
+  {
+    thread.join();
+  }
+  for (int thread_index = 0; thread_index < num_threads; ++thread_index)
+  {
+    numbers.insert(numbers.end(), thread_numbers[thread_index].begin(),
+                   thread_numbers[thread_index].end());
+  }
+}
+
+void RunNumberConsumer(OptimizedCircularBuffer<uint32_t> &buffer,
+                       std::atomic<bool> &exit,
+                       std::vector<uint32_t> &numbers)
+{
+  while (true)
+  {
+    if (exit && buffer.Peek().empty())
+    {
+      return;
+    }
+    auto n = std::uniform_int_distribution<size_t>{0, buffer.Peek().size()}(RandomNumberGenerator);
+    buffer.Consume(n, [&](CircularBufferRange<AtomicUniquePtr<uint32_t>> range) noexcept {
+      assert(range.size() == n);
+      range.ForEach([&](AtomicUniquePtr<uint32_t> &ptr) noexcept {
+        assert(!ptr.IsNull());
+        numbers.push_back(*ptr);
+        ptr.Reset();
+        return true;
+      });
+    });
+  }
+}
+
+TEST(OptmizeCircularBufferTest, Add)
+{
+  OptimizedCircularBuffer<int> buffer{10};
+
+  std::unique_ptr<int> x{new int{11}};
+  EXPECT_TRUE(buffer.Add(x));
+  EXPECT_EQ(x, nullptr);
+  auto range = buffer.Peek();
+  EXPECT_EQ(range.size(), 1);
+  range.ForEach([](const AtomicUniquePtr<int> &y) {
+    EXPECT_EQ(*y, 11);
+    return true;
+  });
+}
+
+TEST(OptmizeCircularBufferTest, Clear)
+{
+  OptimizedCircularBuffer<int> buffer{10};
+
+  std::unique_ptr<int> x{new int{11}};
+  EXPECT_TRUE(buffer.Add(x));
+  EXPECT_EQ(x, nullptr);
+  buffer.Clear();
+  EXPECT_TRUE(buffer.empty());
+}
+
+TEST(OptmizeCircularBufferTest, AddOnFull)
+{
+  OptimizedCircularBuffer<int> buffer{10};
+  for (int i = 0; i < static_cast<int>(buffer.max_size()); ++i)
+  {
+    std::unique_ptr<int> x{new int{i}};
+    EXPECT_TRUE(buffer.Add(x));
+  }
+  std::unique_ptr<int> x{new int{33}};
+  EXPECT_FALSE(buffer.Add(x));
+  EXPECT_NE(x, nullptr);
+  EXPECT_EQ(*x, 33);
+}
+
+TEST(OptmizeCircularBufferTest, Consume)
+{
+  OptimizedCircularBuffer<int> buffer{10};
+  for (int i = 0; i < static_cast<int>(buffer.max_size()); ++i)
+  {
+    std::unique_ptr<int> x{new int{i}};
+    EXPECT_TRUE(buffer.Add(x));
+  }
+  int count = 0;
+  buffer.Consume(5, [&](CircularBufferRange<AtomicUniquePtr<int>> range) noexcept {
+    range.ForEach([&](AtomicUniquePtr<int> &ptr) {
+      EXPECT_EQ(*ptr, count++);
+      ptr.Reset();
+      return true;
+    });
+  });
+  EXPECT_EQ(count, 5);
+}
+
+TEST(OptmizeCircularBufferTest, Simulation)
+{
+  const int num_producer_threads = 4;
+  const int n                    = 25000;
+  for (size_t max_size : {1, 2, 10, 50, 100, 1000})
+  {
+    OptimizedCircularBuffer<uint32_t> buffer{max_size};
+    std::vector<uint32_t> producer_numbers;
+    std::vector<uint32_t> consumer_numbers;
+    auto producers = std::thread{RunNumberProducers, std::ref(buffer), std::ref(producer_numbers),
+                                 num_producer_threads, n};
+    std::atomic<bool> exit{false};
+    auto consumer = std::thread{RunNumberConsumer, std::ref(buffer), std::ref(exit),
+                                std::ref(consumer_numbers)};
+    producers.join();
+    exit = true;
+    consumer.join();
+    std::sort(producer_numbers.begin(), producer_numbers.end());
+    std::sort(consumer_numbers.begin(), consumer_numbers.end());
+
+    EXPECT_EQ(producer_numbers.size(), consumer_numbers.size());
+    EXPECT_EQ(producer_numbers, consumer_numbers);
+  }
+}