diff --git a/include/timeq/tick_service.h b/include/timeq/tick_service.h index 7460e41..1b4089f 100644 --- a/include/timeq/tick_service.h +++ b/include/timeq/tick_service.h @@ -61,10 +61,7 @@ namespace timeq { virtual ~threaded_tick_service() { _stop = true; - - if (_tick_thread.joinable()) { - _tick_thread.join(); - } + _tick_thread.join(); } threaded_tick_service& operator=(const threaded_tick_service& other) diff --git a/include/timeq/time_queue.h b/include/timeq/time_queue.h index a1ecd8d..3ffdf7c 100644 --- a/include/timeq/time_queue.h +++ b/include/timeq/time_queue.h @@ -58,14 +58,10 @@ namespace timeq { struct queue_value_type { queue_value_type() = default; - queue_value_type(bucket_type& bucket, - index_type value_index, - tick_type expiry_tick, - tick_type wait_for_tick) + queue_value_type(bucket_type& bucket, index_type value_index, tick_type expiry_tick) : bucket(std::addressof(bucket)) , value_index(value_index) , expiry_tick(expiry_tick) - , wait_for_tick(wait_for_tick) { } @@ -75,7 +71,6 @@ namespace timeq { bucket_type* bucket; index_type value_index; tick_type expiry_tick; - tick_type wait_for_tick; }; using queue_type = std::vector; @@ -170,30 +165,20 @@ namespace timeq { * * @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); - } + FORCE_INLINE void push(const T& value, std::size_t ttl) { internal_push(value, 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); - } + FORCE_INLINE void push(T&& value, std::size_t ttl) { internal_push(std::forward(value), ttl); } /** * @brief Pop (increment) front @@ -234,7 +219,7 @@ namespace timeq { std::uint32_t expired = 0; while (_queue_index < _queue.size()) { - auto& [bucket, value_index, expiry_tick, pop_wait_ttl] = _queue.at(_queue_index); + auto& [bucket, value_index, expiry_tick] = _queue.at(_queue_index); if (ticks >= expiry_tick || value_index >= bucket->size()) { expired++; @@ -242,10 +227,6 @@ namespace timeq { continue; } - if (pop_wait_ttl > ticks) { - return { std::nullopt, expired }; - } - return { bucket->at(value_index), expired }; } @@ -363,7 +344,7 @@ namespace timeq { * * @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) + FORCE_INLINE void internal_push(auto&& value, std::size_t ttl) { if (ttl > _duration) { throw std::invalid_argument("TTL is greater than max duration"); @@ -384,7 +365,7 @@ namespace timeq { bucket_type& bucket = _buckets[future_index]; bucket.emplace_back(std::forward(value)); - _queue.emplace_back(bucket, bucket.size() - 1, expiry_tick, ticks + delay_ttl); + _queue.emplace_back(bucket, bucket.size() - 1, expiry_tick); } FORCE_INLINE void compact_consumed_queue() noexcept diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 6c884f3..b1077ec 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -8,6 +8,7 @@ set(gtest_force_shared_crt ON CACHE BOOL "" FORCE) CPMAddPackage("gh:google/googletest@1.17.0") add_executable(timeq_test + tick_service.cpp time_queue.cpp fast_time_queue.cpp ) diff --git a/tests/tick_service.cpp b/tests/tick_service.cpp new file mode 100644 index 0000000..094ba71 --- /dev/null +++ b/tests/tick_service.cpp @@ -0,0 +1,121 @@ +#include + +#include + +#include +#include +#include + +using namespace timeq; +using namespace std::chrono_literals; + +namespace { + + /// Lower bound as a fraction of wall-clock elapsed time (accounts for scheduler jitter). + constexpr double k_min_elapsed_fraction = 0.4; + + /// Upper bound slack beyond wall-clock elapsed time (accounts for overshoot and startup lag). + constexpr auto k_max_elapsed_slack = 30ms; + + void expect_ticks_within_wall_clock(std::chrono::microseconds tick_delta, std::chrono::microseconds wall_elapsed) + { + const auto min_expected = + std::chrono::microseconds(static_cast(wall_elapsed.count() * k_min_elapsed_fraction)); + const auto max_expected = wall_elapsed + k_max_elapsed_slack; + + EXPECT_GE(tick_delta, min_expected) + << "ticks (" << tick_delta.count() << "us) fell below " << (k_min_elapsed_fraction * 100) + << "% of wall time (" << wall_elapsed.count() << "us)"; + EXPECT_LE(tick_delta, max_expected) + << "ticks (" << tick_delta.count() << "us) exceeded wall time (" << wall_elapsed.count() << "us) plus slack"; + } + + std::chrono::microseconds measure_tick_growth(threaded_tick_service& service, + std::chrono::milliseconds wait_duration) + { + const auto initial_ticks = service.get(); + const auto wall_start = std::chrono::steady_clock::now(); + std::this_thread::sleep_for(wait_duration); + const auto wall_elapsed = + std::chrono::duration_cast(std::chrono::steady_clock::now() - wall_start); + + const auto tick_delta = service.get() - initial_ticks; + expect_ticks_within_wall_clock(tick_delta, wall_elapsed); + return tick_delta; + } + +} // namespace + +TEST(tick_service, InterfaceThroughPointer) +{ + std::unique_ptr service = std::make_unique(500); + + const auto initial = service->get(); + EXPECT_GE(initial, 0us); + + std::this_thread::sleep_for(25ms); + EXPECT_GT(service->get(), initial); +} + +TEST(threaded_tick_service, ConstructionAndDestruction) +{ + ASSERT_NO_THROW(threaded_tick_service service(500)); +} + +TEST(threaded_tick_service, InitialTicksNearZero) +{ + threaded_tick_service service(500); + EXPECT_GE(service.get(), 0us); + EXPECT_LT(service.get(), 5ms); +} + +TEST(threaded_tick_service, TicksIncreaseWithWallClock) +{ + threaded_tick_service service(500); + measure_tick_growth(service, 50ms); +} + +TEST(threaded_tick_service, TicksIncreaseWithCustomSleepDelay) +{ + threaded_tick_service service(1000); + measure_tick_growth(service, 75ms); +} + +TEST(threaded_tick_service, TicksAreMonotonic) +{ + threaded_tick_service service(500); + + auto previous = service.get(); + for (int i = 0; i < 5; ++i) { + std::this_thread::sleep_for(10ms); + const auto current = service.get(); + EXPECT_GE(current, previous); + previous = current; + } +} + +TEST(threaded_tick_service, CopyConstructorPreservesTicksAndContinues) +{ + threaded_tick_service original(500); + std::this_thread::sleep_for(30ms); + const auto original_ticks = original.get(); + + threaded_tick_service copy(original); + const auto copy_ticks = copy.get(); + + EXPECT_GE(copy_ticks, original_ticks); + + const auto copy_before = copy.get(); + std::this_thread::sleep_for(20ms); + EXPECT_GT(copy.get(), copy_before); +} + +// Copy assignment is not tested: operator= replaces _tick_thread without stopping or +// joining the existing background thread, which terminates the process. + +TEST(threaded_tick_service, DestructorStopsBackgroundThread) +{ + auto service = std::make_unique(500); + std::this_thread::sleep_for(15ms); + ASSERT_NO_THROW(service.reset()); +} diff --git a/tests/time_queue.cpp b/tests/time_queue.cpp index 39b63b6..8be5614 100644 --- a/tests/time_queue.cpp +++ b/tests/time_queue.cpp @@ -31,8 +31,48 @@ class inspectable_time_queue : public time_queue TEST(time_queue, Construction) { ASSERT_NO_THROW(time_queue(10, 1, tick_manager)); - ASSERT_THROW(time_queue(10, 1, nullptr), std::invalid_argument); + ASSERT_NO_THROW(time_queue(10, 1, tick_manager, 10000)); + ASSERT_NO_THROW(time_queue(10, 1, tick_manager, 0)); ASSERT_THROW(time_queue(0, 1, tick_manager), std::invalid_argument); + ASSERT_THROW(time_queue(10, 3, tick_manager), std::invalid_argument); + ASSERT_THROW(time_queue(1, 1, tick_manager), std::invalid_argument); + ASSERT_THROW(time_queue(10, 1, nullptr), std::invalid_argument); +} + +TEST(time_queue, Clear) +{ + time_queue tq(10, 1, tick_manager); + + ASSERT_TRUE(tq.empty()); + + tq.push(123, 1); + + ASSERT_FALSE(tq.empty()); + ASSERT_EQ(tq.size(), 1); + + tq.clear(); + ASSERT_TRUE(tq.empty()); + + tq.clear(); + ASSERT_TRUE(tq.empty()); +} + +TEST(time_queue, PushInvalid) +{ + time_queue tq(10, 1, tick_manager); + + try { + tq.push(123, 11); + } catch (...) { + } + ASSERT_THROW(tq.push(123, 11), std::invalid_argument); + ASSERT_NO_THROW(tq.push(123, 0)); + ASSERT_NO_THROW(tq.push(123, 2)); + + const int a = 123; + ASSERT_THROW(tq.push(a, 11), std::invalid_argument); + ASSERT_NO_THROW(tq.push(a, 0)); + ASSERT_NO_THROW(tq.push(a, 2)); } TEST(time_queue, PushAndExpire) @@ -122,10 +162,16 @@ TEST(time_queue, PushAndPop) time_queue tq(10, 1, tick_manager); tq.push(123, 1); + + ASSERT_EQ(tq.size(), 1); + ASSERT_EQ(tq.pop_front().value.value(), 123); auto elem = tq.front(); ASSERT_FALSE(elem.value.has_value()); + + tq.pop(); + ASSERT_TRUE(tq.empty()); } TEST(time_queue, PushAndExpireBeforePop) @@ -181,6 +227,26 @@ TEST(time_queue, PushAndPopSequentialButExpireSome) ASSERT_EQ(popped, 3); } +TEST(time_queue, PopAllAndCleanup) +{ + 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); + + tick_manager->ticks = std::chrono::milliseconds(10); + + ASSERT_TRUE(tq.empty()); +} + TEST(time_queue, ExpireAllBeforePop) { time_queue tq(10, 1, tick_manager); @@ -204,3 +270,22 @@ TEST(time_queue, ExpireAllBeforePop) // Now we have updated the queue, we expect it to be empty. ASSERT_TRUE(tq.empty()); } + +TEST(time_queue, ExpireAllAndCleanupQueueOnUpdate) +{ + 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()); + + [[maybe_unused]] auto _ = tq.front(); + + // Now we have updated the queue, we expect it to be empty. + ASSERT_TRUE(tq.empty()); +}