fix crash by using normal timeout; add sendq todo.

lichao

2021-04-16 94a455aba299af5ffe476560d859dfd007cd5467

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
/*
 * =====================================================================================
 *
 *       Filename:  sendq.cpp
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  2021年04月14日 09时22分50秒
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Li Chao (), lichao@aiotlink.com
 *   Organization:  
 *
 * =====================================================================================
 */
#include "sendq.h"
#include "shm_queue.h"
#include <chrono>
 
//TODO change to save head, body, instead of MsgI.
// as MsgI which is in shm, but head, body are in current process.
// Then if node crashes, shm will not be affected by msgs in sendq.
// but pulishing ref-counted msg need some work.
 
int SendQ::DoSend1Remote(bhome_shm::ShmMsgQueue &mq, const Remote &remote, Array &arr)
{
    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);
    };
 
    auto pos = FirstNotExpired(arr);
    for (auto it = arr.begin(); it != pos; ++it) {
        auto &info = it->data();
        if (info.on_expire_) {
            info.on_expire_(info.msg_);
        }
        info.msg_.Release(mq.shm());
    }
 
    int n = mq.TrySendAll(*(MQId *) remote.data(), MsgIter(pos), MsgIter(arr.end()));
    for (int i = 0; i < n; ++i) {
        auto &msg = pos->data().msg_;
        if (msg.IsCounted()) {
            msg.Release(mq.shm());
        }
        ++pos;
    }
 
    arr.erase(arr.begin(), pos);
    return n;
}
 
int SendQ::DoSend1Remote(bhome_shm::ShmMsgQueue &mq, const Remote &remote, ArrayList &al)
{
    int nsend = 0;
    auto AllSent = [&](Array &arr) {
        nsend += DoSend1Remote(mq, remote, arr);
        return arr.empty();
    };
    for (auto it = al.begin(); it != al.end() && AllSent(*it); it = al.erase(it)) {}
    return nsend;
}
 
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();
        do {
            nsend += DoSend1Remote(mq, rec->first, rec->second);
            if (rec->second.empty()) {
                rec = out_.erase(rec);
            } else {
                ++rec;
            }
        } while (rec != out_.end());
    }
 
    auto Collect = [&]() {
        std::unique_lock<std::mutex> lock(mutex_in_);
        if (out_.empty()) {
            out_.swap(in_);
        } else if (nsend == 0) { // remote blocked
            for (auto &kv : in_) {
                auto &to = out_[kv.first];
                to.splice(to.end(), kv.second);
            }
            in_.clear();
        }
    };
 
    Collect();
 
    return !out_.empty();
}