add queue length status.

lichao

2021-03-31 4353f73ea0c30c776a3957dc674d750e51519ca3

add queue length status.

 src/pubsub.cpp

@@ -34,6 +34,15 @@
bool BusManager::Start(const int nworker)
{
    auto onRecv = [&](MsgI &imsg) {
#ifndef NDEBUG
        static std::atomic<time_t> last(0);
        time_t now = 0;
        time(&now);
        if (last.exchange(now) < now) {
            printf("bus queue size: %ld\n", socket_.Pending());
        }
#endif

        BHMsg msg;
        if (!imsg.Unpack(msg)) {
            return;
@@ -130,7 +139,7 @@
        }
    };

    return socket_.StartRaw(onRecv, std::min(nworker, kMaxWorker));
    return socket_.StartRaw(onRecv, std::min((nworker > 0 ? nworker : 2), kMaxWorker));
}

} // namespace bhome_shm

 src/shm_queue.h

@@ -138,6 +138,7 @@
    {
        return Send(shm(), remote_id, msg, timeout_ms);
    }
    size_t Pending() const { return data()->size(); }
};

} // namespace bhome_shm

 src/socket.cpp

@@ -63,7 +63,6 @@
        }
        DEFER1(imsg.Release(shm_));
        return Queue::Send(shm_, kBHBusQueueId, imsg, timeout_ms);

    } catch (...) {
        return false;
    }

 src/socket.h

@@ -61,6 +61,7 @@
    bool Start(const RecvCB &onData, int nworker = 1);
    bool StartRaw(const RecvRawCB &onData, int nworker = 1);
    bool Stop();
    size_t Pending() const { return mq_ ? mq_->Pending() : 0; }

private:
    bool StopNoLock();

 utest/utest.cpp

@@ -66,20 +66,24 @@
BOOST_AUTO_TEST_CASE(PubSubTest)
{
    const std::string shm_name("ShmPubSub");
    ShmRemover auto_remove(shm_name); //remove twice? in case of killed?
    // ShmRemover auto_remove(shm_name); //remove twice? in case of killed?
    SharedMemory shm(shm_name, 1024 * 1024 * 50);
    DEFER1(shm.Remove());
    auto Avail = [&]() { return shm.get_free_memory(); };
    auto init_avail = Avail();
    int *flag = shm.find_or_construct<int>("flag")(123);
    printf("flag = %d\n", *flag);
    ++*flag;

    BusManager bus(shm);
    bus.Start(1);
    bus.Start();
    std::this_thread::sleep_for(100ms);

    std::atomic<uint64_t> count(0);
    std::atomic<ptime> last_time(Now() - seconds(1));
    std::atomic<uint64_t> last_count(0);

    const uint64_t nmsg = 100;
    const uint64_t nmsg = 100 * 2;
    const int timeout = 1000;
    auto Sub = [&](int id, const std::vector<std::string> &topics) {
        ShmSocket client(ShmSocket::eSockSubscribe, shm);
@@ -87,7 +91,7 @@
        std::mutex mutex;
        std::condition_variable cv;

        int i = 0;
        uint64_t i = 0;
        auto OnRecv = [&](BHMsg &msg) {
            if (msg.type() != kMsgTypePublish) {
                BOOST_CHECK(false);
@@ -117,7 +121,7 @@

    auto Pub = [&](const std::string &topic) {
        ShmSocket provider(ShmSocket::eSockPublish, shm);
        for (int i = 0; i < nmsg; ++i) {
        for (unsigned i = 0; i < nmsg; ++i) {
            std::string data = topic + std::to_string(i) + std::string(1000, '-');

            bool r = provider.Publish(topic, data.data(), data.size(), timeout);
@@ -134,7 +138,7 @@
        topics.push_back("t" + std::to_string(i));
    }
    Topics part;
    for (int i = 0; i < topics.size(); ++i) {
    for (size_t i = 0; i < topics.size(); ++i) {
        part.push_back(topics[i]);
        threads.Launch(Sub, i, topics);
    }