#include "robust.h"
|
#include "util.h"
|
|
using namespace robust;
|
|
typedef CircularBuffer<int64_t, Allocator<int64_t>> Rcb;
|
Rcb *GetRCBImpl(SharedMemory &shm, const int nelem)
|
{
|
int cap = nelem + 1;
|
typedef uint64_t Data;
|
auto size = sizeof(Rcb) + sizeof(Data) * cap;
|
void *p = shm.Alloc(size);
|
if (p) {
|
return new (p) Rcb(cap, shm.get_segment_manager());
|
}
|
return nullptr;
|
}
|
Rcb *GetRCB(SharedMemory &shm, const int nelem)
|
{
|
void **pStore = shm.FindOrCreate<void *>("test_rcb_pointer", nullptr);
|
if (pStore) {
|
if (!*pStore) {
|
*pStore = GetRCBImpl(shm, nelem);
|
}
|
return (Rcb *) *pStore;
|
}
|
return nullptr;
|
}
|
|
void MySleep()
|
{
|
std::this_thread::sleep_for(2us);
|
}
|
|
BOOST_AUTO_TEST_CASE(RobustTest)
|
{
|
SharedMemory &shm = TestShm();
|
shm.Remove();
|
pid_t pid = getpid();
|
printf("pid : %d\n", pid);
|
auto Access = [](pid_t pid) {
|
char buf[100] = {0};
|
sprintf(buf, "/proc/%d/stat", pid);
|
int r = access(buf, F_OK);
|
printf("access %d\n", r);
|
};
|
Access(pid);
|
Access(pid + 1);
|
// Sleep(10s);
|
// return;
|
|
int nelement = 640;
|
auto rcb = GetRCB(shm, nelement);
|
BOOST_CHECK(rcb != nullptr);
|
BOOST_CHECK(rcb->empty());
|
BOOST_CHECK(rcb->push_back(1));
|
BOOST_CHECK(rcb->size() == 1);
|
int64_t d;
|
BOOST_CHECK(rcb->pop_front(d));
|
BOOST_CHECK(rcb->empty());
|
|
const uint64_t nmsg = 1000 * 1000 * 1;
|
uint64_t correct_total = nmsg * (nmsg - 1) / 2;
|
std::atomic<uint64_t> total(0);
|
std::atomic<uint64_t> nwrite(0);
|
std::atomic<uint64_t> writedone(0);
|
auto Writer = [&]() {
|
uint64_t n = 0;
|
while ((n = nwrite++) < nmsg) {
|
while (!rcb->push_back(n)) {
|
// MySleep();
|
}
|
++writedone;
|
}
|
};
|
std::atomic<uint64_t> nread(0);
|
auto Reader = [&]() {
|
while (nread.load() < nmsg) {
|
int64_t d;
|
if (rcb->pop_front(d)) {
|
++nread;
|
total += d;
|
} else {
|
MySleep();
|
}
|
}
|
};
|
|
auto status = [&]() {
|
auto next = steady_clock::now();
|
uint32_t lw = 0;
|
uint32_t lr = 0;
|
do {
|
std::this_thread::sleep_until(next);
|
next += 1s;
|
auto w = writedone.load();
|
auto r = nread.load();
|
printf("write: %6ld, spd: %6ld, read: %6ld, spd: %6ld , queue size: %d\n", w, w - lw, r, r - lr, rcb->size());
|
lw = w;
|
lr = r;
|
} while (nread.load() < nmsg);
|
};
|
|
ThreadManager threads;
|
boost::timer::auto_cpu_timer timer;
|
printf("Testing Robust Buffer, msgs %ld, queue size: %d \n", nmsg, nelement);
|
threads.Launch(status);
|
for (int i = 0; i < 10; ++i) {
|
threads.Launch(Reader);
|
threads.Launch(Writer);
|
}
|
threads.WaitAll();
|
printf("total: %ld, expected: %ld\n", total.load(), correct_total);
|
BOOST_CHECK_EQUAL(total.load(), correct_total);
|
}
|
