Compare commits

..

No commits in common. "cb0c2642f872874a74ea239930fea49795926267" and "e961b8aec318c2389a9122e4c6f61d9f36cc2345" have entirely different histories.

2 changed files with 26 additions and 92 deletions

View File

@ -71,7 +71,7 @@ struct List {
std::map<AgentType, uint64_t> last_seen_seq; std::map<AgentType, uint64_t> last_seen_seq;
std::optional<size_t> findIdx(const ListID& list_id) const { std::optional<size_t> findIdx(const ListID& list_id) const {
//verify(); // too expensive verify();
for (size_t i = 0; i < list.size(); i++) { for (size_t i = 0; i < list.size(); i++) {
if (list[i].id == list_id) { if (list[i].id == list_id) {
return i; return i;

View File

@ -80,7 +80,7 @@ namespace std {
template<typename T> template<typename T>
static void from_json(const nlohmann::json& nlohmann_json_j, std::optional<T>& nlohmann_json_t) { static void from_json(const nlohmann::json& nlohmann_json_j, std::optional<T>& nlohmann_json_t) {
if (!nlohmann_json_j.is_null()) { if (nlohmann_json_j != nullptr) {
nlohmann_json_t = static_cast<T>(nlohmann_json_j); nlohmann_json_t = static_cast<T>(nlohmann_json_j);
} else { } else {
nlohmann_json_t = std::nullopt; nlohmann_json_t = std::nullopt;
@ -117,23 +117,14 @@ namespace std {
} else if (std::holds_alternative<Doc::ListType::OpDel>(nlohmann_json_t)) { } else if (std::holds_alternative<Doc::ListType::OpDel>(nlohmann_json_t)) {
nlohmann_json_j["t"] = "del"; nlohmann_json_j["t"] = "del";
nlohmann_json_j["d"] = std::get<Doc::ListType::OpDel>(nlohmann_json_t); nlohmann_json_j["d"] = std::get<Doc::ListType::OpDel>(nlohmann_json_t);
} else {
assert(false && "missing op type");
} }
} }
static void from_json(const nlohmann::json& nlohmann_json_j, Doc::Op& nlohmann_json_t) { static void from_json(const nlohmann::json& nlohmann_json_j, Doc::Op& nlohmann_json_t) {
if (nlohmann_json_j.is_null()) {
std::cerr << "got null j\n";
return;
}
if (nlohmann_json_j.at("t") == "add") { if (nlohmann_json_j.at("t") == "add") {
nlohmann_json_t = static_cast<Doc::ListType::OpAdd>(nlohmann_json_j.at("d")); nlohmann_json_j.at("d").get_to(std::get<Doc::ListType::OpAdd>(nlohmann_json_t));
} else if (nlohmann_json_j.at("t") == "del") { } else if (nlohmann_json_j.at("t") == "del") {
nlohmann_json_t = static_cast<Doc::ListType::OpDel>(nlohmann_json_j.at("d")); nlohmann_json_j.at("d").get_to(std::get<Doc::ListType::OpDel>(nlohmann_json_t));
} else {
assert(false && "missing op type");
} }
} }
} // namespace std } // namespace std
@ -484,44 +475,7 @@ void toxThread(SharedContext* ctx) {
std::cout << "tox connected to group\n"; std::cout << "tox connected to group\n";
} }
} else { // do the thing } else { // do the thing
{ // pump from buffer to staging // pump from buffer to staging
const size_t max_commands = 1;
size_t number_of_commands_done = 0;
std::lock_guard lg_staging{ctx->staging_mutex};
for (auto& [agent, buffer] : ctx->buffer) {
if (agent == agent_local) {
// skip ? self
continue;
}
if (number_of_commands_done >= max_commands) {
break;
}
// determain the seq we are looking for in buffer
uint64_t seq {0};
if (ctx->staging_frontier.count(agent)) {
seq = ctx->staging_frontier.at(agent) + 1;
}
if (!buffer.count(seq)) { // not in buffer, skip
continue;
}
// this can lead to dead locks, if other code does this wrong
std::lock_guard lg{ctx->command_lists_mutex};
for (; buffer.count(seq); seq++) {
ctx->command_lists[agent][seq] = buffer.at(seq);
number_of_commands_done++;
if (number_of_commands_done >= max_commands) {
break;
}
}
ctx->staging_frontier[agent] = seq;
ctx->should_gossip_remote.emplace(agent);
}
}
// request missing in buffer // request missing in buffer
@ -552,7 +506,6 @@ void toxThread(SharedContext* ctx) {
f_pkg.seq = ctx->command_frontier.at(agent_local); f_pkg.seq = ctx->command_frontier.at(agent_local);
c_pkg = ctx->command_lists[agent_local][f_pkg.seq]; c_pkg = ctx->command_lists[agent_local][f_pkg.seq];
assert(!c_pkg.ops.empty());
} }
{ // gossip { // gossip
@ -819,20 +772,10 @@ int main(void) {
for (const auto& [agent, staging_seq] : ctx.staging_frontier) { for (const auto& [agent, staging_seq] : ctx.staging_frontier) {
// check if remote newer // check if remote newer
if (!ctx.command_frontier.count(agent) || ctx.command_frontier.at(agent) < staging_seq) { if (!ctx.command_frontier.count(agent) || ctx.command_frontier.at(agent) < staging_seq) {
uint64_t seq {0}; for (uint64_t seq = ctx.command_frontier[agent]; seq <= staging_seq; seq++) {
if (ctx.command_frontier.count(agent)) {
seq = ctx.command_frontier[agent] + 1;
}
for (; seq <= staging_seq; seq++) {
// !! this can get expensive, while we are holding locks :/ // !! this can get expensive, while we are holding locks :/
bool r = doc.apply(ctx.command_lists.at(agent).at(seq).ops); bool r = doc.apply(ctx.command_lists.at(agent).at(seq).ops);
if (!r) {
std::cout << "faild to apply:\n";
for (const auto& op : ctx.command_lists.at(agent).at(seq).ops) {
std::cout << " " << op << "\n";
}
}
assert(r && "this should not happen"); assert(r && "this should not happen");
} }
@ -891,10 +834,10 @@ int main(void) {
//std::cout << "new_text:\n" << new_text << "\n"; //std::cout << "new_text:\n" << new_text << "\n";
//std::cout << "old_text:\n" << doc.getText() << "\n"; //std::cout << "old_text:\n" << doc.getText() << "\n";
std::cout << "doc state: "; std::cout << "doc state: ";
for (const auto& e : doc.state.list) { for (const auto& e : doc.state.list) {
std::cout << e << " "; std::cout << e << " ";
} }
std::cout << "\n"; std::cout << "\n";
const auto ops = doc.merge(new_text); const auto ops = doc.merge(new_text);
if (!ops.empty()) { if (!ops.empty()) {
@ -907,34 +850,25 @@ int main(void) {
} }
assert(doc.getText() == new_text); assert(doc.getText() == new_text);
if (!ops.empty()) { // TODO: make something less locky
// TODO: make something less locky // TODO: make method
// TODO: make method {
{ std::lock_guard mg{ctx.command_lists_mutex};
std::lock_guard mg{ctx.command_lists_mutex}; assert(ctx.command_lists.size() == ctx.command_frontier.size());
//assert(ctx.command_lists.size() == ctx.command_frontier.size());
auto& local_command_list = ctx.command_lists[ctx.agent]; auto& local_command_list = ctx.command_lists[ctx.agent];
uint64_t seq {0};
uint64_t seq {0}; if (ctx.command_frontier.count(ctx.agent)) { // get last own seq
if (ctx.command_frontier.count(ctx.agent)) { // get last own seq seq = ctx.command_frontier[ctx.agent] + 1;
seq = ctx.command_frontier[ctx.agent] + 1;
}
const size_t max_ops {5}; // limit ops per command so we can fit them into packets
for (size_t i = 0; i < ops.size(); i+=max_ops, seq++) {
std::vector<Doc::Op> tmp_ops {ops.cbegin()+i, ops.cbegin()+i+1};
assert(!tmp_ops.empty());
local_command_list.emplace(seq, Command{
ctx.agent,
seq,
tmp_ops
});
ctx.command_frontier[ctx.agent] = seq;
}
} }
ctx.should_gossip_local.store(true); local_command_list.emplace(seq, Command{
ctx.agent,
seq,
ops
});
ctx.command_frontier[ctx.agent] = seq;
} }
ctx.should_gossip_local.store(true);
} else { } else {
std::cout << "unknown command '" << command << "'\n"; std::cout << "unknown command '" << command << "'\n";
} }