Team2890/allwpilib
4
0
Fork 0
mirror of https://github.com/wpilibsuite/allwpilib synced 2026-06-26 01:51:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Dispatcher: Implement transmit combining.

Browse Source 
This handles most of the major cases, but not a loop of delete/assign, which
should be uncommon anyway.
This commit is contained in:
Peter Johnson
2015-08-02 23:21:23 -07:00
parent 53a0531def
commit a86f65db1e
3 changed files with 98 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
src/Dispatcher.cpp
View File

@@ -181,6 +181,7 @@ void DispatcherBase::DispatchThreadMain() {
connections.push_back(ConnectionRef());
connections.back().net = conn.net.get();
connections.back().outgoing.swap(conn.outgoing);
conn.last_update.resize(0); // clear "previous" updates
}
if (!m_server && conn.net->state() == NetworkConnection::kDead)
reconnect = true;
@@ -192,8 +193,6 @@ void DispatcherBase::DispatchThreadMain() {
}
}
// scan outgoing messages to remove unnecessary updates
// send outgoing messages
for (auto& conn : connections) {
if (!conn.outgoing.empty())
@@ -202,6 +201,99 @@ void DispatcherBase::DispatchThreadMain() {
}
}
void DispatcherBase::Connection::QueueOutgoing(std::shared_ptr<Message> msg) {
// Merge with previous. One case we don't combine: delete/assign loop.
switch (msg->type()) {
case Message::kEntryAssign:
case Message::kEntryUpdate: {
// don't do this for unassigned id's
unsigned int id = msg->id();
if (id == 0xffff) {
outgoing.push_back(msg);
break;
}
if (id < last_update.size() && last_update[id].first != 0) {
// overwrite the previous one for this id
auto& oldmsg = outgoing[last_update[id].first - 1];
if (oldmsg && oldmsg->Is(Message::kEntryAssign) &&
msg->Is(Message::kEntryUpdate)) {
// need to update assignment with new seq_num and value
oldmsg = Message::EntryAssign(oldmsg->str(), id, msg->seq_num_uid(),
msg->value(), oldmsg->flags());
} else
oldmsg = msg; // easy update
} else {
// new, but remember it
std::size_t pos = outgoing.size();
outgoing.push_back(msg);
if (id >= last_update.size()) last_update.resize(id + 1);
last_update[id].first = pos + 1;
}
break;
}
case Message::kEntryDelete: {
// don't do this for unassigned id's
unsigned int id = msg->id();
if (id == 0xffff) {
outgoing.push_back(msg);
break;
}
// clear previous updates
if (id < last_update.size()) {
if (last_update[id].first != 0) {
outgoing[last_update[id].first - 1].reset();
last_update[id].first = 0;
}
if (last_update[id].second != 0) {
outgoing[last_update[id].second - 1].reset();
last_update[id].second = 0;
}
}
// add deletion
outgoing.push_back(msg);
break;
}
case Message::kFlagsUpdate: {
// don't do this for unassigned id's
unsigned int id = msg->id();
if (id == 0xffff) {
outgoing.push_back(msg);
break;
}
if (id < last_update.size() && last_update[id].second != 0) {
// overwrite the previous one for this id
outgoing[last_update[id].second - 1] = msg;
} else {
// new, but remember it
std::size_t pos = outgoing.size();
outgoing.push_back(msg);
if (id >= last_update.size()) last_update.resize(id + 1);
last_update[id].second = pos + 1;
}
break;
}
case Message::kClearEntries: {
// knock out all previous assigns/updates!
for (auto& i : outgoing) {
if (!i) continue;
auto t = i->type();
if (t == Message::kEntryAssign || t == Message::kEntryUpdate ||
t == Message::kFlagsUpdate || t == Message::kEntryDelete ||
t == Message::kClearEntries)
i.reset();
}
last_update.resize(0);
outgoing.push_back(msg);
break;
}
default:
outgoing.push_back(msg);
break;
}
}
void DispatcherBase::QueueOutgoing(std::shared_ptr<Message> msg,
NetworkConnection* only,
NetworkConnection* except) {
@@ -212,7 +304,7 @@ void DispatcherBase::QueueOutgoing(std::shared_ptr<Message> msg,
auto state = conn.net->state();
if (state != NetworkConnection::kSynchronized &&
state != NetworkConnection::kActive) continue;
conn.outgoing.push_back(msg);
conn.QueueOutgoing(msg);
}
}