1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
|
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0
#include <stddef.h>
#include <atomic>
#include <cstdint>
#include <exception>
#include <functional>
#include <iostream>
#include <memory>
#include <random>
#include <thread>
#include <vector>
#include "benchmark/benchmark.h"
#include "opentelemetry/sdk/common/atomic_unique_ptr.h"
#include "opentelemetry/sdk/common/circular_buffer.h"
#include "opentelemetry/sdk/common/circular_buffer_range.h"
#include "test/common/baseline_circular_buffer.h"
using opentelemetry::sdk::common::AtomicUniquePtr;
using opentelemetry::sdk::common::CircularBuffer;
using opentelemetry::sdk::common::CircularBufferRange;
using opentelemetry::testing::BaselineCircularBuffer;
const int N = 10000;
static uint64_t ConsumeBufferNumbers(BaselineCircularBuffer<uint64_t> &buffer) noexcept
{
uint64_t result = 0;
buffer.Consume([&](std::unique_ptr<uint64_t> &&x) {
result += *x;
x.reset();
});
return result;
}
static uint64_t ConsumeBufferNumbers(CircularBuffer<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
{
thread_local std::mt19937_64 random_number_generator{std::random_device{}()};
for (int i = 0; i < n; ++i)
{
auto x = random_number_generator();
std::unique_ptr<uint64_t> element{new uint64_t{x}};
if (buffer.Add(element))
{
sum += x;
}
}
}
template <class Buffer>
static uint64_t GenerateNumbers(Buffer &buffer, int num_threads, int n) noexcept
{
std::atomic<uint64_t> sum{0};
std::vector<std::thread> threads(num_threads);
for (auto &thread : threads)
{
thread = std::thread{GenerateNumbersForThread<Buffer>, std::ref(buffer), n, std::ref(sum)};
}
for (auto &thread : threads)
{
thread.join();
}
return sum;
}
template <class Buffer>
static void ConsumeNumbers(Buffer &buffer, uint64_t &sum, std::atomic<bool> &finished) noexcept
{
while (!finished)
{
sum += ConsumeBufferNumbers(buffer);
}
sum += ConsumeBufferNumbers(buffer);
}
template <class Buffer>
static void RunSimulation(Buffer &buffer, int num_threads, int n) noexcept
{
std::atomic<bool> finished{false};
uint64_t consumer_sum{0};
std::thread consumer_thread{ConsumeNumbers<Buffer>, std::ref(buffer), std::ref(consumer_sum),
std::ref(finished)};
uint64_t producer_sum = GenerateNumbers(buffer, num_threads, n);
finished = true;
consumer_thread.join();
if (consumer_sum != producer_sum)
{
std::cerr << "Sumulation failed: consumer_sum != producer_sum\n";
std::terminate();
}
}
static void BM_BaselineBuffer(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);
BaselineCircularBuffer<uint64_t> buffer{max_elements};
for (auto _ : state)
{
RunSimulation(buffer, num_threads, n);
}
}
BENCHMARK(BM_BaselineBuffer)->Arg(1)->Arg(2)->Arg(4);
static void BM_LockFreeBuffer(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);
CircularBuffer<uint64_t> buffer{max_elements};
for (auto _ : state)
{
RunSimulation(buffer, num_threads, n);
}
}
BENCHMARK(BM_LockFreeBuffer)->Arg(1)->Arg(2)->Arg(4);
BENCHMARK_MAIN();
|
