diff --git a/benchmark/time_queue.cpp b/benchmark/time_queue.cpp index d8fcf96..d25615d 100644 --- a/benchmark/time_queue.cpp +++ b/benchmark/time_queue.cpp @@ -1,16 +1,33 @@ #include +#include #include +#include + +struct mock_tick_service : timeq::tick_service +{ + explicit mock_tick_service(std::chrono::milliseconds ticks) + : ticks(ticks) + { + } + + std::chrono::microseconds get() const override { return ticks; } + + std::chrono::microseconds ticks; +}; + static auto service = std::make_shared(); constexpr size_t kIterations = 100'000'000; +template static void -BM_TimeQueue_Construct(benchmark::State& state) +Construct(benchmark::State& state) { + const std::size_t duration = state.range(0); for (auto _ : state) { - auto tq = timeq::time_queue(300, 1, service, kIterations); + auto tq = TimeQueue(duration, 1, service); std::size_t size = tq.size(); benchmark::DoNotOptimize(size); @@ -18,26 +35,45 @@ BM_TimeQueue_Construct(benchmark::State& state) } } +template static void -BM_TimeQueue_Push(benchmark::State& state) +Push(benchmark::State& state) { - timeq::time_queue tq(state.range(0), 1, service, kIterations); + TimeQueue tq(state.range(0), 1, service, kIterations); int64_t items_count = 0; for (auto _ : state) { ++items_count; - tq.push(items_count, 20); + tq.push({}, 20); } state.SetItemsProcessed(items_count); } +template +static void +FirstPushAfterIdle(benchmark::State& state) +{ + const std::size_t duration = state.range(0); + auto aged_service = std::make_shared(std::chrono::milliseconds(duration - 1)); + TimeQueue tq(duration, 1, aged_service, 1); + + for (auto _ : state) { + tq.push({}, duration); + + std::size_t size = tq.size(); + benchmark::DoNotOptimize(size); + benchmark::ClobberMemory(); + } +} + +template static void -BM_TimeQueue_Pop(benchmark::State& state) +Pop(benchmark::State& state) { - timeq::time_queue tq(state.range(0), 1, service, kIterations); + TimeQueue tq(state.range(0), 1, service, kIterations); for (size_t i = 0; i < kIterations; ++i) { - tq.push(i, 10); + tq.push({}, 10); } int64_t items_count = 0; @@ -49,12 +85,13 @@ BM_TimeQueue_Pop(benchmark::State& state) state.SetItemsProcessed(items_count); } +template static void -BM_TimeQueue_Front(benchmark::State& state) +Front(benchmark::State& state) { - timeq::time_queue tq(state.range(0), 1, service, kIterations); + TimeQueue tq(state.range(0), 1, service, kIterations); for (size_t i = 0; i < kIterations; ++i) { - tq.push(i, 15); + tq.push({}, 15); } int64_t items_count = 0; @@ -68,12 +105,13 @@ BM_TimeQueue_Front(benchmark::State& state) state.SetItemsProcessed(items_count); } +template static void -BM_TimeQueue_PopFront(benchmark::State& state) +PopFront(benchmark::State& state) { - timeq::time_queue tq(state.range(0), 1, service, kIterations); + TimeQueue tq(state.range(0), 1, service, kIterations); for (size_t i = 0; i < kIterations; ++i) { - tq.push(i, 15); + tq.push({}, 15); } int64_t items_count = 0; @@ -87,12 +125,13 @@ BM_TimeQueue_PopFront(benchmark::State& state) state.SetItemsProcessed(items_count); } +template static void -BM_TimeQueue_Size(benchmark::State& state) +Size(benchmark::State& state) { - timeq::time_queue tq(300, 1, service, kIterations); + TimeQueue tq(300, 1, service, kIterations); for (size_t i = 0; i < kIterations; ++i) { - tq.push(i, 10); + tq.push({}, 10); } for (auto _ : state) { @@ -102,12 +141,13 @@ BM_TimeQueue_Size(benchmark::State& state) } } +template static void -BM_TimeQueue_Empty(benchmark::State& state) +Empty(benchmark::State& state) { - timeq::time_queue tq(300, 1, service, kIterations); + TimeQueue tq(300, 1, service, kIterations); for (size_t i = 0; i < kIterations; ++i) { - tq.push(i, 10); + tq.push({}, 10); } for (auto _ : state) { @@ -117,15 +157,16 @@ BM_TimeQueue_Empty(benchmark::State& state) } } +template static void -BM_TimeQueue_PushAndPopLoaded(benchmark::State& state) +PushAndPopLoaded(benchmark::State& state) { - timeq::time_queue tq(state.range(0), 1, service, kIterations); + TimeQueue tq(state.range(0), 1, service, kIterations); int64_t items_count = 0; for (auto _ : state) { ++items_count; - tq.push(items_count, 1000); + tq.push({}, 1000); // Simulate load by not popping all items if (items_count % 100 == 0) { @@ -139,15 +180,16 @@ BM_TimeQueue_PushAndPopLoaded(benchmark::State& state) state.SetItemsProcessed(items_count); } +template static void -BM_TimeQueue_PushAndPop_Interval_1ms(benchmark::State& state) +PushAndPop_Interval_1ms(benchmark::State& state) { - timeq::time_queue tq(state.range(0), 1, service, kIterations); + TimeQueue tq(state.range(0), 1, service, kIterations); int64_t items_count = 0; for (auto _ : state) { ++items_count; - tq.push(items_count, 1000); + tq.push({}, 1000); auto elem = tq.front(); tq.pop(); @@ -159,15 +201,16 @@ BM_TimeQueue_PushAndPop_Interval_1ms(benchmark::State& state) state.SetItemsProcessed(items_count); } +template static void -BM_TimeQueue_PushAndPop_Interval_125ms(benchmark::State& state) +PushAndPop_Interval_125ms(benchmark::State& state) { - timeq::time_queue tq(state.range(0), 125, service, kIterations); + TimeQueue tq(state.range(0), 125, service, kIterations); int64_t items_count = 0; for (auto _ : state) { ++items_count; - tq.push(items_count, 1000); + tq.push({}, 1000); auto elem = tq.front(); tq.pop(); @@ -179,13 +222,72 @@ BM_TimeQueue_PushAndPop_Interval_125ms(benchmark::State& state) state.SetItemsProcessed(items_count); } -BENCHMARK(BM_TimeQueue_Construct)->Arg(300); -BENCHMARK(BM_TimeQueue_Push)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); -BENCHMARK(BM_TimeQueue_Pop)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); -BENCHMARK(BM_TimeQueue_Front)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); -BENCHMARK(BM_TimeQueue_PopFront)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); -BENCHMARK(BM_TimeQueue_Size)->Iterations(kIterations); -BENCHMARK(BM_TimeQueue_Empty)->Iterations(kIterations); -BENCHMARK(BM_TimeQueue_PushAndPopLoaded)->Arg(5000)->Arg(1'000'000); -BENCHMARK(BM_TimeQueue_PushAndPop_Interval_1ms)->Arg(5000)->Arg(1'000'000); -BENCHMARK(BM_TimeQueue_PushAndPop_Interval_125ms)->Arg(5000)->Arg(1'000'000); +struct TrivialType +{}; + +static_assert(std::is_trivially_copyable_v); + +struct NonTrivialType +{ + NonTrivialType() {} + ~NonTrivialType() {} +}; + +static_assert(!std::is_trivially_copyable_v); + +using namespace timeq; + +BENCHMARK(Construct>)->Arg(300)->Arg(1'000'000); +BENCHMARK(Construct>)->Arg(300)->Arg(1'000'000); +BENCHMARK(Construct>)->Arg(300)->Arg(1'000'000); +BENCHMARK(Construct>)->Arg(300)->Arg(1'000'000); + +BENCHMARK(Push>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); +BENCHMARK(Push>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); +BENCHMARK(Push>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); +BENCHMARK(Push>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); + +BENCHMARK(FirstPushAfterIdle>)->Arg(300)->Arg(1'000'000); +BENCHMARK(FirstPushAfterIdle>)->Arg(300)->Arg(1'000'000); +BENCHMARK(FirstPushAfterIdle>)->Arg(300)->Arg(1'000'000); +BENCHMARK(FirstPushAfterIdle>)->Arg(300)->Arg(1'000'000); + +BENCHMARK(Pop>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); +BENCHMARK(Pop>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); +BENCHMARK(Pop>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); +BENCHMARK(Pop>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); + +BENCHMARK(Front>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); +BENCHMARK(Front>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); +BENCHMARK(Front>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); +BENCHMARK(Front>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); + +BENCHMARK(PopFront>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); +BENCHMARK(PopFront>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); +BENCHMARK(PopFront>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); +BENCHMARK(PopFront>)->Iterations(kIterations)->Arg(300)->Arg(1'000'000); + +BENCHMARK(Size>)->Iterations(kIterations); +BENCHMARK(Size>)->Iterations(kIterations); +BENCHMARK(Size>)->Iterations(kIterations); +BENCHMARK(Size>)->Iterations(kIterations); + +BENCHMARK(Empty>)->Iterations(kIterations); +BENCHMARK(Empty>)->Iterations(kIterations); +BENCHMARK(Empty>)->Iterations(kIterations); +BENCHMARK(Empty>)->Iterations(kIterations); + +BENCHMARK(PushAndPopLoaded>)->Arg(5000)->Arg(1'000'000); +BENCHMARK(PushAndPopLoaded>)->Arg(5000)->Arg(1'000'000); +BENCHMARK(PushAndPopLoaded>)->Arg(5000)->Arg(1'000'000); +BENCHMARK(PushAndPopLoaded>)->Arg(5000)->Arg(1'000'000); + +BENCHMARK(PushAndPop_Interval_1ms>)->Arg(5000)->Arg(1'000'000); +BENCHMARK(PushAndPop_Interval_1ms>)->Arg(5000)->Arg(1'000'000); +BENCHMARK(PushAndPop_Interval_1ms>)->Arg(5000)->Arg(1'000'000); +BENCHMARK(PushAndPop_Interval_1ms>)->Arg(5000)->Arg(1'000'000); + +BENCHMARK(PushAndPop_Interval_125ms>)->Arg(5000)->Arg(1'000'000); +BENCHMARK(PushAndPop_Interval_125ms>)->Arg(5000)->Arg(1'000'000); +BENCHMARK(PushAndPop_Interval_125ms>)->Arg(5000)->Arg(1'000'000); +BENCHMARK(PushAndPop_Interval_125ms>)->Arg(5000)->Arg(1'000'000); diff --git a/include/timeq/fast_time_queue.h b/include/timeq/fast_time_queue.h new file mode 100644 index 0000000..db16db0 --- /dev/null +++ b/include/timeq/fast_time_queue.h @@ -0,0 +1,376 @@ +// SPDX-FileCopyrightText: Copyright (c) 2023 Cisco Systems +// SPDX-License-Identifier: BSD-2-Clause + +/** + * fast_time_queue.h + * + * Description: + * A time based queue, where the length of the queue is a duration, + * divided into buckets based on a given time interval. As time + * progresses, buckets in the past are cleared, and the main queue + * is updated so that the front only returns a valid object that + * has not expired. To improve performance, buckets are only cleared + * on push or pop operations. Thus, buckets in the past can be + * cleared in bulk based on how many we should have advanced since + * the last time we updated. + * + * Portability Issues: + * None. + */ + +#pragma once + +#include "tick_service.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace timeq { +#define FORCE_INLINE __attribute__((always_inline)) + + /** + * @brief Aging element FIFO queue. + * + * @details Time based queue that maintains the push/pop order, but expires + * older values given a specific ttl. + * + * @tparam T The element type to be stored. + */ + template + class fast_time_queue + { + protected: + using tick_type = tick_service::tick_type; + using index_type = std::uint64_t; + + struct queue_value_type + { + T value; + tick_type expiry_tick; + tick_type wait_for_tick; + }; + + using queue_type = std::vector; + + public: + template + struct element + { + /// Value of front object + std::optional value; + + /// Number of items expired before on this front access + std::uint32_t expired{ 0 }; + }; + + template + struct element + { + /// Value of front object + std::optional> value; + + /// Number of items expired before on this front access + std::uint32_t expired{ 0 }; + }; + +#if defined(__clang__) || (defined(__GNUC__) && !defined(__clang__) && __GNUC__ >= 12) + template + element(U) -> element; +#endif + + using value_type = element; + using reference = element; + + /** + * @brief Construct a fast_time_queue with defaults or supplied parameters + * + * @param duration Duration of the queue in milliseconds. Value must be > 0, and != interval. + * @param interval Interval of ticks in milliseconds. Value must be > 0, < duration, duration % interval == 0. + * @param tick_service Shared pointer to tick_service. + * @param initial_queue_size Initial size of the queue to reserve. + * + * @throws std::invalid_argument If the duration or interval do not meet requirements or the tick_service is + * null. + */ + fast_time_queue(std::size_t duration, + std::size_t interval, + std::shared_ptr tick_service, + std::size_t initial_queue_size) + : _duration{ duration } + , _interval{ interval } + , _tick_service(std::move(tick_service)) + { + if (duration == 0 || interval == 0 || duration % interval != 0 || duration == interval) { + throw std::invalid_argument("Invalid fast_time_queue constructor args"); + } + + if (!_tick_service) { + throw std::invalid_argument("Tick service cannot be null"); + } + + _queue.reserve(initial_queue_size); + } + + /** + * @brief Construct a fast_time_queue with defaults or supplied parameters + * + * @param duration Duration of the queue in milliseconds. Value must be > 0, and != interval. + * @param interval Interval of ticks in milliseconds. Must be > 0, < duration, duration % interval == 0. + * @param tick_service Shared pointer to tick_service service. + * + * @throws std::invalid_argument If the duration or interval do not meet requirements or If the tick_service is + * null. + */ + fast_time_queue(std::size_t duration, std::size_t interval, std::shared_ptr tick_service) + : fast_time_queue(duration, interval, std::move(tick_service), duration / interval) + { + } + + fast_time_queue() = delete; + fast_time_queue(const fast_time_queue&) = default; + fast_time_queue(fast_time_queue&&) noexcept = default; + + fast_time_queue& operator=(const fast_time_queue&) = default; + fast_time_queue& operator=(fast_time_queue&&) noexcept = default; + + /** + * @brief pushes a new value onto the queue with a time-to-live. + * + * @param value The value to push onto the queue. + * @param ttl Time to live for an object in milliseconds + * @param delay_ttl pop wait Time to live for an object in milliseconds + * This will cause pop to be delayed by this TTL value + * + * @throws std::invalid_argument If ttl is greater than duration. + */ + FORCE_INLINE void push(const T& value, std::size_t ttl, std::size_t delay_ttl = 0) + { + internal_push(value, ttl, delay_ttl); + } + + /** + * @brief pushes a new value onto the queue with a time-to-live. + * + * @param value The value to push onto the queue. + * @param ttl Time to live for an object in milliseconds + * @param delay_ttl pop wait Time to live for an object in milliseconds + * This will cause pop to be delayed by this TTL value + * + * @throws std::invalid_argument If ttl is greater than duration. + */ + FORCE_INLINE void push(T&& value, std::size_t ttl, std::size_t delay_ttl = 0) + { + internal_push(std::move(value), ttl, delay_ttl); + } + + /** + * @brief Pop (increment) front + * + * @details This method should be called after front when the object is + * processed. This will move the queue forward. If at the end of the queue, + * it'll be cleared and reset. + */ + FORCE_INLINE void pop() noexcept + { + if (_queue.empty()) { + return; + } + + if (++_queue_index >= _queue.size()) { + clear(); + return; + } + + if (_queue_index >= size()) { + compact_consumed_queue(); + } + } + + /** + * @brief Returns the most valid front of the queue without popping. + * + * @returns Element of the front value + */ + FORCE_INLINE reference front() + { + const tick_type ticks = advance(); + + if (_queue.empty()) { + return { std::nullopt, 0 }; + } + + std::uint32_t expired = 0; + + while (_queue_index < _queue.size()) { + auto& [value, expiry_tick, pop_wait_ttl] = _queue.at(_queue_index); + + if (ticks >= expiry_tick) { + expired++; + _queue_index++; + continue; + } + + if (pop_wait_ttl > ticks) { + return { std::nullopt, expired }; + } + + return { value, expired }; + } + + clear(); + + return { std::nullopt, expired }; + } + + /** + * @brief Pops (removes) the front of the queue. + * + * @returns element of the popped value + */ + [[nodiscard]] FORCE_INLINE value_type pop_front() + { + auto&& [value, expired] = front(); + value_type elem{ value.has_value() ? std::make_optional(std::move(value->get())) : std::nullopt, expired }; + + if (elem.value.has_value()) { + pop(); + } + + return elem; + } + + FORCE_INLINE constexpr std::size_t size() const noexcept { return _queue.size() - _queue_index; } + + FORCE_INLINE constexpr bool empty() const noexcept { return _queue.empty() || _queue_index >= _queue.size(); } + + FORCE_INLINE void update() { [[maybe_unused]] auto _ = advance(); } + + /** + * @brief Clear/reset the queue to no objects + */ + FORCE_INLINE void clear() noexcept + { + if (!_queue.empty()) { + _queue.clear(); + } + + _queue_index = 0; + _last_tick_queue_cleared = _current_ticks; + } + + protected: + /** + * @brief Based on current time, adjust and move the bucket index with time + * (sliding window) + * + * @returns Current tick value at time of advance + */ + [[nodiscard]] FORCE_INLINE tick_type advance() + { + const tick_type new_tick_count = + std::chrono::duration_cast(_tick_service->get()).count(); + const tick_type delta = new_tick_count - _current_ticks; + _current_ticks = new_tick_count; + + if (delta < _interval) { + return _current_ticks; + } + + if (delta > _duration) { + clear(); + return _current_ticks; + } + + if (_current_ticks - _last_tick_queue_cleared > _duration && !_queue.empty()) { + compact_consumed_queue(); + _last_tick_queue_cleared = _current_ticks; + } + + return _current_ticks; + } + + /** + * @brief pushes new element onto the queue and adds it to future bucket. + * + * @details Internal definition of push. pushes value into specified + * bucket, and then emplaces the location info into the queue. + * + * @param value The value to push onto the queue. + * @param ttl Time to live for an object in milliseconds + * @param delay_ttl pop wait Time to live for an object in milliseconds + * This will cause pop to be delayed by this TTL value + * + * @throws std::invalid_argument If ttl is greater than duration. + */ + FORCE_INLINE void internal_push(auto&& value, std::size_t ttl, std::size_t delay_ttl) + { + if (ttl > _duration) { + throw std::invalid_argument("TTL is greater than max duration"); + } + + if (ttl == 0) { + ttl = _duration; + } + + const tick_type ticks = advance(); + + const tick_type expiry_tick = ticks + ttl; + + _queue.emplace_back(std::forward(value), expiry_tick, ticks + delay_ttl); + } + + FORCE_INLINE void compact_consumed_queue() noexcept + { + if (_queue_index == 0) { + return; + } + + if (_queue_index >= _queue.size()) { + clear(); + return; + } + + const std::size_t logical_size = _queue.size() - _queue_index; + + if constexpr (std::is_trivially_copyable_v) { + std::memmove(_queue.data(), _queue.data() + _queue_index, logical_size * sizeof(queue_value_type)); + } else { + std::move(_queue.begin() + static_cast(_queue_index), _queue.end(), _queue.begin()); + } + + _queue.resize(logical_size); + _queue_index = 0; + } + + protected: + /// The duration in ticks of the entire queue. + const std::size_t _duration; + + /// The interval at which buckets are cleared in ticks. + const std::size_t _interval; + + /// The index of the first valid item in the queue. + index_type _queue_index{ 0 }; + + /// Last calculated tick value. + tick_type _current_ticks{ 0 }; + + /// Last calculated tick value when queue was cleared. + tick_type _last_tick_queue_cleared{ 0 }; + + /// The FIFO ordered queue of values as they were inserted. + queue_type _queue; + + /// Tick service for calculating new tick and jumps in time. + std::shared_ptr _tick_service; + }; + +#undef FORCE_INLINE +}; // namespace timeq diff --git a/include/timeq/tick_service.h b/include/timeq/tick_service.h index 62f12be..7460e41 100644 --- a/include/timeq/tick_service.h +++ b/include/timeq/tick_service.h @@ -95,10 +95,8 @@ namespace timeq { std::chrono::duration_cast(now - prev_time).count(); std::this_thread::sleep_for(std::chrono::microseconds(_sleep_delay_us)); - if (delta >= _sleep_delay_us) { - _ticks.fetch_add(delta, std::memory_order_relaxed); - prev_time = now; - } + _ticks.fetch_add(delta, std::memory_order_relaxed); + prev_time = now; } } diff --git a/include/timeq/time_queue.h b/include/timeq/time_queue.h index 9885f5a..045a97c 100644 --- a/include/timeq/time_queue.h +++ b/include/timeq/time_queue.h @@ -23,6 +23,7 @@ #include "tick_service.h" #include +#include #include #include #include @@ -52,6 +53,7 @@ namespace timeq { struct queue_value_type { + queue_value_type() = default; queue_value_type(bucket_type& bucket, index_type value_index, tick_type expiry_tick, @@ -122,7 +124,7 @@ namespace timeq { , _interval{ interval } , _tick_service(std::move(tick_service)) { - if (duration == 0 || duration % interval != 0 || duration == interval) { + if (duration == 0 || interval == 0 || duration % interval != 0 || duration == interval) { throw std::invalid_argument("Invalid time_queue constructor args"); } @@ -194,11 +196,18 @@ namespace timeq { */ FORCE_INLINE void pop() noexcept { - if (_queue.empty() || ++_queue_index < _queue.size()) { + if (_queue.empty()) { return; } - clear(); + if (++_queue_index >= _queue.size()) { + clear(); + return; + } + + if (_queue_index >= size()) { + compact_consumed_queue(); + } } /** @@ -219,7 +228,7 @@ namespace timeq { while (_queue_index < _queue.size()) { auto& [bucket, value_index, expiry_tick, pop_wait_ttl] = _queue.at(_queue_index); - if (ticks > expiry_tick || value_index >= bucket->size()) { + if (ticks >= expiry_tick || value_index >= bucket->size()) { expired++; _queue_index++; continue; @@ -265,22 +274,26 @@ namespace timeq { */ FORCE_INLINE void clear() noexcept { - if (_queue.empty()) { - return; + if (!_queue.empty()) { + _queue.clear(); + _buckets.clear(); } - _queue.clear(); - - _buckets.clear(); - _queue_index = _bucket_index = 0; _last_tick_queue_cleared = _current_ticks; + _last_bucket_advance_tick = _current_ticks; } protected: + [[nodiscard]] FORCE_INLINE constexpr index_type bucket_count() const noexcept + { + // The spare bucket prevents max-duration TTLs pushed mid-interval from being cleared early. + return (_duration / _interval) + 1; + } + [[nodiscard]] FORCE_INLINE constexpr index_type get_future_bucket_index(index_type delta) { - return (_bucket_index + delta) % (_duration / _interval); + return (_bucket_index + delta) % bucket_count(); } /** @@ -293,34 +306,36 @@ namespace timeq { { const tick_type new_tick_count = std::chrono::duration_cast(_tick_service->get()).count(); - const tick_type delta = new_tick_count - _current_ticks; + const tick_type delta = new_tick_count - _last_bucket_advance_tick; _current_ticks = new_tick_count; - if (delta == 0) { + if (delta < _interval) { return new_tick_count; } - if (delta >= _duration) { + const auto buckets_elapsed = delta / _interval; + + if (buckets_elapsed >= bucket_count()) { clear(); return new_tick_count; } - for (std::size_t i = 0; i < delta / _interval; ++i) { - bucket_type& bucket = _buckets[get_future_bucket_index(i)]; + for (std::size_t i = 0; i < buckets_elapsed; ++i) { + const index_type future_index = get_future_bucket_index(i); + if (!_buckets.contains(future_index)) { + continue; + } + + bucket_type& bucket = _buckets.at(future_index); bucket.clear(); bucket.shrink_to_fit(); } - _bucket_index = get_future_bucket_index(delta / _interval); - - if (_last_tick_queue_cleared > _duration && !_queue.empty()) { - if (_queue_index >= _queue.size()) { - _queue.clear(); - } else { - _queue.erase(_queue.begin(), std::next(_queue.begin(), _queue_index)); - } + _bucket_index = get_future_bucket_index(buckets_elapsed); + _last_bucket_advance_tick += delta; - _queue_index = 0; + if (_current_ticks - _last_tick_queue_cleared > _duration && !_queue.empty()) { + compact_consumed_queue(); _last_tick_queue_cleared = _current_ticks; } @@ -350,20 +365,38 @@ namespace timeq { ttl = _duration; } - auto relative_ttl = ttl / _interval; - const tick_type ticks = advance(); const tick_type expiry_tick = ticks + ttl; - const index_type future_index = get_future_bucket_index(relative_ttl - 1); + const tick_type ticks_until_expiry = expiry_tick - _last_bucket_advance_tick; + const index_type intervals_until_clear = (ticks_until_expiry + _interval - 1) / _interval; + const index_type future_index = get_future_bucket_index(intervals_until_clear - 1); bucket_type& bucket = _buckets[future_index]; - bucket.emplace_back(value); + bucket.emplace_back(std::forward(value)); _queue.emplace_back(bucket, bucket.size() - 1, expiry_tick, ticks + delay_ttl); } + FORCE_INLINE void compact_consumed_queue() noexcept + { + if (_queue_index == 0) { + return; + } + + if (_queue_index >= _queue.size()) { + clear(); + return; + } + + const std::size_t logical_size = _queue.size() - _queue_index; + std::memmove(_queue.data(), _queue.data() + _queue_index, logical_size * sizeof(queue_value_type)); + + _queue.resize(logical_size); + _queue_index = 0; + } + protected: /// The duration in ticks of the entire queue. const std::size_t _duration; @@ -383,8 +416,11 @@ namespace timeq { /// Last calculated tick value when queue was cleared. tick_type _last_tick_queue_cleared{ 0 }; + /// Last tick represented by the current bucket index. + tick_type _last_bucket_advance_tick{ 0 }; + /// The memory storage for all elements to be managed. - std::map _buckets; + std::map _buckets; /// The FIFO ordered queue of values as they were inserted. queue_type _queue; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 60b9252..6c884f3 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -9,6 +9,7 @@ CPMAddPackage("gh:google/googletest@1.17.0") add_executable(timeq_test time_queue.cpp + fast_time_queue.cpp ) set_target_properties(timeq_test PROPERTIES diff --git a/tests/fast_time_queue.cpp b/tests/fast_time_queue.cpp new file mode 100644 index 0000000..9458b86 --- /dev/null +++ b/tests/fast_time_queue.cpp @@ -0,0 +1,206 @@ +#include + +#include +#include + +using namespace timeq; + +struct test_tick_service : public tick_service +{ + test_tick_service() = default; + virtual ~test_tick_service() = default; + + std::chrono::microseconds get() const override { return std::chrono::microseconds(ticks); } + + std::chrono::microseconds ticks; +}; + +static auto tick_manager = std::make_shared(); + +template +class inspectable_fast_time_queue : public fast_time_queue +{ + public: + using fast_time_queue::fast_time_queue; + + std::size_t raw_queue_size() const noexcept { return this->_queue.size(); } + + std::size_t queue_index() const noexcept { return this->_queue_index; } +}; + +TEST(fast_time_queue, Construction) +{ + ASSERT_NO_THROW(fast_time_queue(10, 1, tick_manager)); + ASSERT_THROW(fast_time_queue(10, 1, nullptr), std::invalid_argument); + ASSERT_THROW(fast_time_queue(0, 1, tick_manager), std::invalid_argument); +} + +TEST(fast_time_queue, PushAndExpire) +{ + fast_time_queue tq(10, 1, tick_manager); + + tq.push(123, 2); + ASSERT_EQ(tq.front().value.value(), 123); + + tick_manager->ticks += std::chrono::milliseconds(1); + ASSERT_EQ(tq.front().value.value(), 123); + + tick_manager->ticks += std::chrono::milliseconds(1); + ASSERT_FALSE(tq.front().value.has_value()); +} + +TEST(fast_time_queue, PushMidIntervalWithMaxTtlExpiresAtTtl) +{ + auto tick_manager = std::make_shared(); + fast_time_queue tq(5000, 500, tick_manager); + + tick_manager->ticks = std::chrono::milliseconds(100); + tq.push(123, 5000); + + for (int ticks = 200; ticks <= 5000; ticks += 100) { + tick_manager->ticks = std::chrono::milliseconds(ticks); + tq.update(); + } + + tick_manager->ticks = std::chrono::milliseconds(5099); + ASSERT_EQ(tq.front().value.value(), 123); + + tick_manager->ticks = std::chrono::milliseconds(5100); + ASSERT_FALSE(tq.front().value.has_value()); +} + +TEST(fast_time_queue, SubIntervalUpdatesStillAdvanceBuckets) +{ + auto tick_manager = std::make_shared(); + fast_time_queue tq(5000, 500, tick_manager); + + tq.push(123, 500); + + for (int ticks = 100; ticks <= 500; ticks += 100) { + tick_manager->ticks = std::chrono::milliseconds(ticks); + tq.update(); + } + + ASSERT_FALSE(tq.front().value.has_value()); + ASSERT_TRUE(tq.empty()); +} + +TEST(fast_time_queue, DelayedPopBacklogCompactsQueue) +{ + auto tick_manager = std::make_shared(); + inspectable_fast_time_queue tq(5000, 500, tick_manager); + + for (int i = 0; i < 80; ++i) { + tick_manager->ticks = std::chrono::milliseconds(i * 100); + tq.push(i, 5000); + } + + ASSERT_EQ(tq.size(), std::size_t{ 80 }); + ASSERT_EQ(tq.raw_queue_size(), std::size_t{ 80 }); + + tick_manager->ticks = std::chrono::milliseconds(8000); + while (tq.size() > std::size_t{ 5 }) { + auto elem = tq.pop_front(); + ASSERT_TRUE(elem.value.has_value()); + } + + EXPECT_EQ(tq.size(), std::size_t{ 5 }); + EXPECT_EQ(tq.raw_queue_size(), std::size_t{ 5 }); + EXPECT_EQ(tq.queue_index(), std::size_t{ 0 }); + + tick_manager->ticks = std::chrono::milliseconds(11000); + tq.update(); + + EXPECT_EQ(tq.size(), std::size_t{ 5 }); + EXPECT_EQ(tq.raw_queue_size(), std::size_t{ 5 }); + EXPECT_EQ(tq.queue_index(), std::size_t{ 0 }); + EXPECT_EQ(tq.front().value.value(), 75); +} + +TEST(fast_time_queue, PushAndPop) +{ + fast_time_queue tq(10, 1, tick_manager); + + tq.push(123, 1); + ASSERT_EQ(tq.pop_front().value.value(), 123); + + auto elem = tq.front(); + ASSERT_FALSE(elem.value.has_value()); +} + +TEST(fast_time_queue, PushAndExpireBeforePop) +{ + fast_time_queue tq(10, 1, tick_manager); + + tq.push(123, 1); + + tick_manager->ticks += std::chrono::milliseconds(1); + ASSERT_FALSE(tq.pop_front().value.has_value()); +} + +TEST(fast_time_queue, PushAndPopSequential) +{ + fast_time_queue tq(10, 1, tick_manager); + + for (int i = 0; i < 10; ++i) { + tq.push(i, 1); + } + + size_t popped = 0; + for (auto elem = tq.pop_front(); elem.value.has_value(); elem = tq.pop_front()) { + ASSERT_EQ(elem.value.value(), popped++); + } + + ASSERT_EQ(popped, 10); +} + +TEST(fast_time_queue, PushAndPopSequentialButExpireSome) +{ + fast_time_queue tq(10, 1, tick_manager); + + for (int i = 0; i < 10; ++i) { + tq.push(i, i + 1); + } + + size_t popped = 0; + size_t expected_value = 0; + size_t expected_expired = 0; + + for (auto elem = tq.pop_front(); elem.value.has_value(); elem = tq.pop_front()) { + ++popped; + EXPECT_EQ(elem.value.value(), expected_value); + EXPECT_EQ(elem.expired, expected_expired); + + expected_value += 4; + expected_expired = 3; + + tick_manager->ticks += std::chrono::milliseconds(4); + tq.update(); + } + + ASSERT_EQ(popped, 3); +} + +TEST(fast_time_queue, ExpireAllBeforePop) +{ + fast_time_queue tq(10, 1, tick_manager); + + for (int i = 0; i < 10; ++i) { + tq.push(i, i + 1); + } + + tick_manager->ticks = std::chrono::milliseconds(10); + + // Even though all elements have expired, we haven't updated the queue, so it won't know it is empty + ASSERT_FALSE(tq.empty()); + + // Try to pop an element, updating the queue. + for (auto elem = tq.pop_front(); elem.value.has_value();) { + // If we successfully popped an item, then we failed to expire the whole queue after the duration of the queue + // has passed. + FAIL(); + } + + // Now we have updated the queue, we expect it to be empty. + ASSERT_TRUE(tq.empty()); +} diff --git a/tests/time_queue.cpp b/tests/time_queue.cpp index 65c5378..39b63b6 100644 --- a/tests/time_queue.cpp +++ b/tests/time_queue.cpp @@ -17,6 +17,17 @@ struct test_tick_service : public tick_service static auto tick_manager = std::make_shared(); +template +class inspectable_time_queue : public time_queue +{ + public: + using time_queue::time_queue; + + std::size_t raw_queue_size() const noexcept { return this->_queue.size(); } + + std::size_t queue_index() const noexcept { return this->_queue_index; } +}; + TEST(time_queue, Construction) { ASSERT_NO_THROW(time_queue(10, 1, tick_manager)); @@ -38,6 +49,74 @@ TEST(time_queue, PushAndExpire) ASSERT_FALSE(tq.front().value.has_value()); } +TEST(time_queue, PushMidIntervalWithMaxTtlExpiresAtTtl) +{ + auto tick_manager = std::make_shared(); + time_queue tq(5000, 500, tick_manager); + + tick_manager->ticks = std::chrono::milliseconds(100); + tq.push(123, 5000); + + for (int ticks = 200; ticks <= 5000; ticks += 100) { + tick_manager->ticks = std::chrono::milliseconds(ticks); + tq.update(); + } + + tick_manager->ticks = std::chrono::milliseconds(5099); + ASSERT_EQ(tq.front().value.value(), 123); + + tick_manager->ticks = std::chrono::milliseconds(5100); + ASSERT_FALSE(tq.front().value.has_value()); +} + +TEST(time_queue, SubIntervalUpdatesStillAdvanceBuckets) +{ + auto tick_manager = std::make_shared(); + time_queue tq(5000, 500, tick_manager); + + tq.push(123, 500); + + for (int ticks = 100; ticks <= 500; ticks += 100) { + tick_manager->ticks = std::chrono::milliseconds(ticks); + tq.update(); + } + + ASSERT_FALSE(tq.front().value.has_value()); + ASSERT_TRUE(tq.empty()); +} + +TEST(time_queue, DelayedPopBacklogCompactsQueue) +{ + auto tick_manager = std::make_shared(); + inspectable_time_queue tq(5000, 500, tick_manager); + + for (int i = 0; i < 80; ++i) { + tick_manager->ticks = std::chrono::milliseconds(i * 100); + tq.push(i, 5000); + } + + ASSERT_EQ(tq.size(), std::size_t{ 80 }); + ASSERT_EQ(tq.raw_queue_size(), std::size_t{ 80 }); + + tick_manager->ticks = std::chrono::milliseconds(8000); + while (tq.size() > std::size_t{ 5 }) { + auto elem = tq.pop_front(); + ASSERT_TRUE(elem.value.has_value()); + } + + EXPECT_EQ(tq.size(), std::size_t{ 5 }); + EXPECT_EQ(tq.raw_queue_size(), std::size_t{ 5 }); + EXPECT_EQ(tq.queue_index(), std::size_t{ 0 }); + + tick_manager->ticks = std::chrono::milliseconds(11000); + tq.update(); + + EXPECT_EQ(tq.size(), std::size_t{ 5 }); + EXPECT_EQ(tq.raw_queue_size(), std::size_t{ 5 }); + EXPECT_EQ(tq.queue_index(), std::size_t{ 0 }); + EXPECT_EQ(tq.front().value.value(), 75); +} + TEST(time_queue, PushAndPop) { time_queue tq(10, 1, tick_manager);