mirror of https://github.com/oxen-io/lokinet
You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
340 lines
8.8 KiB
C++
340 lines
8.8 KiB
C++
#include <util/thread/timerqueue.hpp>
|
|
|
|
#include <thread>
|
|
|
|
#include <gtest/gtest.h>
|
|
#include <gmock/gmock.h>
|
|
|
|
using CharQueue = llarp::thread::TimerQueue< const char* >;
|
|
using CharItem = llarp::thread::TimerQueueItem< const char* >;
|
|
|
|
TEST(TimerQueue, smoke)
|
|
{
|
|
CharQueue queue;
|
|
const absl::Time TA = absl::Time();
|
|
const absl::Time TB = TA + absl::Seconds(1);
|
|
const absl::Time TC = TB + absl::Seconds(1);
|
|
const absl::Time TD = TC + absl::Seconds(1);
|
|
const absl::Time TE = TD + absl::Seconds(1);
|
|
|
|
const char* VA = "hello";
|
|
const char* VB = "world,";
|
|
const char* VC = "how";
|
|
const char* VD = "are";
|
|
const char* VE = "you";
|
|
|
|
int HA = queue.add(TA, VA);
|
|
int HB = queue.add(TB, VB);
|
|
int HC = queue.add(TC, VC);
|
|
int HD = queue.add(TD, VD);
|
|
int HE = queue.add(TE, VE);
|
|
|
|
CharItem tItem;
|
|
absl::Time newMinTime;
|
|
size_t newSize;
|
|
|
|
ASSERT_TRUE(queue.popFront(&tItem));
|
|
ASSERT_EQ(VA, tItem.value());
|
|
ASSERT_EQ(TA, tItem.time());
|
|
ASSERT_EQ(HA, tItem.handle());
|
|
|
|
ASSERT_TRUE(queue.popFront(&tItem, &newSize, &newMinTime));
|
|
ASSERT_EQ(3, newSize);
|
|
ASSERT_EQ(TC, newMinTime);
|
|
ASSERT_EQ(TB, tItem.time());
|
|
ASSERT_EQ(VB, tItem.value());
|
|
ASSERT_EQ(HB, tItem.handle());
|
|
|
|
std::vector< CharItem > a1;
|
|
|
|
queue.popLess(TD, &a1, &newSize, &newMinTime);
|
|
ASSERT_EQ(2, a1.size());
|
|
ASSERT_EQ(1, newSize);
|
|
ASSERT_EQ(TE, newMinTime);
|
|
ASSERT_EQ(TC, a1[0].time());
|
|
ASSERT_EQ(VC, a1[0].value());
|
|
ASSERT_EQ(HC, a1[0].handle());
|
|
ASSERT_EQ(TD, a1[1].time());
|
|
ASSERT_EQ(VD, a1[1].value());
|
|
ASSERT_EQ(HD, a1[1].handle());
|
|
|
|
std::vector< CharItem > a2;
|
|
|
|
queue.popLess(TD, &a2, &newSize, &newMinTime);
|
|
ASSERT_EQ(0, a2.size());
|
|
ASSERT_EQ(1, newSize);
|
|
ASSERT_EQ(TE, newMinTime);
|
|
|
|
std::vector< CharItem > a3;
|
|
|
|
queue.popLess(TE, &a3, &newSize, &newMinTime);
|
|
ASSERT_EQ(1, a3.size());
|
|
ASSERT_EQ(0, newSize);
|
|
ASSERT_EQ(TE, a3[0].time());
|
|
ASSERT_EQ(VE, a3[0].value());
|
|
ASSERT_EQ(HE, a3[0].handle());
|
|
}
|
|
|
|
TEST(TimerQueue, KeySmoke)
|
|
{
|
|
CharQueue x1;
|
|
const absl::Time TA = absl::Time();
|
|
const absl::Time TB = TA + absl::Seconds(1);
|
|
const absl::Time TC = TB + absl::Seconds(1);
|
|
const absl::Time TD = TC + absl::Seconds(1);
|
|
const absl::Time TE = TD + absl::Seconds(1);
|
|
|
|
const char* VA = "hello";
|
|
const char* VB = "world,";
|
|
const char* VC = "how";
|
|
const char* VD = "are";
|
|
const char* VE = "you";
|
|
|
|
typedef CharQueue::Key Key;
|
|
|
|
const Key KA = Key(&TA);
|
|
const Key KB = Key(&TB);
|
|
const Key KC = Key(382);
|
|
const Key KD = Key(123);
|
|
const Key KE = Key(&VE);
|
|
|
|
int HA = x1.add(TA, VA, KA);
|
|
int HB = x1.add(TB, VB, KB);
|
|
int HC = x1.add(TC, VC, KC);
|
|
int HD = x1.add(TD, VD, KD);
|
|
int HE = x1.add(TE, VE, KE);
|
|
|
|
ASSERT_FALSE(x1.remove(HA, KB));
|
|
ASSERT_TRUE(x1.isValid(HA, KA));
|
|
ASSERT_FALSE(x1.update(HC, KD, TE));
|
|
|
|
CharItem tItem;
|
|
absl::Time newMinTime;
|
|
size_t newSize;
|
|
|
|
ASSERT_TRUE(x1.popFront(&tItem));
|
|
ASSERT_EQ(VA, tItem.value());
|
|
ASSERT_EQ(TA, tItem.time());
|
|
ASSERT_EQ(HA, tItem.handle());
|
|
ASSERT_EQ(KA, tItem.key());
|
|
|
|
ASSERT_TRUE(x1.popFront(&tItem, &newSize, &newMinTime));
|
|
ASSERT_EQ(3, newSize);
|
|
ASSERT_EQ(TC, newMinTime);
|
|
ASSERT_EQ(TB, tItem.time());
|
|
ASSERT_EQ(VB, tItem.value());
|
|
ASSERT_EQ(HB, tItem.handle());
|
|
ASSERT_EQ(KB, tItem.key());
|
|
|
|
std::vector< CharItem > a1;
|
|
|
|
x1.popLess(TD, &a1, &newSize, &newMinTime);
|
|
ASSERT_EQ(2, a1.size());
|
|
ASSERT_EQ(1, newSize);
|
|
ASSERT_EQ(TE, newMinTime);
|
|
ASSERT_EQ(TC, a1[0].time());
|
|
ASSERT_EQ(VC, a1[0].value());
|
|
ASSERT_EQ(HC, a1[0].handle());
|
|
ASSERT_EQ(KC, a1[0].key());
|
|
ASSERT_EQ(TD, a1[1].time());
|
|
ASSERT_EQ(VD, a1[1].value());
|
|
ASSERT_EQ(HD, a1[1].handle());
|
|
ASSERT_EQ(KD, a1[1].key());
|
|
|
|
std::vector< CharItem > a2;
|
|
|
|
x1.popLess(TD, &a2, &newSize, &newMinTime);
|
|
ASSERT_EQ(0, a2.size());
|
|
ASSERT_EQ(1, newSize);
|
|
ASSERT_EQ(TE, newMinTime);
|
|
|
|
std::vector< CharItem > a3;
|
|
|
|
x1.popLess(TE, &a3, &newSize, &newMinTime);
|
|
ASSERT_EQ(1, a3.size());
|
|
ASSERT_EQ(0, newSize);
|
|
ASSERT_EQ(TE, a3[0].time());
|
|
ASSERT_EQ(VE, a3[0].value());
|
|
ASSERT_EQ(HE, a3[0].handle());
|
|
ASSERT_EQ(KE, a3[0].key());
|
|
}
|
|
|
|
TEST(TimerQueue, Update)
|
|
{
|
|
const char VA[] = "A";
|
|
const char VB[] = "B";
|
|
const char VC[] = "C";
|
|
const char VD[] = "D";
|
|
const char VE[] = "E";
|
|
|
|
// clang-format off
|
|
static const struct
|
|
{
|
|
int m_secs;
|
|
int m_nsecs;
|
|
const char* m_value;
|
|
int m_updsecs;
|
|
int m_updnsecs;
|
|
bool m_isNewTop;
|
|
} VALUES[] = {
|
|
{2, 1000000, VA, 0, 1000000, false},
|
|
{2, 1000000, VB, 3, 1000000, false},
|
|
{2, 1000000, VC, 0, 4000, false},
|
|
{2, 1000001, VB, 0, 3999, true},
|
|
{1, 9999998, VC, 4, 9999998, false},
|
|
{1, 9999999, VD, 0, 0, true},
|
|
{0, 4000, VE, 10, 4000, false}};
|
|
// clang-format on
|
|
|
|
static const int POP_ORDER[] = {5, 3, 2, 0, 1, 4, 6};
|
|
|
|
const int NUM_VALUES = sizeof VALUES / sizeof *VALUES;
|
|
int handles[NUM_VALUES];
|
|
|
|
CharQueue queue;
|
|
|
|
{
|
|
CharItem item;
|
|
ASSERT_FALSE(queue.popFront(&item));
|
|
}
|
|
|
|
for(int i = 0; i < NUM_VALUES; ++i)
|
|
{
|
|
const char* VAL = VALUES[i].m_value;
|
|
const int SECS = VALUES[i].m_secs;
|
|
const int NSECS = VALUES[i].m_nsecs;
|
|
absl::Time TIME =
|
|
absl::Time() + absl::Seconds(SECS) + absl::Nanoseconds(NSECS);
|
|
|
|
handles[i] = queue.add(TIME, VAL);
|
|
|
|
ASSERT_EQ(i + 1, queue.size());
|
|
ASSERT_TRUE(queue.isValid(handles[i]));
|
|
}
|
|
|
|
for(int i = 0; i < NUM_VALUES; ++i)
|
|
{
|
|
const int UPDSECS = VALUES[i].m_updsecs;
|
|
const bool EXPNEWTOP = VALUES[i].m_isNewTop;
|
|
const int UPDNSECS = VALUES[i].m_updnsecs;
|
|
absl::Time UPDTIME =
|
|
absl::Time() + absl::Seconds(UPDSECS) + absl::Nanoseconds(UPDNSECS);
|
|
|
|
bool isNewTop;
|
|
|
|
CharItem item;
|
|
ASSERT_TRUE(queue.isValid(handles[i])) << i;
|
|
ASSERT_TRUE(queue.update(handles[i], UPDTIME, &isNewTop)) << i;
|
|
EXPECT_EQ(EXPNEWTOP, isNewTop) << i;
|
|
ASSERT_TRUE(queue.isValid(handles[i])) << i;
|
|
}
|
|
|
|
for(int i = 0; i < NUM_VALUES; ++i)
|
|
{
|
|
const int I = POP_ORDER[i];
|
|
const char* EXPVAL = VALUES[I].m_value;
|
|
const int EXPSECS = VALUES[I].m_updsecs;
|
|
const int EXPNSECS = VALUES[I].m_updnsecs;
|
|
absl::Time EXPTIME =
|
|
absl::Time() + absl::Seconds(EXPSECS) + absl::Nanoseconds(EXPNSECS);
|
|
|
|
CharItem item;
|
|
ASSERT_TRUE(queue.isValid(handles[I]));
|
|
ASSERT_TRUE(queue.popFront(&item));
|
|
ASSERT_EQ(EXPTIME, item.time());
|
|
ASSERT_EQ(EXPVAL, item.value());
|
|
ASSERT_FALSE(queue.isValid(handles[I]));
|
|
}
|
|
}
|
|
|
|
TEST(TimerQueue, ThreadSafety)
|
|
{
|
|
using Data = std::string;
|
|
using StringQueue = llarp::thread::TimerQueue< std::string >;
|
|
using StringItem = llarp::thread::TimerQueueItem< std::string >;
|
|
|
|
using Info = std::pair< int, std::vector< StringItem >* >;
|
|
|
|
static constexpr size_t NUM_THREADS = 10;
|
|
static constexpr size_t NUM_ITERATIONS = 1000;
|
|
static constexpr size_t NUM_REMOVE_ALL = NUM_ITERATIONS / 2;
|
|
|
|
Info info[NUM_THREADS];
|
|
std::thread threads[NUM_THREADS + 1];
|
|
std::vector< StringItem > items[NUM_THREADS];
|
|
|
|
absl::Barrier barrier(NUM_THREADS + 1);
|
|
|
|
StringQueue queue;
|
|
|
|
for(size_t i = 0; i < NUM_THREADS; ++i)
|
|
{
|
|
info[i].first = i;
|
|
info[i].second = &items[i];
|
|
threads[i] = std::thread(
|
|
[](Info* nfo, absl::Barrier* b, StringQueue* q) {
|
|
const int THREAD_ID = nfo->first;
|
|
std::vector< StringItem >* vPtr = nfo->second;
|
|
|
|
// We stagger the removeAll steps among the threads.
|
|
const unsigned int STEP_REMOVE_ALL =
|
|
THREAD_ID * NUM_REMOVE_ALL / NUM_THREADS;
|
|
|
|
std::ostringstream oss;
|
|
oss << THREAD_ID;
|
|
Data V(oss.str());
|
|
|
|
b->Block();
|
|
|
|
size_t newSize;
|
|
absl::Time newMinTime;
|
|
StringItem item;
|
|
for(size_t j = 0; j < NUM_ITERATIONS; ++j)
|
|
{
|
|
const absl::Time TIME =
|
|
absl::Time() + absl::Seconds((j * (j + 3)) % NUM_ITERATIONS);
|
|
int h = q->add(TIME, V);
|
|
q->update(h, TIME);
|
|
if(q->popFront(&item, &newSize, &newMinTime))
|
|
{
|
|
vPtr->push_back(item);
|
|
}
|
|
h = q->add(newMinTime, V);
|
|
q->popLess(newMinTime, vPtr);
|
|
if(q->remove(h, &item, &newSize, &newMinTime))
|
|
{
|
|
vPtr->push_back(item);
|
|
}
|
|
if(j % NUM_REMOVE_ALL == STEP_REMOVE_ALL)
|
|
{
|
|
q->removeAll(vPtr);
|
|
}
|
|
}
|
|
},
|
|
&info[i], &barrier, &queue);
|
|
}
|
|
|
|
threads[NUM_THREADS] = std::thread(
|
|
[](absl::Barrier* b, StringQueue* q) {
|
|
b->Block();
|
|
for(size_t i = 0; i < NUM_ITERATIONS; ++i)
|
|
{
|
|
size_t size = q->size();
|
|
ASSERT_GE(size, 0);
|
|
ASSERT_LE(size, NUM_THREADS);
|
|
}
|
|
},
|
|
&barrier, &queue);
|
|
|
|
size_t size = 0;
|
|
for(size_t i = 0; i < NUM_THREADS; ++i)
|
|
{
|
|
threads[i].join();
|
|
size += static_cast< int >(items[i].size());
|
|
}
|
|
threads[NUM_THREADS].join();
|
|
|
|
ASSERT_EQ(0, queue.size());
|
|
ASSERT_EQ(1000 * NUM_THREADS * 2, size);
|
|
}
|