#include #include #include #include #include #include #include #include #include #include #include #include #include #include "shm_queue.h" #include "bh_util.h" #include #include using namespace std::chrono_literals; using namespace bhome_shm; using namespace boost::posix_time; auto Now = []() { return second_clock::universal_time(); }; struct s1000 { char a[1000]; }; typedef std::function FuncVV; class ScopeCall : private boost::noncopyable { FuncVV f_; public: ScopeCall(FuncVV f):f_(f) { f_(); } ~ScopeCall() { f_(); } }; class ThreadManager { std::vector threads_; public: ~ThreadManager() { WaitAll(); } template void Launch(T t, P...p) { threads_.emplace_back(t, p...); } void WaitAll() { for (auto &t : threads_) { if (t.joinable()) { t.join(); } } } }; class ProcessManager { std::vector procs_; public: ~ProcessManager() { WaitAll(); } template void Launch(T t, P...p) { auto pid = fork(); if (pid == 0) { // child t(p...); exit(0); } else if (pid != -1) { // Ok procs_.push_back(pid); } }; void WaitAll() { for (auto &pid: procs_) { int status = 0; int options = WUNTRACED | WCONTINUED; waitpid(pid, &status, options); } procs_.clear(); } }; struct ShmRemover { std::string name_; ShmRemover(const std::string &name):name_(name) { SharedMemory::Remove(name_); } ~ShmRemover() { SharedMemory::Remove(name_); } }; template struct IsSameType { static const bool value = false; }; template struct IsSameType { static const bool value = true; }; BOOST_AUTO_TEST_CASE(BasicTest) { const std::string shm_name("basic"); ShmRemover auto_remove(shm_name); SharedMemory shm(shm_name, 1024*1024*10); auto Avail = [&]() { return shm.get_free_memory(); }; offset_ptr p; BOOST_CHECK(!p); BOOST_CHECK(p.get() == 0); p = 0; BOOST_CHECK(!p); BOOST_CHECK(p.get() == 0); const char *str = "basic"; p = str; BOOST_CHECK(p); BOOST_CHECK(p.get() == str); p = 0; BOOST_CHECK(!p); BOOST_CHECK(p.get() == 0); auto init_avail = Avail(); auto BasicTest = [&](int tid, int nloop) { auto Code = [&](int id) { typedef ShmObject Int; std::string name = std::to_string(id); auto a0 = Avail(); Int i1(shm, name); auto a1 = Avail(); BOOST_CHECK_LT(a1, a0); printf("s1000 size: %ld\n", a0 - a1); i1->a[0] = 5; Int i2(shm, name); auto a2 = Avail(); BOOST_CHECK_EQUAL(a1, a2); BOOST_CHECK_EQUAL(i1.data(), i2.data()); int i = i1.Remove(); BOOST_CHECK_EQUAL(Avail(), a0); { auto old = Avail(); void *p = shm.Alloc(1024); shm.Dealloc(p); BOOST_CHECK_EQUAL(old, Avail()); } bool r = shared_memory_object::remove(shm_name.c_str()); BOOST_CHECK(r); }; for (int i = 0; i < nloop; ++i) { Code(i + tid*nloop); } }; // boost::timer::auto_cpu_timer timer; ThreadManager threads; int nthread = 1; int nloop = 1; for (int i = 0; i < nthread; ++i) { threads.Launch(BasicTest, i, nloop); } BOOST_CHECK_EQUAL(init_avail, Avail()); } BOOST_AUTO_TEST_CASE(ForkTest) { ProcessManager procs; const int nproc = 10; printf("Testing fork:\n"); auto child = [&](int id) { std::this_thread::sleep_for(100ms *id); printf("child id: %3d/%d ends\r", id, nproc); }; for (int i = 0; i < nproc; ++i) { procs.Launch(child, i+1); } } BOOST_AUTO_TEST_CASE(TimedWaitTest) { const std::string shm_name("shm_wait"); ShmRemover auto_remove(shm_name); SharedMemory shm(shm_name, 1024*1024); ShmMsgQueue q(shm, 64); for (int i = 0; i < 2; ++i) { int ms = i * 100; printf("Timeout Test %4d: ", ms); boost::timer::auto_cpu_timer timer; MQId id; void *data; size_t size; bool r = q.Recv(id, data, size, ms); BOOST_CHECK(!r); } } BOOST_AUTO_TEST_CASE(RefCountTest) { const std::string shm_name("ShmRefCount"); ShmRemover auto_remove(shm_name); SharedMemory shm(shm_name, 1024*1024); Msg m0(shm.Alloc(1000), shm.New()); BOOST_CHECK(m0.IsCounted()); BOOST_CHECK_EQUAL(m0.Count(), 1); Msg m1 = m0; BOOST_CHECK(m1.IsCounted()); BOOST_CHECK_EQUAL(m1.AddRef(), 2); BOOST_CHECK_EQUAL(m0.AddRef(), 3); BOOST_CHECK_EQUAL(m0.Release(shm), 2); BOOST_CHECK_EQUAL(m0.Release(shm), 1); BOOST_CHECK_EQUAL(m1.Release(shm), 0); BOOST_CHECK(!m1.IsCounted()); } BOOST_AUTO_TEST_CASE(MsgHeaderTest) { MsgMetaV1 head; BOOST_CHECK_EQUAL(head.self_size_, sizeof(head)); BOOST_CHECK_EQUAL(head.type_, kMsgTypeNormal); BOOST_CHECK_EQUAL(head.tag_, kMsgMetaTag); BOOST_CHECK_EQUAL(head.data_size_, 0); BOOST_CHECK(head.src_id_ == boost::uuids::nil_uuid()); head.data_size_ = 100; head.src_id_ = boost::uuids::random_generator()(); head.type_ = 123; char buf[100] = {0}; head.Pack(buf); MsgMetaV1 result; result.Parse(buf); BOOST_CHECK_EQUAL(memcmp(&head, &result, sizeof(head)), 0); } BOOST_AUTO_TEST_CASE(SpeedTest) { const std::string shm_name("ShmSpeed"); ShmRemover auto_remove(shm_name); const int mem_size = 1024*1024*50; MQId id = boost::uuids::random_generator()(); const int timeout = 100; const uint32_t data_size = 4000; auto Writer = [&](int writer_id, uint64_t n) { SharedMemory shm(shm_name, mem_size); ShmMsgQueue mq(shm, 64); std::string str(data_size, 'a'); Msg msg; DEFER1(msg.Release(shm);); msg.Build(shm, mq.Id(), str.data(), str.size(), true); for (int i = 0; i < n; ++i) { // mq.Send(id, str.data(), str.size(), timeout); mq.Send(id, msg, timeout); } }; auto Reader = [&](int reader_id, std::atomic *run, bool isfork){ SharedMemory shm(shm_name, mem_size); ShmMsgQueue mq(id, shm, 1000); while(*run) { Msg msg; if (mq.Recv(msg, timeout)) { MsgMetaV1 header; if (!header.Parse(msg.get())) { BOOST_CHECK(false); } if (header.data_size_ != data_size) { BOOST_CHECK(false); } msg.Release(shm); } else if (isfork) { exit(0); // for forked quit after 1s. } } }; auto State = [&](std::atomic *run){ SharedMemory shm(shm_name, mem_size); auto init = shm.get_free_memory(); printf("shm init : %ld\n", init); while (*run) { auto cur = shm.get_free_memory(); printf("shm used : %8ld/%ld\n", init - cur, init); std::this_thread::sleep_for(1s); } }; int nwriters[] = {1,2,4}; int nreaders[] = {1,2}; auto Test = [&](auto &www, auto &rrr, bool isfork) { for (auto nreader : nreaders) { for (auto nwriter : nwriters) { const uint64_t nmsg = 1000 * 1000 * 10 / nwriter; const uint64_t total_msg = nmsg * nwriter; std::atomic run(true); std::this_thread::sleep_for(10ms); boost::timer::auto_cpu_timer timer; for (int i = 0; i < nreader; ++i) { rrr.Launch(Reader, i, &run, isfork); } for (int i = 0; i < nwriter; ++i) { www.Launch(Writer, i, nmsg); } www.WaitAll(); run.store(false); rrr.WaitAll(); printf("%3d Write %ld msg R(%3d) W(%3d), : ", getpid(), total_msg, nreader, nwriter); } } }; std::atomic run(true); ThreadManager state; state.Launch(State, &run); // typedef ProcessManager Manager; // typedef ThreadManager Manager; // const bool isfork = IsSameType::value; ProcessManager pw, pr; printf("================ Testing process io: =======================================================\n"); Test(pw, pr, true); ThreadManager tw, tr; printf("---------------- Testing thread io: -------------------------------------------------------\n"); Test(tw, tr, false); run.store(false); } // Request Reply Test BOOST_AUTO_TEST_CASE(RRTest) { const std::string shm_name("ShmReqRep"); ShmRemover auto_remove(shm_name); const int qlen = 64; const size_t msg_length = 1000; std::string msg_content(msg_length, 'a'); msg_content[20] = '\0'; SharedMemory shm(shm_name, 1024*1024*50); auto Avail = [&]() { return shm.get_free_memory(); }; auto init_avail = Avail(); ShmMsgQueue srv(shm, qlen); ShmMsgQueue cli(shm, qlen); Msg ref_counted_msg; ref_counted_msg.Build(shm, cli.Id(), msg_content.data(), msg_content.size(), true); std::atomic count(0); std::atomic last_time(Now() - seconds(1)); std::atomic last_count(0); auto Client = [&](int cli_id, int nmsg){ for (int i = 0; i < nmsg; ++i) { auto Send = [&]() { return cli.Send(srv.Id(), msg_content.data(), msg_content.size(), 1000); }; auto SendRefCounted = [&]() { return cli.Send(srv.Id(), ref_counted_msg, 1000); }; if (!SendRefCounted()) { printf("********** client send error.\n"); continue; } MQId id; void *data = 0; size_t size = 0; if (!cli.Recv(id, data, size, 1000)) { printf("********** client recv error.\n"); } else { DEFER1(free(data)); if(size != msg_length) { BOOST_CHECK(false); } ++count; auto cur = Now(); if (last_time.exchange(cur) != cur) { std::cout << "time: " << Now(); printf(", total msg:%10ld, speed:[%8ld/s], used mem:%8ld, refcount:%d\n", count.load(), count - last_count.exchange(count), init_avail - Avail(), ref_counted_msg.Count()); last_time = cur; } } } }; std::atomic stop(false); auto Server = [&](){ void *data = 0; size_t size = 0; MQId src_id; while (!stop) { if (srv.Recv(src_id, data, size, 100)) { DEFER1(free(data)); auto Send = [&](){ return srv.Send(src_id, data, size, 100); }; auto SendRefCounted = [&](){ return srv.Send(src_id, ref_counted_msg, 100); }; if (SendRefCounted()) { if (size != msg_content.size()) { BOOST_TEST(false, "server msg size error"); } } } } }; boost::timer::auto_cpu_timer timer; DEFER1(printf("Request Reply Test:");); ThreadManager clients, servers; for (int i = 0; i < qlen; ++i) { servers.Launch(Server); } int ncli = 100*1; uint64_t nmsg = 100*100*2; printf("client threads: %d, msgs : %ld, total msg: %ld\n", ncli, nmsg, ncli * nmsg); for (int i = 0; i < ncli; ++i) { clients.Launch(Client, i, nmsg); } clients.WaitAll(); printf("request ok: %ld\n", count.load()); stop = true; servers.WaitAll(); BOOST_CHECK(ref_counted_msg.IsCounted()); BOOST_CHECK_EQUAL(ref_counted_msg.Count(), 1); ref_counted_msg.Release(shm); BOOST_CHECK(!ref_counted_msg.IsCounted()); // BOOST_CHECK_THROW(reply.Count(), int); } inline int MyMin(int a, int b) { printf("MyMin\n"); return a < b ? a : b; } int test_main(int argc, char *argv[]) { printf("test main\n"); int a = 0; int b = 0; BOOST_CHECK_EQUAL(a, b); int n = MyMin(4,6); for (int i = 0; i < n; ++i) { printf("i = %d\n", i); } return 0; }