test (4424411f) · Commits · tuxist / libparitypp

include/paritypp/client.h

+5 −0

Original line number	Diff line number	Diff line
		@@ -12,6 +12,7 @@
		#include <memory>
		#include <functional>
		#include <fstream>
		#include <mutex>

		namespace paritypp {

		@@ -140,11 +141,15 @@ private:

		std::vector<quic_connection> connections_;
		netplus::socketwait wait_;
		mutable std::mutex mutex_; // thread-safety for all public methods

		void connect_to_node(size_t node_index);
		void ensure_connected(size_t node_index);
		bool authenticate_node(size_t node_index);

		// Unlocked node status probe (for internal parallel use)
		bool get_node_status_impl(size_t node_index, status_info& info);

		// Send a message and receive response on a node connection
		std::vector<uint8_t> send_recv(size_t node_index,
		msg_type type,

src/client.cpp

+20 −2

Original line number	Diff line number	Diff line
		@@ -773,6 +773,7 @@ bool client::retrieve_stripe(uint64_t group_id, uint32_t stripe_index,
		}

		void client::store(uint64_t group_id, const uint8_t* data, size_t data_len) {
		std::lock_guard<std::mutex> guard(mutex_);
		if (nodes_.size() < k_ + m_) {
		throw std::runtime_error("Not enough nodes for parity store ("
		+ std::to_string(nodes_.size()) + " < "
		@@ -887,6 +888,7 @@ public:

		private:
		void flush_stripe() {
		std::lock_guard<std::mutex> guard(cli_.mutex_);
		try {
		cli_.store_stripe(group_id_, stripe_idx_,
		stripe_buf_.data(), stripe_buf_.size());
		@@ -914,6 +916,7 @@ std::unique_ptr<store_session> client::begin_store(uint64_t group_id, size_t tot
		// ---- Stripe-based retrieve ----

		std::vector<uint8_t> client::retrieve(uint64_t group_id, size_t original_size) {
		std::lock_guard<std::mutex> guard(mutex_);
		std::vector<uint8_t> result;
		for (uint32_t s = 0; ; ++s) {
		std::vector<uint8_t> stripe;
		@@ -927,6 +930,7 @@ std::vector<uint8_t> client::retrieve(uint64_t group_id, size_t original_size) {
		}

		std::vector<uint8_t> client::retrieve(uint64_t group_id) {
		std::lock_guard<std::mutex> guard(mutex_);
		std::vector<uint8_t> result;
		result.reserve(STRIPE_SIZE * 4);

		@@ -953,6 +957,7 @@ std::vector<uint8_t> client::retrieve(uint64_t group_id) {

		std::vector<uint8_t> client::retrieve_range(uint64_t group_id,
		uint64_t offset, uint64_t length) {
		std::lock_guard<std::mutex> guard(mutex_);
		if (length == 0) return {};

		// The stored framed data is: [8-byte length header][raw data]
		@@ -988,6 +993,7 @@ std::vector<uint8_t> client::retrieve_range(uint64_t group_id,
		}

		void client::remove(uint64_t group_id) {
		std::lock_guard<std::mutex> guard(mutex_);
		std::vector<uint8_t> payload(8);
		protocol::encode_u64(payload.data(), group_id);

		@@ -1001,6 +1007,7 @@ void client::remove(uint64_t group_id) {
		}

		std::vector<std::pair<size_t, std::vector<uint32_t>>> client::list(uint64_t group_id) {
		std::lock_guard<std::mutex> guard(mutex_);
		std::vector<std::pair<size_t, std::vector<uint32_t>>> result;
		std::vector<uint8_t> payload(8);
		protocol::encode_u64(payload.data(), group_id);
		@@ -1029,6 +1036,7 @@ std::vector<std::pair<size_t, std::vector<uint32_t>>> client::list(uint64_t grou

		size_t client::replicate_to_peers(uint64_t group_id, uint32_t block_index,
		const uint8_t* data, size_t len) {
		std::lock_guard<std::mutex> guard(mutex_);
		static constexpr int MAX_RETRIES = 2;
		static constexpr int BASE_DELAY_MS = 100;
		size_t success_count = 0;
		@@ -1089,6 +1097,7 @@ size_t client::replicate_to_peers(uint64_t group_id, uint32_t block_index,

		bool client::fetch_from_peers(uint64_t group_id, uint32_t block_index,
		std::vector<uint8_t>& out) {
		std::lock_guard<std::mutex> guard(mutex_);
		static constexpr int MAX_RETRIES = 2;
		static constexpr int BASE_DELAY_MS = 100;

		@@ -1134,7 +1143,7 @@ bool client::fetch_from_peers(uint64_t group_id, uint32_t block_index,
		return false;
		}

		bool client::get_node_status(size_t node_index, status_info& info) {
		bool client::get_node_status_impl(size_t node_index, status_info& info) {
		if (node_index >= nodes_.size()) return false;
		try {
		auto resp = send_recv(node_index, msg_type::GET_STATUS, {});
		@@ -1155,7 +1164,13 @@ bool client::get_node_status(size_t node_index, status_info& info) {
		}
		}

		bool client::get_node_status(size_t node_index, status_info& info) {
		std::lock_guard<std::mutex> guard(mutex_);
		return get_node_status_impl(node_index, info);
		}

		bool client::list_groups_on_node(size_t node_index, std::vector<uint64_t>& groups) {
		std::lock_guard<std::mutex> guard(mutex_);
		if (node_index >= nodes_.size()) return false;
		if (static_cast<int>(node_index) == local_node_index_ && local_store_) {
		groups = local_store_->list_groups();
		@@ -1181,6 +1196,7 @@ bool client::list_groups_on_node(size_t node_index, std::vector<uint64_t>& group
		}

		client::cluster_health client::get_cluster_status() {
		std::lock_guard<std::mutex> guard(mutex_);
		cluster_health health;
		health.nodes_total = static_cast<uint32_t>(nodes_.size());
		health.nodes_online = 0;
		@@ -1198,7 +1214,7 @@ client::cluster_health client::get_cluster_status() {
		for (size_t i = 0; i < nodes_.size(); i++) {
		results[i].index = i;
		futures.push_back(std::async(std::launch::async, [this, i, &results]() {
		results[i].reachable = get_node_status(i, results[i].info);
		results[i].reachable = get_node_status_impl(i, results[i].info);
		}));
		}
		for (auto& f : futures) f.get();
		@@ -1226,11 +1242,13 @@ client::cluster_health client::get_cluster_status() {
		}

		void client::set_local_node(size_t node_index, std::shared_ptr<block_store> store) {
		std::lock_guard<std::mutex> guard(mutex_);
		local_node_index_ = static_cast<int>(node_index);
		local_store_ = std::move(store);
		}

		size_t client::vacuum_all_nodes() {
		std::lock_guard<std::mutex> guard(mutex_);
		size_t ok_count = 0;
		for (size_t i = 0; i < nodes_.size(); i++) {
		try {