test (20d00405) · Commits · tuxist / authdb

src/backends/cluster.cpp

+53 −58

Changes for src/backends/cluster.cpp: 53 added lines, 58 removed lines.
Original line number	Diff line number	Diff line
		@@ -70,13 +70,11 @@ namespace authdb {
		store.fetch(dgid, 0, local_data);
		} catch (...) {}

		// Only fetch from peers every 5 seconds to avoid blocking
		// the backend lock with slow network I/O on every request.
		// Only fetch from peers every 5 seconds
		std::vector<uint8_t> peer_data;
		auto now = std::chrono::steady_clock::now();
		bool do_peer_fetch = (now - _LastPeerFetch) >= std::chrono::seconds(5);

		// Network I/O runs WITHOUT _BufMtx to avoid blocking other threads
		if (do_peer_fetch) {
		auto &client = g_Cluster->getClient();
		if (client) {
		@@ -96,15 +94,11 @@ namespace authdb {
		}
		}

		// Only hold _BufMtx for the buffer merge — no network I/O here
		std::lock_guard<std::mutex> guard(_BufMtx);

		// Pick the buffer with the highest revision (most recent writes)
		// Merge into _Buffer — caller (lock()) holds _BufLock after this returns
		size_t local_rev = bufferRevision(local_data);
		size_t peer_rev = bufferRevision(peer_data);
		size_t cur_rev = bufferRevision(_Buffer);

		// Find the best candidate among local store, peer, and current in-memory
		std::vector<uint8_t> *best = nullptr;
		size_t best_rev = cur_rev;

		@@ -119,7 +113,6 @@ namespace authdb {

		if (best) {
		_Buffer = std::move(*best);
		// Update local cache with the better version
		store.store(dgid, 0, _Buffer.data(), _Buffer.size());
		} else if (_Buffer.empty() && !local_data.empty()) {
		_Buffer = std::move(local_data);
		@@ -129,52 +122,13 @@ namespace authdb {
		}

		void ClusterBackend::pushToCluster() {
		if (!g_Cluster \|\| !g_Cluster->isRunning()) {
		std::cerr << "ClusterBackend::pushToCluster: cluster not available" << std::endl;
		return;
		}

		// Copy buffer and state under _BufMtx, then release before network I/O
		std::vector<uint8_t> buf_copy;
		uint64_t dgid;
		{
		std::lock_guard<std::mutex> guard(_BufMtx);
		if (!_Dirty) {
		return;
		}
		// Bump revision so peers can detect newer data
		// Legacy entry point for constructor/destructor
		if (!_Dirty) return;
		newRevesion();
		dgid = domainGroupId();
		buf_copy = _Buffer;
		uint64_t dgid = domainGroupId();
		std::vector<uint8_t> buf_copy = _Buffer;
		_Dirty = false;
		}

		// Store in local block store (file or memory backed)
		auto &store = g_Cluster->getStore();
		if (!store.store(dgid, 0, buf_copy.data(), buf_copy.size())) {
		std::cerr << "ClusterBackend::pushToCluster: store.store() failed for group "
		<< std::hex << dgid << std::dec
		<< " (domain=" << _Domain << " bufsize=" << buf_copy.size() << ")"
		<< std::endl;
		}

		// Replicate raw data to all peer nodes — no _BufMtx held
		auto &client = g_Cluster->getClient();
		if (client) {
		try {
		auto &cmtx = g_Cluster->getClientMutex();
		std::unique_lock<std::timed_mutex> lock(cmtx, std::chrono::seconds(3));
		if (lock.owns_lock()) {
		client->replicate_to_peers(dgid, 0, buf_copy.data(), buf_copy.size());
		} else {
		std::cerr << "ClusterBackend::pushToCluster: skipped replication (lock busy)" << std::endl;
		}
		} catch (const netplus::NetException &e) {
		std::cerr << "Cluster replicate failed: " << e.what() << std::endl;
		} catch (const std::exception &e) {
		std::cerr << "Cluster replicate failed: " << e.what() << std::endl;
		}
		}
		pushToClusterAsync(dgid, buf_copy);
		}

		ClusterBackend::ClusterBackend(const char domain, size_t version, const char filepath)
		@@ -233,15 +187,56 @@ namespace authdb {
		}

		void ClusterBackend::lock() {
		// fetchFromCluster handles _BufMtx internally — only during buffer merge,
		// NOT during network I/O, so other threads aren't blocked.
		// 1. Acquire _BufMtx first to protect _Buffer during fetch merge
		_BufLock = std::unique_lock<std::mutex>(_BufMtx);
		// 2. Fetch latest data (network I/O + merge into _Buffer)
		// _BufLock stays held after this so read/write are safe
		fetchFromCluster();
		}

		void ClusterBackend::unlock() {
		// pushToCluster handles _BufMtx internally — copies buffer under lock,
		// then does network I/O without holding it.
		pushToCluster();
		// 1. Prepare push data while still holding _BufLock
		// pushToCluster() copies _Buffer internally before network I/O
		bool dirty = _Dirty;
		std::vector<uint8_t> buf_copy;
		uint64_t dgid = 0;
		if (dirty) {
		newRevesion();
		dgid = domainGroupId();
		buf_copy = _Buffer;
		_Dirty = false;
		}
		// 2. Release _BufMtx — other requests can now lock()
		if (_BufLock.owns_lock())
		_BufLock.unlock();
		// 3. Network I/O: push to peers (no locks held)
		if (dirty)
		pushToClusterAsync(dgid, buf_copy);
		}

		void ClusterBackend::pushToClusterAsync(uint64_t dgid, const std::vector<uint8_t> &buf) {
		if (!g_Cluster \|\| !g_Cluster->isRunning())
		return;

		auto &store = g_Cluster->getStore();
		if (!store.store(dgid, 0, buf.data(), buf.size())) {
		std::cerr << "ClusterBackend::pushToCluster: store.store() failed" << std::endl;
		}

		auto &client = g_Cluster->getClient();
		if (client) {
		try {
		auto &cmtx = g_Cluster->getClientMutex();
		std::unique_lock<std::timed_mutex> lock(cmtx, std::chrono::seconds(3));
		if (lock.owns_lock()) {
		client->replicate_to_peers(dgid, 0, buf.data(), buf.size());
		} else {
		std::cerr << "ClusterBackend::pushToCluster: skipped (lock busy)" << std::endl;
		}
		} catch (const std::exception &e) {
		std::cerr << "Cluster replicate failed: " << e.what() << std::endl;
		}
		}
		}

		void ClusterBackend::setPos(size_t pos) {

src/backends/cluster.h

+2 −0

Changes for src/backends/cluster.h: 2 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -65,6 +65,7 @@ namespace authdb {
		uint64_t domainGroupId() const;
		void fetchFromCluster();
		void pushToCluster();
		void pushToClusterAsync(uint64_t dgid, const std::vector<uint8_t> &buf);

		std::string _Domain;
		std::vector<uint8_t> _Buffer;
		@@ -72,5 +73,6 @@ namespace authdb {
		bool _Dirty;
		std::chrono::steady_clock::time_point _LastPeerFetch{};
		std::mutex _BufMtx;
		std::unique_lock<std::mutex> _BufLock; // held between lock() and unlock()
		};
		};