add ref count.

lichao

2021-03-25 d3e7f93e69cb24c766292d8780e745caf24d42a8

 utest/utest.cpp

@@ -1,6 +1,4 @@
#include <stdio.h>
#include "shm.h"
#include "../src/bh_util.h"
#include <string>
#include <vector>
#include <thread>
@@ -8,9 +6,12 @@
#include <atomic>
#include <boost/noncopyable.hpp>
#include <boost/timer/timer.hpp>
#include <boost/test/auto_unit_test.hpp>
#include <boost/test/unit_test.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/date_time/microsec_time_clock.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include "shm_queue.h"
#include "bh_util.h"

using namespace std::chrono_literals;
using namespace bhome_shm;
@@ -47,11 +48,26 @@
    ~ShmRemover() { SharedMemory::Remove(name_); }
};

BOOST_AUTO_TEST_CASE(ShmBasic)
BOOST_AUTO_TEST_CASE(ShmBasicTest)
{
    const std::string shm_name("basic");
    ShmRemover auto_remove(shm_name);
    SharedMemory shm(shm_name, 1024*1024*10);

    offset_ptr<const void> 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 Avail = [&]() { return shm.get_free_memory(); };
    auto init_avail = Avail();
@@ -76,8 +92,8 @@

            {
                auto old = Avail();
                void *p = shm.allocate(1024);
                shm.deallocate(p);
                void *p = shm.Alloc(1024);
                shm.Dealloc(p);
                BOOST_CHECK_EQUAL(old, Avail());
            }

@@ -104,12 +120,12 @@
    BOOST_CHECK_EQUAL(init_avail, Avail());
}

BOOST_AUTO_TEST_CASE(TimedWait)
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(10000, shm, 64);
    ShmMsgQueue q(shm, 64);
    for (int i = 0; i < 5; ++i) {
        int ms = i * 100;
        printf("Timeout Test %d: ", ms);
@@ -122,37 +138,84 @@
    }
}

BOOST_AUTO_TEST_CASE(RequestReply)
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<RefCount>());
    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.RemoveRef(), 2);
    BOOST_CHECK_EQUAL(m0.RemoveRef(), 1);
    BOOST_CHECK_EQUAL(m1.RemoveRef(), 0);
    BOOST_CHECK_EQUAL(m1.Count(), 0);
}

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(RequestReplyTest)
{
    const std::string shm_name("ShmReqRep");
    ShmRemover auto_remove(shm_name);
    SharedMemory shm(shm_name, 1024*1024*50);
    auto Avail = [&]() { return shm.get_free_memory(); };
    auto init_avail = Avail();
    // DEFER1(BOOST_CHECK_EQUAL(init_avail, Avail()); printf("Request Reply Test shm No Leak.\n"););

    auto f0 = init_avail;
    const int qlen = 64;
    ShmMsgQueue srv(1, shm, qlen);
    ShmMsgQueue cli(2, shm, qlen);
    ShmMsgQueue srv(shm, qlen);
    ShmMsgQueue cli(shm, qlen);
    auto f1= shm.get_free_memory();

    const size_t msg_length = 1000;
    std::string msg_content(msg_length, 'a');
    msg_content[20] = '\0';
    Msg request;
    request.Build(shm, cli.Id(), msg_content.data(), msg_content.size(), true);
    Msg reply(request);

    std::atomic<bool> stop(false);
    std::atomic<uint64_t> count(0);
    using namespace boost::posix_time;
    auto Now = []() { return second_clock::universal_time(); };
    std::atomic<ptime> last_time(Now());
    std::atomic<ptime> last_time(Now() - seconds(1));
    std::atomic<uint64_t> last_count(0);

    auto Client = [&](int tid, int nmsg){
        for (int i = 0; i < nmsg; ++i) {
            if (!cli.Send(srv.id(), msg_content.data(), msg_content.size(), 1000)) {
            auto Send = [&]() { return cli.Send(srv.Id(), msg_content.data(), msg_content.size(), 1000); };
            auto SendRefCounted = [&]() { return cli.Send(srv.Id(), request, 1000); };

            if (!Send()) {
                printf("********** client send error.\n");
                continue;
            }
            MQId id = 0;
            MQId id;
            void *data = 0;
            size_t size = 0;
            if (!cli.Recv(id, data, size, 1000)) {
@@ -166,7 +229,8 @@
                auto cur = Now();
                if (last_time.exchange(cur) != cur) {
                    std::cout << "time: " << Now();
                    printf(", total msg:%10ld, speed:%8ld, used mem:%8ld\n",  count.load(), count - last_count.exchange(count), init_avail - Avail());
                    printf(", total msg:%10ld, speed:%8ld/s, used mem:%8ld, refcount:%d\n",
                           count.load(), count - last_count.exchange(count), init_avail - Avail(), request.Count());
                    last_time = cur;
                }

@@ -177,11 +241,14 @@
    auto Server = [&](){
        void *data = 0;
        size_t size = 0;
        MQId src_id = 0;
        MQId src_id;
        while (!stop) {
            if (srv.Recv(src_id, data, size, 100)) {
                DEFER1(free(data));
                if (srv.Send(src_id, data, size, 100)) {
                auto Send = [&](){ return srv.Send(src_id, data, size, 100); };
                auto SendRefCounted = [&](){ return srv.Send(src_id, reply, 100); };

                if (SendRefCounted()) {
                    if (size != msg_content.size()) {
                        BOOST_TEST(false, "server msg size error");
                    }
@@ -195,19 +262,22 @@

    ThreadManager clients, servers;
    for (int i = 0; i < qlen; ++i) { servers.Launch(Server); }
    int ncli = 100;
    uint64_t nmsg = 1000*10;
    int ncli = 100*1;
    uint64_t nmsg = 100*100;
    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();

    srv.Remove();
    cli.Remove();
    BOOST_CHECK_EQUAL(init_avail, Avail());
    printf("Request Reply Test shm No Leak.\n");
    BOOST_CHECK(request.IsCounted());
    BOOST_CHECK_EQUAL(request.Count(), 1);
    BOOST_CHECK(reply.IsCounted());
    BOOST_CHECK_EQUAL(reply.Count(), 1);
    if (request.RemoveRef() == 0) {
        BOOST_CHECK_EQUAL(reply.Count(), 0);
        request.FreeFrom(shm);
    }
}

int test_main(int argc, char *argv[])