fix files

git-svn-id: svn://db.shs.com.ru/pip@11 12ceb7fc-bf1f-11e4-8940-5bc7170c53b5

src/io/pibasetransfer.cpp

@@ -0,0 +1,361 @@
+#include "pibasetransfer.h"
+
+const uint PIBaseTransfer::signature = 0x54444950;
+
+PIBaseTransfer::PIBaseTransfer(): crc(standardCRC_16()) {
+	header.sig = signature;
+	header.session_id = 0;
+	packet_header_size = sizeof(PacketHeader) + customHeader().size();
+	part_header_size = sizeof(Part) + sizeof(int);
+	is_sending = is_receiving = false;
+	break_ = true;
+	bytes_all = bytes_cur = 0;
+	timeout_ = 1.;
+	setPacketSize(4096);
+	srand(PISystemTime::current().toMilliseconds());
+}
+
+
+PIBaseTransfer::~PIBaseTransfer() {
+	break_ = true;
+	session.clear();
+	replies.clear();
+}
+
+
+void PIBaseTransfer::stopSend() {
+	if (!is_sending) return;
+	break_ = true;
+}
+
+
+void PIBaseTransfer::stopReceive() {
+	if (!is_receiving) return;
+	break_ = true;
+	finish_receive(false);
+}
+
+
+void PIBaseTransfer::received(PIByteArray& data) {
+	packet_header_size = sizeof(PacketHeader) + customHeader().size();
+// 	piCoutObj << "r" << data.size() << sizeof(PacketHeader);
+	if (data.size() < sizeof(PacketHeader)) return;
+	PacketHeader h;
+	data >> h;
+	PacketType pt = (PacketType)h.type;
+// 	piCoutObj << "receive" << h.session_id << h.type << h.id;
+	switch (pt) {
+	case pt_Unknown:
+	break;
+	case pt_Data:
+		if (h.session_id != header.session_id || !is_receiving) {
+			sendBreak(h.session_id);
+			return;
+		} else {
+			uint rcrc = h.crc;
+			uint ccrc = crc.calculate(data.data(), data.size_s());
+			if (rcrc != ccrc) {
+				header.id = h.id;
+				sendReply(pt_ReplyInvalid);
+				piCoutObj << "invalid CRC";
+			} else {
+				processData(h.id, data);
+			}
+		}
+	break;
+	case pt_ReplySuccess:
+	case pt_ReplyInvalid:
+		if (h.session_id != header.session_id) return;
+		if (is_sending) {
+			if (h.id >= 0 && h.id < replies.size())
+				replies[h.id] = pt;
+		}
+		if (is_receiving && h.id == 0) {
+			if (checkSession() == 0 && pt == pt_ReplySuccess) finish_receive(true);
+		}
+	break;
+	case pt_Break:
+		break_ = true;
+// 		piCoutObj << "BREAK";
+		if (is_receiving) {
+			stopReceive();
+			return;
+		}
+		if (is_sending) {
+			stopSend();
+			return;
+		}
+	break;
+	case pt_Start:
+		if (is_sending) {
+			sendBreak(h.session_id);
+			return;
+		}
+		if (header.session_id != h.session_id && is_receiving) {
+			sendBreak(h.session_id);
+			return;
+		}
+		if (data.size() == sizeof(StartRequest)) {
+			StartRequest sr;
+			data >> sr;
+			bytes_cur = 0;
+			state_string = "start request";
+			bytes_all = sr.size;
+			header.session_id = h.session_id;
+			header.id = 0;
+			session.clear();
+			replies.clear();
+			session.resize(sr.packets);
+			replies.resize(sr.packets + 1);
+			replies.fill(pt_Unknown);
+			is_receiving = true;
+			break_ = false;
+			receiveStarted();
+			state_string = "receiving";
+			replies[0] = pt_ReplySuccess;
+			sendReply(pt_ReplySuccess);
+		}
+	break;
+	default:
+	break;
+	}
+}
+
+
+bool PIBaseTransfer::send_process() {
+	packet_header_size = sizeof(PacketHeader) + customHeader().size();
+	break_ = false;
+	is_sending = true;
+	sendStarted();
+	replies.resize(session.size() + 1);
+	replies.fill(pt_Unknown);
+	PIByteArray ba;
+	if (!getStartRequest()) return finish_send(false);
+	state_string = "sending";
+	for (int i = 0; i < session.size_s(); i++) {
+		ba = build_packet(i);
+		sendRequest(ba);
+		if (break_) return finish_send(false);
+	}
+	PITimeMeasurer tm;
+	int prev_chk = 0;
+	while (tm.elapsed_s() < timeout_) {
+		int chk = checkSession();
+		if (chk != prev_chk) tm.reset();
+		if (chk == 0) return finish_send(true);
+		if (chk > 0) {
+// 			piCoutObj << "recovery packet" << chk;
+			piMSleep(1);
+			ba = build_packet(chk - 1);
+			sendRequest(ba);
+		}
+//		if (chk == -1) return finish_send(false);
+		if (break_) return finish_send(false);
+		prev_chk = chk;
+		piMSleep(1);
+	}
+	return finish_send(false);
+}
+
+
+int PIBaseTransfer::checkSession() {
+	int miss = 0;
+	for (int i = 1; i < replies.size_s(); i++) {
+		if (replies[i] != pt_ReplySuccess) miss++;
+		if (replies[i] == pt_ReplyInvalid) return i;
+	}
+	for (int i = 1; i < replies.size_s(); i++) {
+		if (replies[i] !=  pt_ReplySuccess) return i;
+	}
+	if (miss > 0) {
+		piCoutObj << "missing" << miss << "packets";
+		return -miss;
+	} else return 0;
+}
+
+
+void PIBaseTransfer::buildSession(PIVector<Part> parts) {
+	state_string = "calculating files... ";
+	session.clear();
+	header.session_id = rand();
+	bytes_all = 0;
+	Part fi;
+	int fi_index, fi_prts;
+	PIVector<Part> lfi;
+	int min_size = packet_header_size + part_header_size;
+	int cur_size = min_size;
+	for (int i = 0; i < parts.size_s(); i++) {
+		state_string = "calculating files... " + PIString::fromNumber(i) + " of " + PIString::fromNumber(parts.size());
+		fi.id = parts[i].id;
+// 		piCout << fi.id << state_string;
+		bytes_all += parts[i].size;
+//		fi.size = fi.entry.size;
+		fi.start = 0;
+		int rest = parts[i].size - (packet_size - cur_size);
+// 		piCout << i << fi.size << rest << bytes_all;
+		if (rest <= 0) {
+			fi.size = parts[i].size;
+			lfi << fi;
+			cur_size += fi.size + part_header_size;
+		} else {
+			fi.size = parts[i].size - rest;
+			fi_index = 1;
+			fi_prts = 1 + 1 + piMaxi(1, rest / (packet_size - min_size));
+//			piCout << fi_prts;
+			lfi << fi;
+			session << lfi;
+			lfi.clear();
+			cur_size = min_size;
+			llong fs = fi.size;
+			for (int j = 1; j < fi_prts; j++) {
+				fi_index++;
+				fi.start = fs;
+				fi.size = piMin<ullong>(parts[i].size - fs, packet_size - min_size);
+				lfi << fi;
+				cur_size += fi.size + part_header_size;
+				if (fi_index != fi_prts) {
+					session << lfi;
+					lfi.clear();
+					cur_size = min_size;
+					fs += fi.size;
+				}
+			}
+		}
+		if (packet_size - cur_size < min_size) {
+			session << lfi;
+			lfi.clear();
+			cur_size = min_size;
+		}
+	}
+	if (cur_size > min_size) session << lfi;
+}
+
+
+void PIBaseTransfer::sendBreak(int session_id) {
+	uint psid = header.session_id;
+	header.session_id = session_id;
+	sendReply(pt_Break);
+	header.session_id = psid;
+}
+
+
+void PIBaseTransfer::sendReply(PacketType reply) {
+	header.type = reply;
+	PIByteArray ba;
+	ba << header;
+	sendRequest(ba);
+}
+
+
+bool PIBaseTransfer::getStartRequest() {
+	PITimeMeasurer tm;
+	header.type = pt_Start;
+	header.id = 0;
+	PIByteArray ba;
+	StartRequest st;
+	st.packets = (uint)session.size();
+	st.size = bytes_all;
+	ba << header;
+	ba << st;
+	state_string = "send request";
+	for (int i = 0; i < 5; i++) {
+		tm.reset();
+		sendRequest(ba);
+		while (tm.elapsed_s() < timeout_) {
+			if (break_) return false;
+			//piCoutObj << send_replyes[0];
+			if (replies[0] == pt_ReplySuccess) {
+				state_string = "send permited!";
+				return true;
+			}
+			piMSleep(10);
+		}
+	}
+	return false;
+}
+
+
+void PIBaseTransfer::processData(int id, PIByteArray & data) {
+// 	piCoutObj << "received packet" << id << ", size" << data.size();
+	if (id < 1 || id > replies.size_s()) return;
+	if (!session[id - 1].isEmpty()) {
+		header.id = id;
+		replies[id] = pt_ReplySuccess;
+		sendReply(pt_ReplySuccess);
+		if (checkSession() == 0) state_string = "receive ok";
+		return;
+	}
+	Part fi;
+	PIByteArray ba, pheader;
+	pheader.resize(packet_header_size - sizeof(PacketHeader));
+	if (!pheader.isEmpty()) {
+		memcpy(pheader.data(), data.data(), pheader.size());
+		data.remove(0, pheader.size_s());
+	}
+	while (!data.isEmpty()) {
+		ba.clear();
+		data >> fi;
+		/*if (fi.size > 0) */data >> ba;
+// 		fi.fsize = ba.size();
+		bytes_cur += fi.size;
+// 		piCoutObj << "recv" << fi;
+		session[id - 1] << fi;
+		state_string = "receiving...";
+		receivePart(fi, ba, pheader);
+	}
+	header.id = id;
+	replies[id] = pt_ReplySuccess;
+	if (checkSession() == 0) state_string = "receive ok";
+	sendReply(pt_ReplySuccess);
+}
+
+
+PIByteArray PIBaseTransfer::build_packet(int id) {
+	PIByteArray ret;
+	PIByteArray ba;
+	header.id = id + 1;
+	header.type = pt_Data;
+// 	piCoutObj << "Packet" << header.id;
+// 	piCoutObj << "session id" << header.session_id;
+	//ret << header;
+	ret.append(customHeader());
+	for (int i = 0; i < session[id].size_s(); i++) {
+		Part fi = session[id][i];
+		ret << fi;
+// 		piCout << "biuld" << fi; 
+		ba = buildPacket(fi);
+		bytes_cur += ba.size();
+		if (ba.size() != fi.size) piCoutObj << "***error while build packet, wrong part size";
+		ret << ba;
+	}
+	header.crc = crc.calculate(ret);
+	PIByteArray hdr; hdr << header;
+	ret.insert(0, hdr);
+	return ret;
+}
+
+
+bool PIBaseTransfer::finish_send(bool ok) {
+	if (ok) state_string = "send done";
+	else state_string = "send failed";
+// 	piCoutObj << state_string << PIString::readableSize(bytes_all);
+	is_sending = false;
+	header.id = 0;
+	if (!ok) sendBreak(header.session_id);
+	else sendReply(pt_ReplySuccess);
+	sendFinished(ok);
+	bytes_all = bytes_cur = 0;
+	return ok;
+}
+
+
+void PIBaseTransfer::finish_receive(bool ok) {
+	if (ok) state_string = "receive done";
+	else state_string = "receive failed";
+// 	piCoutObj << state_string << PIString::readableSize(bytes_all);
+	is_receiving = false;
+	if (!ok) sendBreak(header.session_id);
+	receiveFinished(ok);
+	bytes_all = bytes_cur = 0;
+}
+

src/io/pibasetransfer.h

@@ -0,0 +1,114 @@
+#ifndef PIBASETRANSFER_H
+#define PIBASETRANSFER_H
+
+#include "picrc.h"
+#include "pitimer.h"
+
+class PIBaseTransfer : public PIObject
+{
+	PIOBJECT(PIBaseTransfer)
+public:
+	PIBaseTransfer();
+	~PIBaseTransfer();
+
+# pragma pack(push,1)
+	struct PacketHeader {
+		uint sig;
+		int type; // PacketType
+		int session_id;
+		uint id;
+		uint crc;
+		bool check_sig() {return (sig == signature);}
+	};
+	
+	struct Part {
+		Part(uint id_ = 0, ullong size_ = 0, ullong start_ = 0) : id(id_), size(size_), start(start_) {}
+		uint id;
+		ullong size;
+		ullong start;
+	};
+# pragma pack(pop)
+
+	void stopSend();
+	void stopReceive();
+
+	bool isSending() const {return is_sending;}
+	bool isReceiving() const {return is_receiving;}
+
+	void setPacketSize(int size) {packet_size = size;}
+	int packetSize() const {return packet_size;}
+
+	void setTimeout(double sec) {timeout_ = sec;}
+	double timeout() const {return timeout_;}
+
+	const PIString & stateString() const {return state_string;}
+	llong bytesAll() const {return bytes_all;}
+	llong bytesCur() const {return bytes_cur;}
+	const PIString * stateString_ptr() const {return &state_string;}
+	const llong * bytesAll_ptr() const {return &bytes_all;}
+	const llong * bytesCur_ptr() const {return &bytes_cur;}
+
+	EVENT(receiveStarted)
+	EVENT1(receiveFinished, bool, ok)
+	EVENT(sendStarted)
+	EVENT1(sendFinished, bool, ok)
+	EVENT1(sendRequest, PIByteArray &, data)
+	EVENT_HANDLER1(void, received, PIByteArray &, data);
+
+protected:
+	uint packet_header_size, part_header_size;
+	bool break_, is_sending, is_receiving;
+	PIString state_string;
+	llong bytes_all, bytes_cur;
+	
+	void buildSession(PIVector<Part> parts);
+	virtual PIByteArray buildPacket(Part fi) = 0;
+    virtual void receivePart(Part fi, PIByteArray ba, PIByteArray pheader) = 0;
+	virtual PIByteArray customHeader() {return PIByteArray();}
+	bool send_process();
+	
+private:
+
+	enum PacketType {pt_Unknown, pt_Data, pt_ReplySuccess, pt_ReplyInvalid, pt_Break, pt_Start};
+
+# pragma pack(push,1)
+	struct StartRequest {
+		uint packets;
+		ullong size;
+	};
+# pragma pack(pop)
+	
+	friend PIByteArray & operator >>(PIByteArray & s, PIBaseTransfer::StartRequest & v);
+	friend PIByteArray & operator <<(PIByteArray & s, const PIBaseTransfer::StartRequest & v);
+
+	static const uint signature;
+	int packet_size;
+	double timeout_;
+	PIVector<PIVector<Part> > session;
+	PIVector<PacketType> replies;
+	PacketHeader header;
+	CRC_16 crc;
+
+	void processData(int id, PIByteArray &data);
+	PIByteArray build_packet(int id);
+	int checkSession();
+	void sendBreak(int session_id);
+	void sendReply(PacketType reply);
+	bool getStartRequest();
+	bool finish_send(bool ok);
+	void finish_receive(bool ok);
+};
+
+inline PIByteArray & operator <<(PIByteArray & s, const PIBaseTransfer::PacketHeader & v) {s << v.sig << v.type << v.session_id << v.id << v.crc; return s;}
+inline PIByteArray & operator >>(PIByteArray & s, PIBaseTransfer::PacketHeader & v) {s >> v.sig >> v.type >> v.session_id >> v.id >> v.crc; return s;}
+
+inline PIByteArray & operator <<(PIByteArray & s, const PIBaseTransfer::Part & v) {s << v.id << v.size << v.start; return s;}
+inline PIByteArray & operator >>(PIByteArray & s, PIBaseTransfer::Part & v) {s >> v.id >> v.size >> v.start; return s;}
+
+inline PIByteArray & operator <<(PIByteArray & s, const PIBaseTransfer::StartRequest & v) {s << v.packets << v.size; return s;}
+inline PIByteArray & operator >>(PIByteArray & s, PIBaseTransfer::StartRequest & v) {s >> v.packets >> v.size; return s;}
+
+inline PICout operator <<(PICout s, const PIBaseTransfer::Part & v) {s.setControl(0, true); s << "Part(\"" << v.id  << "\", " << PIString::readableSize(v.start) << " b | " << PIString::readableSize(v.size) << " b)"; s.restoreControl(); return s;}
+
+
+#endif // PIBASETRANSFER_H