fix sendq lock, use different mutexes for in,out.

lichao

2021-04-15 af86015d724e5edf001aa024fe7d8581c45cffd9

fix sendq lock, use different mutexes for in,out.

 src/sendq.cpp

@@ -21,22 +21,6 @@

int SendQ::DoSend1Remote(bhome_shm::ShmMsgQueue &mq, const Remote &remote, Array &arr)
{
    static size_t total = 0;
    static size_t count = 0;
    static size_t max_len = 0;
    static time_t last = 0;
    ++count;
    total += arr.size();
    if (arr.size() > max_len) {
        max_len = arr.size();
    }
    time_t now;
    time(&now);
    if (now > last && count > 0) {
        last = now;
        printf("avg size : %ld, max size: %ld\n", total / count, max_len);
    }

    auto FirstNotExpired = [](Array &l) {
        auto Less = [](const TimedMsg &msg, const TimePoint &tp) { return msg.expire() < tp; };
        return std::lower_bound(l.begin(), l.end(), Now(), Less);
@@ -77,6 +61,7 @@

bool SendQ::TrySend(bhome_shm::ShmMsgQueue &mq)
{
    std::unique_lock<std::mutex> lock(mutex_out_);
    size_t nsend = 0;
    if (!out_.empty()) {
        auto rec = out_.begin();
@@ -91,7 +76,7 @@
    }

    auto Collect = [&]() {
        std::unique_lock<std::mutex> lock(mutex_);
        std::unique_lock<std::mutex> lock(mutex_in_);
        if (out_.empty()) {
            out_.swap(in_);
        } else if (nsend == 0) { // remote blocked

 src/sendq.h

@@ -68,7 +68,7 @@

        msg.AddRef();
        TimedMsg tmp(expire, MsgInfo{msg, onExpire});
        std::unique_lock<std::mutex> lock(mutex_);
        std::unique_lock<std::mutex> lock(mutex_in_);
        auto &al = in_[addr];
        if (!al.empty()) {
            al.front().emplace_back(std::move(tmp));
@@ -95,7 +95,8 @@
        MsgI &operator*() { return iter_->data().msg_; }
    };

    std::mutex mutex_;
    std::mutex mutex_in_;
    std::mutex mutex_out_;
    Store in_;
    Store out_;
};