vim_research/test2.cpp

@@ -1,4 +1,3 @@
-#include "toxcore/tox.h"
 #include <crdt/text_document.hpp>
 #include <nlohmann/json.hpp>
 
@@ -11,7 +10,6 @@ extern "C" {
 #include <optional>
 #include <memory>
 #include <unordered_map>
-#include <unordered_set>
 #include <string_view>
 #include <variant>
 #include <thread>
@@ -62,7 +60,7 @@ using Doc = GreenCRDT::TextDocument<Agent>;
 using ListType = Doc::ListType;
 
 struct Command {
-	Agent agent;
+	Agent actor;
 	uint64_t seq {0}; // independed of the ops inside, theoretically
 	//...
 	std::vector<Doc::Op> ops;
@@ -130,7 +128,7 @@ namespace std {
 } // namespace std
 
 NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE(Command,
-	agent,
+	actor,
 	seq,
 	ops
 )
@@ -284,22 +282,14 @@ struct SharedContext {
 	ToxPubKey agent;
 	std::promise<void> agent_set;
 
-	std::mutex command_lists_mutex; // for list and frontier!!
+	// TODO: this is inefficent
-	std::unordered_map<ToxPubKey, std::unordered_map<uint64_t, Command>> command_lists;
+	std::mutex command_lists_mutex;
-	std::unordered_map<ToxPubKey, uint64_t> command_frontier; // last applied seq
+	std::unordered_map<ToxPubKey, std::vector<Command>> command_lists;
+	// remote op queue for receive
+	// local op list for remote lookups
 
-	// contains remote changes we can apply in the main thread
+	// last seq for all known agents
-	std::mutex staging_mutex;
+	// bool dirty
-	std::unordered_map<ToxPubKey, uint64_t> staging_frontier; // last seq we have in command_lists, via tox
-	// (can be lower then command_frontier for local agent
-
-	// contains remote changes with missing parent seq
-	// could merge into comamnd_lists
-	std::unordered_map<ToxPubKey, std::unordered_map<uint64_t, Command>> buffer;
-
-	std::atomic_bool should_gossip_local{false}; // local changes (set by main thread, reset by tox thread)
-	std::unordered_set<ToxPubKey> should_gossip_remote; // list of ids we have new seq for (only modified by tox thread)
-	std::unordered_map<ToxPubKey, uint64_t> heard_gossip; // seq frontiers we have heard about
 
 	Tox* tox {nullptr};
 	bool tox_dht_online {false};
@@ -309,51 +299,6 @@ struct SharedContext {
 
 namespace tox {
 
-namespace pkg {
-
-	enum PKGID : uint8_t {
-		FRONTIER = 32,
-		REQUEST_FRONTIER,
-
-		COMMAND,
-		REQUEST_COMMANDS,
-	};
-
-	// send the currently last known seq you have (excluding buffer)
-	struct Frontier {
-		Agent agent;
-		uint64_t seq{0};
-	};
-
-	NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE(Frontier,
-		agent,
-		seq
-	)
-
-	// request the last known seq another peer has for agent
-	struct RequestFrontier {
-		Agent agent;
-	};
-
-	NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE(RequestFrontier,
-		agent
-	)
-
-	using Command = ::Command;
-
-	// request every command for agent after seq (inclusive)
-	struct RequestCommands {
-		Agent agent;
-		uint64_t seq{0};
-	};
-
-	NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE(RequestCommands,
-		agent,
-		seq
-	)
-
-} // namespace pkg
-
 static std::vector<uint8_t> hex2bin(const std::string& str) {
 	std::vector<uint8_t> bin{};
 	bin.resize(str.size()/2, 0);
@@ -437,9 +382,6 @@ void toxThread(SharedContext* ctx) {
 		}
 	}
 
-	// "thread local"
-	Agent agent_local;
-
 	//tox_group_self_get_public_key()
 	//tox_group_send_custom_packet()
 	//tox_group_send_custom_private_packet()
@@ -466,7 +408,6 @@ void toxThread(SharedContext* ctx) {
 					ctx->agent_set.set_value();
 					return; // fuck everything
 				}
-				agent_local = ctx->agent;
 				ctx->agent_set.set_value();
 			}
 		} else if (!ctx->tox_group_online) { // then wait for group to connect
@@ -475,66 +416,9 @@ void toxThread(SharedContext* ctx) {
 				std::cout << "tox connected to group\n";
 			}
 		} else { // do the thing
-			// pump from buffer to staging
+			// staging?
-
-			// request missing in buffer
-
-			// request frontier (implicit list of agents)
-			// only every couple of second, can get large
-			// OR get back random agent and do it often
-
 			// handle requests
-
+			// send tip (prio self)
-			{ // gossip frontier
-				// for mutex locking simplicity this is an either-or
-				if (ctx->should_gossip_local.exchange(false)) {
-					pkg::Frontier f_pkg{
-						agent_local,
-						0u
-					};
-
-					pkg::Command c_pkg{
-						agent_local,
-						0u,
-						{}
-					};
-
-					{ // lock
-						std::lock_guard lg{ctx->command_lists_mutex};
-						assert(ctx->command_frontier.count(agent_local));
-
-						f_pkg.seq = ctx->command_frontier.at(agent_local);
-
-						c_pkg = ctx->command_lists[agent_local][f_pkg.seq];
-					}
-
-					{ // gossip
-						std::vector<uint8_t> data = nlohmann::json::to_msgpack(f_pkg);
-						// prepend pkgid
-						data.emplace(data.begin(), static_cast<uint8_t>(pkg::PKGID::FRONTIER));
-
-						if (!tox_group_send_custom_packet(ctx->tox, 0, true, data.data(), data.size(), nullptr)) {
-							std::cerr << "failed to send gossip packet of local agent\n";
-							// TODO: set should_gossip_local back to true?
-						} else {
-							std::cout << "sent gossip of local agent\n";
-						}
-					}
-					{ // command
-						std::vector<uint8_t> data = nlohmann::json::to_msgpack(c_pkg);
-						// prepend pkgid
-						data.emplace(data.begin(), static_cast<uint8_t>(pkg::PKGID::COMMAND));
-
-						if (!tox_group_send_custom_packet(ctx->tox, 0, true, data.data(), data.size(), nullptr)) {
-							std::cerr << "failed to send command packet of local agent\n";
-						} else {
-							std::cout << "sent command of local agent\n";
-						}
-					}
-				} else if (!ctx->should_gossip_remote.empty()) {
-					std::lock_guard lg{ctx->command_lists_mutex};
-				}
-			}
 		}
 
 		std::this_thread::sleep_for(20ms);
@@ -619,43 +503,28 @@ int main(void) {
 	std::cout << "waiting for agent id\n";
 	ctx.agent_set.get_future().wait();
 	if (ctx.should_quit) {
-		tox_thread.join(); // wait for thread
 		return -1;
 	}
 
 	std::cout << "starting vim ipc server\n";
 
 	if (zed_net_init() != 0) {
-		ctx.should_quit.store(true);
 		std::cerr << "zed_net_init failed: " << zed_net_get_error() << "\n";
-		tox_thread.join(); // wait for thread
 		return -1;
 	}
 
 	std::cout << "initialized zed_net\n";
 
-	const uint16_t port_start {1337};
+	const uint16_t port {1337};
-	const uint16_t port_end {1437};
-	uint16_t port = port_start;
 	zed_net_socket_t listen_socket;
-	bool found_free_port {false};
+	if (zed_net_tcp_socket_open(
-	for (; port <= port_end; port++) {
+		&listen_socket,
-		if (zed_net_tcp_socket_open(
+		port, // port
-			&listen_socket,
+		0, // non blocking
-			port, // port
+		1 // listen
-			0, // non blocking
+	) != 0) {
-			1 // listen
-		) == 0) {
-			found_free_port = true;
-			break;
-		}
-	}
-
-	if (!found_free_port) {
-		ctx.should_quit.store(true);
 		std::cerr << "zed_net_tcp_socket_open failed: " << zed_net_get_error() << "\n";
 		zed_net_shutdown();
-		tox_thread.join(); // wait for thread
 		return -1;
 	}
 
@@ -669,11 +538,9 @@ int main(void) {
 	zed_net_socket_t remote_socket;
 	zed_net_address_t remote_address;
 	if (zed_net_tcp_accept(&listen_socket, &remote_socket, &remote_address) != 0) {
-		ctx.should_quit.store(true);
 		std::cerr << "zed_net_tcp_accept failed: " << zed_net_get_error() << "\n";
 		zed_net_socket_close(&listen_socket);
 		zed_net_shutdown();
-		tox_thread.join(); // wait for thread
 		return -1;
 	}
 
@@ -693,12 +560,10 @@ int main(void) {
 		int64_t bytes_received {0};
 		bytes_received = zed_net_tcp_socket_receive(&remote_socket, buffer->data(), buffer->size());
 		if (bytes_received < 0) {
-			ctx.should_quit.store(true);
 			std::cerr << "zed_net_tcp_socket_receive failed: " << zed_net_get_error() << "\n";
 			zed_net_socket_close(&remote_socket);
 			zed_net_socket_close(&listen_socket);
 			zed_net_shutdown();
-			tox_thread.join(); // wait for thread
 			return -1;
 		} else if (bytes_received == 0) {
 			std::cout << "got 0 bytes?\n";
@@ -762,31 +627,9 @@ int main(void) {
 				// apply changes (some) and gen vim inserts
 				std::cout << "got fetch changes\n";
 
-				bool changes {false};
-				{ // apply changes
-					// TODO: make less locky, we dont need to hold both at the same time etc
-					//std::lock_guard lg {ctx.staging_mutex};
-					std::scoped_lock sl {ctx.staging_mutex, ctx.command_lists_mutex};
-
-					// TODO: limit number of ops for responsiveness
-					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) {
-							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);
-								assert(r && "this should not happen");
-							}
-
-							ctx.command_frontier[agent] = staging_seq;
-							changes = true;
-						}
-					}
-				}
-
 				auto j_res_line_list = nlohmann::json::array();
-				if (changes) { // external changes
+
+				if (true) { // external changes
 					const auto crdt_text = doc.getText();
 					std::string_view text_view {crdt_text};
 					for (int64_t i = 1; ; i++) {
@@ -849,26 +692,6 @@ int main(void) {
 					std::cout << "\n";
 				}
 				assert(doc.getText() == new_text);
-
-				// 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());
-
-					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;
-					}
-					local_command_list.emplace(seq, Command{
-						ctx.agent,
-						seq,
-						ops
-					});
-					ctx.command_frontier[ctx.agent] = seq;
-				}
-				ctx.should_gossip_local.store(true);
 			} else {
 				std::cout << "unknown command '" << command << "'\n";
 			}
@@ -897,66 +720,14 @@ static void self_connection_status_cb(Tox*, TOX_CONNECTION connection_status, vo
 	std::cout << "self_connection_status_cb " << connection_status << "\n";
 }
 
-static void handle_pkg(SharedContext& ctx, const uint8_t* data, size_t length) {
-	if (length < 2) {
-		std::cerr << "got too short pkg " << length << "\n";
-		return;
-	}
-
-	pkg::PKGID pkg_id = static_cast<pkg::PKGID>(data[0]);
-	const auto p_j = nlohmann::json::from_msgpack(data+1, data+1 + (length-1), true, false);
-	if (p_j.is_discarded()) {
-		std::cerr << "got invalid msgpack for " << pkg_id << "\n";
-		return;
-	}
-
-	std::cout << "pkg " << pkg_id << " j:" << p_j.dump() << "\n";
-
-	switch (pkg_id) {
-		case pkg::PKGID::FRONTIER: {
-			pkg::Frontier pkg = p_j;
-
-			if (!ctx.heard_gossip.count(pkg.agent) || ctx.heard_gossip[pkg.agent] < pkg.seq) {
-				ctx.heard_gossip[pkg.agent] = pkg.seq;
-				std::cout << "new seq " << pkg.seq << " from " << pkg.agent << "\n";
-			}
-			break;
-		}
-		case pkg::PKGID::REQUEST_FRONTIER: {
-			pkg::RequestFrontier pkg = p_j;
-			break;
-		}
-		case pkg::PKGID::COMMAND: {
-			pkg::Command pkg = p_j;
-
-			// push to buffer, if not in buffer
-			if (!ctx.buffer[pkg.agent].count(pkg.seq)) {
-				ctx.buffer[pkg.agent].emplace(pkg.seq, pkg);
-				std::cout << "pushed to buffer " << pkg.seq << " from " << pkg.agent << "\n";
-			}
-			// TODO: notify something?
-			break;
-		}
-		case pkg::PKGID::REQUEST_COMMANDS: {
-			pkg::RequestCommands pkg = p_j;
-			break;
-		}
-		default:
-			std::cerr << "unknown pkg id " << pkg_id << "\n";
-			break;
-	}
-}
-
 static void group_custom_packet_cb(Tox*, uint32_t group_number, uint32_t peer_id, const uint8_t* data, size_t length, void* user_data) {
 	std::cout << "group_custom_packet_cb\n";
 	SharedContext& ctx = *static_cast<SharedContext*>(user_data);
-	handle_pkg(ctx, data, length);
 }
 
 static void group_custom_private_packet_cb(Tox*, uint32_t group_number, uint32_t peer_id, const uint8_t* data, size_t length, void* user_data) {
 	std::cout << "group_custom_private_packet_cb\n";
 	SharedContext& ctx = *static_cast<SharedContext*>(user_data);
-	handle_pkg(ctx, data, length);
 }
 
 }