| | |
| | | #include "shm_queue.h" |
| | | #include <chrono> |
| | | |
| | | bool SendQ::TrySend(bhome_shm::ShmMsgQueue &mq) |
| | | int SendQ::DoSend1Remote(bhome_shm::ShmMsgQueue &mq, const Remote &remote, Array &arr) |
| | | { |
| | | if (out_.empty()) { |
| | | std::unique_lock<std::mutex> lock(mutex_); |
| | | out_.swap(in_); |
| | | 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 = [](MsgList &l) { |
| | | 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 SendOneRemote = [&](const Remote &remote, MsgList &msg_list) { |
| | | auto pos = FirstNotExpired(msg_list); |
| | | for (auto it = msg_list.begin(); it != pos; ++it) { |
| | | auto &info = it->data(); |
| | | if (info.on_expire_) { |
| | | info.on_expire_(info.msg_); |
| | | } |
| | | info.msg_.Release(mq.shm()); |
| | | 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(msg_list.end())); |
| | | for (int i = 0; i < n; ++i) { |
| | | auto &msg = pos->data().msg_; |
| | | if (msg.IsCounted()) { |
| | | msg.Release(mq.shm()); |
| | | } |
| | | ++pos; |
| | | 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; |
| | | } |
| | | |
| | | msg_list.erase(msg_list.begin(), 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) |
| | | { |
| | | size_t nsend = 0; |
| | | if (!out_.empty()) { |
| | | auto rec = out_.begin(); |
| | | do { |
| | | SendOneRemote(rec->first, rec->second); |
| | | nsend += DoSend1Remote(mq, rec->first, rec->second); |
| | | if (rec->second.empty()) { |
| | | rec = out_.erase(rec); |
| | | } else { |
| | |
| | | } |
| | | } while (rec != out_.end()); |
| | | } |
| | | |
| | | auto Collect = [&]() { |
| | | std::unique_lock<std::mutex> lock(mutex_); |
| | | 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(); |
| | | } |