mirror of https://github.com/grpc/grpc.git
Create a CircularBuffer to store events generated by latent_see. This bounds the total number of generated events.
PiperOrigin-RevId: 658611718pull/37317/head
Vignesh Babu
4 months ago
committed by
Copybara-Service
parent
3e704c7552
commit
449d1b248f
16 changed files with 343 additions and 3 deletions
@ -0,0 +1,109 @@ |
||||
#ifndef GRPC_SRC_CORE_UTIL_RING_BUFFER_H_ |
||||
#define GRPC_SRC_CORE_UTIL_RING_BUFFER_H_ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <array> |
||||
#include <cstddef> |
||||
#include <iterator> |
||||
|
||||
#include "absl/types/optional.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
template <typename T, int kCapacity> |
||||
class RingBuffer { |
||||
public: |
||||
class RingBufferIterator { |
||||
public: |
||||
using iterator_category = std::forward_iterator_tag; |
||||
using value_type = const T; |
||||
using pointer = void; |
||||
using reference = void; |
||||
using difference_type = std::ptrdiff_t; |
||||
|
||||
RingBufferIterator& operator++() { |
||||
if (--size_ <= 0) { |
||||
head_ = 0; |
||||
size_ = 0; |
||||
buffer_ = nullptr; |
||||
} else { |
||||
head_ = (head_ + 1) % kCapacity; |
||||
} |
||||
return *this; |
||||
} |
||||
|
||||
RingBufferIterator operator++(int) { |
||||
RingBufferIterator tmp(*this); |
||||
operator++(); |
||||
return tmp; |
||||
} |
||||
|
||||
bool operator==(const RingBufferIterator& rhs) const { |
||||
return (buffer_ == rhs.buffer_ && head_ == rhs.head_ && |
||||
size_ == rhs.size_); |
||||
} |
||||
|
||||
bool operator!=(const RingBufferIterator& rhs) const { |
||||
return !operator==(rhs); |
||||
} |
||||
|
||||
T operator*() { return buffer_->data_[head_]; } |
||||
|
||||
RingBufferIterator() : buffer_(nullptr), head_(0), size_(0) {}; |
||||
RingBufferIterator(const RingBufferIterator& other) = default; |
||||
RingBufferIterator(const RingBuffer<T, kCapacity>* buffer) |
||||
: buffer_(buffer), head_(buffer->head_), size_(buffer->size_) { |
||||
if (!size_) { |
||||
buffer_ = nullptr; |
||||
} |
||||
} |
||||
|
||||
private: |
||||
friend class RingBuffer<T, kCapacity>; |
||||
const RingBuffer<T, kCapacity>* buffer_; |
||||
int head_ = 0; |
||||
int size_ = 0; |
||||
}; |
||||
|
||||
RingBuffer() = default; |
||||
|
||||
void Append(T data) { |
||||
if (size_ < kCapacity) { |
||||
data_[size_] = std::move(data); |
||||
size_++; |
||||
} else { |
||||
data_[head_] = std::move(data); |
||||
head_ = (head_ + 1) % kCapacity; |
||||
} |
||||
} |
||||
|
||||
// Returns the data of the first element in the buffer and removes it from
|
||||
// the buffer. If the buffer is empty, returns absl::nullopt.
|
||||
absl::optional<T> PopIfNotEmpty() { |
||||
if (!size_) return absl::nullopt; |
||||
T data = std::move(data_[head_]); |
||||
--size_; |
||||
head_ = (head_ + 1) % kCapacity; |
||||
return data; |
||||
} |
||||
|
||||
void Clear() { |
||||
head_ = 0; |
||||
size_ = 0; |
||||
} |
||||
|
||||
RingBufferIterator begin() const { return RingBufferIterator(this); } |
||||
|
||||
RingBufferIterator end() const { return RingBufferIterator(); } |
||||
|
||||
private: |
||||
friend class RingBufferIterator; |
||||
std::array<T, kCapacity> data_; |
||||
int head_ = 0; |
||||
int size_ = 0; |
||||
}; |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
#endif // GRPC_SRC_CORE_UTIL_RING_BUFFER_H_
|
||||
@ -0,0 +1,88 @@ |
||||
//
|
||||
//
|
||||
// Copyright 2015 gRPC authors.
|
||||
//
|
||||
// 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.
|
||||
//
|
||||
//
|
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "gtest/gtest.h" |
||||
#include "src/core/util/ring_buffer.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
constexpr int kBufferCapacity = 1000; |
||||
|
||||
TEST(RingBufferTest, BufferAppendPopTest) { |
||||
RingBuffer<int, kBufferCapacity> buffer; |
||||
EXPECT_FALSE(buffer.PopIfNotEmpty().has_value()); |
||||
|
||||
for (int i = 0; i < (3 * kBufferCapacity)/2; ++i) { |
||||
buffer.Append(i); |
||||
} |
||||
// Pop half of the elements. Elements in [kBufferCapacity / 2,
|
||||
// kBufferCapacity) are popped.
|
||||
int j = kBufferCapacity / 2; |
||||
for (int i = 0; i < kBufferCapacity / 2; ++i) { |
||||
EXPECT_EQ(buffer.PopIfNotEmpty(), j++); |
||||
} |
||||
EXPECT_EQ(j, kBufferCapacity); |
||||
// Iterate over the remaining elements.
|
||||
for (auto it = buffer.begin(); it != buffer.end(); ++it) { |
||||
EXPECT_EQ(*it, j++); |
||||
} |
||||
// Elements in [kBufferCapacity, (3 * kBufferCapacity) / 2) should be present.
|
||||
EXPECT_EQ(j, (3 * kBufferCapacity) / 2); |
||||
|
||||
// Append some more elements. The buffer should now have elements in
|
||||
// [kBufferCapacity, 2 * kBufferCapacity).
|
||||
for (int i = 0; i < kBufferCapacity / 2; ++i) { |
||||
buffer.Append(j++); |
||||
} |
||||
// Pop all the elements.
|
||||
j = kBufferCapacity; |
||||
while (true) { |
||||
auto ret = buffer.PopIfNotEmpty(); |
||||
if (!ret.has_value()) break; |
||||
EXPECT_EQ(*ret, j++); |
||||
} |
||||
EXPECT_EQ(j, 2 * kBufferCapacity); |
||||
} |
||||
|
||||
TEST(RingBufferTest, BufferAppendIterateTest) { |
||||
RingBuffer<int, kBufferCapacity> buffer; |
||||
for (int i = 0; i < 5 * kBufferCapacity; ++i) { |
||||
buffer.Append(i); |
||||
int j = std::max(0, i + 1 - kBufferCapacity); |
||||
// If i >= kBufferCapacity, check that the buffer contains only the last
|
||||
// kBufferCapacity elements [i + 1 - kBufferCapacity, i]. Otherwise check
|
||||
// that the buffer contains all elements from 0 to i.
|
||||
for (auto it = buffer.begin(); it != buffer.end(); ++it) { |
||||
EXPECT_EQ(*it, j++); |
||||
} |
||||
// Check that j was incremented at each step which implies that all the
|
||||
// required elements were present in the buffer.
|
||||
EXPECT_EQ(j, i + 1); |
||||
} |
||||
buffer.Clear(); |
||||
EXPECT_EQ(buffer.begin(), buffer.end()); |
||||
} |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
int main(int argc, char** argv) { |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
return RUN_ALL_TESTS(); |
||||
} |
||||
Loading…
Reference in new issue