test (c7645920) · Commits · tuxist / libparitypp

src/client.cpp

+53 −27

Original line number	Diff line number	Diff line
		@@ -17,6 +17,17 @@
		#define DBG_LOG(x) do { std::cerr << x; } while(0)
		#endif

		// Milliseconds remaining until deadline, clamped to [0, max_poll_ms].
		// Returns 0 when deadline has passed.
		static int ms_left(const std::chrono::steady_clock::time_point &deadline,
		int max_poll_ms = 100) {
		auto now = std::chrono::steady_clock::now();
		if (now >= deadline) return 0;
		int ms = static_cast<int>(
		std::chrono::duration_cast<std::chrono::milliseconds>(deadline - now).count());
		return std::min(ms, max_poll_ms);
		}

		namespace paritypp {

		client::client(size_t data_blocks, size_t parity_blocks,
		@@ -47,11 +58,12 @@ void client::connect_to_node(size_t node_index) {
		nc.socket->connect(nodes_[node_index].address, nodes_[node_index].port, false);

		// Wait for handshake using event-driven wait (epoll/kqueue/IOCP/poll)
		for (int i = 0; i < 50; i++) {
		auto hs_deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
		while (int poll_ms = ms_left(hs_deadline)) {
		if (nc.socket->hasPendingWrite()) {
		if (!wait_.waitWrite(*nc.socket, 100)) continue;
		if (!wait_.waitWrite(*nc.socket, poll_ms)) continue;
		} else {
		if (!wait_.waitRead(*nc.socket, 100)) continue;
		if (!wait_.waitRead(*nc.socket, poll_ms)) continue;
		}
		nc.socket->pumpNetwork(MSG_DONTWAIT);
		if (nc.socket->getHandshakeDone()) {
		@@ -125,9 +137,11 @@ std::vector<uint8_t> client::send_recv(size_t node_index,
		uint64_t stream = conn.openStream(true);
		auto msg = protocol::encode_message(type, payload);

		auto sr_deadline = std::chrono::steady_clock::now() + std::chrono::seconds(15);

		// Non-blocking send: wait for writable, then send
		for (int sw = 0; sw < 150; sw++) {
		if (wait_.waitWrite(conn, 100)) {
		while (int poll_ms = ms_left(sr_deadline)) {
		if (wait_.waitWrite(conn, poll_ms)) {
		conn.sendStreamData(stream, msg, true);
		break;
		}
		@@ -136,8 +150,8 @@ std::vector<uint8_t> client::send_recv(size_t node_index,

		// Wait for response using event-driven waitRead (epoll/kqueue/IOCP/poll)
		std::vector<uint8_t> response;
		for (int attempt = 0; attempt < 150; attempt++) {
		if (!wait_.waitRead(conn, 100)) continue;
		while (int poll_ms = ms_left(sr_deadline)) {
		if (!wait_.waitRead(conn, poll_ms)) continue;
		conn.pumpNetwork(MSG_DONTWAIT);
		if (conn.hasStreamData(stream)) {
		// Read header
		@@ -153,9 +167,10 @@ std::vector<uint8_t> client::send_recv(size_t node_index,
		std::memcpy(response.data(), hdr, protocol::HEADER_SIZE);

		size_t total_read = 0;
		int body_attempts = 0;
		while (total_read < resp_len && body_attempts++ < 150) {
		if (!wait_.waitRead(conn, 100)) continue;
		while (total_read < resp_len) {
		int body_ms = ms_left(sr_deadline);
		if (body_ms <= 0) break;
		if (!wait_.waitRead(conn, body_ms)) continue;
		conn.pumpNetwork(MSG_DONTWAIT);
		size_t rd = conn.recvStreamData(stream,
		response.data() + protocol::HEADER_SIZE + total_read,
		@@ -258,6 +273,8 @@ void client::store_stripe(uint64_t group_id, uint32_t stripe_index,

		size_t n = k_ + m_;

		auto stripe_deadline = std::chrono::steady_clock::now() + std::chrono::seconds(15);

		// Pending stream info per node-slot
		struct Pending {
		size_t node_index;
		@@ -298,8 +315,8 @@ void client::store_stripe(uint64_t group_id, uint32_t stripe_index,
		pending[i].stream = stream;

		// Non-blocking send
		for (int sw = 0; sw < 150; sw++) {
		if (wait_.waitWrite(conn, 100)) {
		while (int poll_ms = ms_left(stripe_deadline)) {
		if (wait_.waitWrite(conn, poll_ms)) {
		conn.sendStreamData(stream, msg, true);
		// Pump immediately to flush queued UDP datagrams
		conn.pumpNetwork(MSG_DONTWAIT);
		@@ -356,7 +373,9 @@ void client::store_stripe(uint64_t group_id, uint32_t stripe_index,
		try { connections_[p.node_index].socket->pumpNetwork(MSG_DONTWAIT); } catch (...) {}
		}

		for (int attempt = 0; attempt < 300 && remaining > 0; ++attempt) {
		for (;;) {
		int poll_ms = ms_left(stripe_deadline);
		if (poll_ms <= 0 \|\| remaining == 0) break;
		// Check buffered data first before blocking
		for (auto& p : pending) {
		if (!p.sent \|\| p.local \|\| p.ok) continue;
		@@ -378,9 +397,10 @@ void client::store_stripe(uint64_t group_id, uint32_t stripe_index,
		if (resp_len > 0) {
		std::vector<uint8_t> discard(resp_len);
		size_t total = 0;
		int body_att = 0;
		while (total < resp_len && body_att++ < 150) {
		if (!wait_.waitRead(conn, 100)) continue;
		while (total < resp_len) {
		int body_ms = ms_left(stripe_deadline);
		if (body_ms <= 0) break;
		if (!wait_.waitRead(conn, body_ms)) continue;
		conn.pumpNetwork(MSG_DONTWAIT);
		total += conn.recvStreamData(p.stream,
		discard.data() + total, resp_len - total);
		@@ -424,7 +444,7 @@ void client::store_stripe(uint64_t group_id, uint32_t stripe_index,
		if (wait_socks.empty()) break;

		std::vector<bool> ready;
		wait_.waitReadMulti(wait_socks, ready, 50);
		wait_.waitReadMulti(wait_socks, ready, std::min(poll_ms, 50));

		// Pump only sockets that are actually ready
		for (size_t wi = 0; wi < wait_socks.size(); ++wi) {
		@@ -479,6 +499,8 @@ bool client::retrieve_stripe(uint64_t group_id, uint32_t stripe_index,
		std::vector<uint8_t>& out) {
		size_t n = k_ + m_;

		auto retr_deadline = std::chrono::steady_clock::now() + std::chrono::seconds(15);

		// Pipelined fetch: send requests to all nodes in parallel,
		// then collect responses using waitReadMulti.
		struct Pending {
		@@ -514,8 +536,8 @@ bool client::retrieve_stripe(uint64_t group_id, uint32_t stripe_index,
		uint64_t stream = conn.openStream(true);
		pending[i].stream = stream;

		for (int sw = 0; sw < 150; ++sw) {
		if (wait_.waitWrite(conn, 100)) {
		while (int poll_ms = ms_left(retr_deadline)) {
		if (wait_.waitWrite(conn, poll_ms)) {
		conn.sendStreamData(stream, msg, true);
		conn.pumpNetwork(MSG_DONTWAIT);
		pending[i].sent = true;
		@@ -546,7 +568,9 @@ bool client::retrieve_stripe(uint64_t group_id, uint32_t stripe_index,
		size_t remaining = remote_sent;
		size_t total_ok = local_ok_count;

		for (int attempt = 0; attempt < 300 && remaining > 0 && total_ok < k_; ++attempt) {
		for (;;) {
		int poll_ms = ms_left(retr_deadline);
		if (poll_ms <= 0 \|\| remaining == 0 \|\| total_ok >= k_) break;
		// Check buffered data
		for (auto& p : pending) {
		if (!p.sent \|\| p.local \|\| p.ok) continue;
		@@ -567,9 +591,10 @@ bool client::retrieve_stripe(uint64_t group_id, uint32_t stripe_index,
		if (resp_type == msg_type::BLOCK_DATA && resp_len >= 12) {
		std::vector<uint8_t> body(resp_len);
		size_t total_read = 0;
		int body_att = 0;
		while (total_read < resp_len && body_att++ < 150) {
		if (!wait_.waitRead(conn, 100)) continue;
		while (total_read < resp_len) {
		int body_ms = ms_left(retr_deadline);
		if (body_ms <= 0) break;
		if (!wait_.waitRead(conn, body_ms)) continue;
		conn.pumpNetwork(MSG_DONTWAIT);
		total_read += conn.recvStreamData(p.stream,
		body.data() + total_read, resp_len - total_read);
		@@ -590,9 +615,10 @@ bool client::retrieve_stripe(uint64_t group_id, uint32_t stripe_index,
		// Drain non-BLOCK_DATA response body
		std::vector<uint8_t> discard(resp_len);
		size_t total_read = 0;
		int body_att = 0;
		while (total_read < resp_len && body_att++ < 150) {
		if (!wait_.waitRead(conn, 100)) continue;
		while (total_read < resp_len) {
		int body_ms = ms_left(retr_deadline);
		if (body_ms <= 0) break;
		if (!wait_.waitRead(conn, body_ms)) continue;
		conn.pumpNetwork(MSG_DONTWAIT);
		total_read += conn.recvStreamData(p.stream,
		discard.data() + total_read, resp_len - total_read);
		@@ -623,7 +649,7 @@ bool client::retrieve_stripe(uint64_t group_id, uint32_t stripe_index,
		if (wait_socks.empty()) break;

		std::vector<bool> ready;
		wait_.waitReadMulti(wait_socks, ready, 50);
		wait_.waitReadMulti(wait_socks, ready, std::min(poll_ms, 50));

		for (size_t wi = 0; wi < wait_socks.size(); ++wi) {
		if (!ready[wi]) continue;