version0/crdt/list.hpp

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

vim_research/test2.cpp

@@ -80,7 +80,7 @@ namespace std {
 
 	template<typename 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);
 		} else {
 			nlohmann_json_t = std::nullopt;
@@ -117,23 +117,14 @@ namespace std {
 		} else if (std::holds_alternative<Doc::ListType::OpDel>(nlohmann_json_t)) {
 			nlohmann_json_j["t"] = "del";
 			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) {
-		if (nlohmann_json_j.is_null()) {
-			std::cerr << "got null j\n";
-			return;
-		}
-
 		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") {
-			nlohmann_json_t = static_cast<Doc::ListType::OpDel>(nlohmann_json_j.at("d"));
-		} else {
-			assert(false && "missing op type");
+			nlohmann_json_j.at("d").get_to(std::get<Doc::ListType::OpDel>(nlohmann_json_t));
 		}
 	}
 } // namespace std
@@ -484,44 +475,7 @@ void toxThread(SharedContext* ctx) {
 				std::cout << "tox connected to group\n";
 			}
 		} else { // do the thing
-			{ // 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);
-				}
-			}
+			// pump from buffer to staging
 
 			// request missing in buffer
 
@@ -552,7 +506,6 @@ void toxThread(SharedContext* ctx) {
 						f_pkg.seq = ctx->command_frontier.at(agent_local);
 
 						c_pkg = ctx->command_lists[agent_local][f_pkg.seq];
-						assert(!c_pkg.ops.empty());
 					}
 
 					{ // gossip
@@ -819,20 +772,10 @@ int main(void) {
 					for (const auto& [agent, staging_seq] : ctx.staging_frontier) {
 						// check if remote newer
 						if (!ctx.command_frontier.count(agent) || ctx.command_frontier.at(agent) < staging_seq) {
-							uint64_t seq {0};
-							if (ctx.command_frontier.count(agent)) {
-								seq = ctx.command_frontier[agent] + 1;
-							}
-							for (; seq <= staging_seq; seq++) {
+							for (uint64_t seq = ctx.command_frontier[agent]; seq <= staging_seq; seq++) {
 
 								// !! this can get expensive, while we are holding locks :/
 								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");
 							}
 
@@ -907,34 +850,25 @@ int main(void) {
 				}
 				assert(doc.getText() == new_text);
 
-				if (!ops.empty()) {
 				// TODO: make something less locky
 				// TODO: make method
 				{
 					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];
-
 					uint64_t seq {0};
 					if (ctx.command_frontier.count(ctx.agent)) { // get last own seq
 						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
+						ops
 					});
 					ctx.command_frontier[ctx.agent] = seq;
 				}
-					}
 				ctx.should_gossip_local.store(true);
-				}
 			} else {
 				std::cout << "unknown command '" << command << "'\n";
 			}