PIDataTransferansfers fixnsfer

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

src/io/pibasetransfer.cpp

@@ -10,7 +10,7 @@ PIBaseTransfer::PIBaseTransfer(): crc(standardCRC_16()) {
 	is_sending = is_receiving = false;
 	break_ = true;
 	bytes_all = bytes_cur = 0;
-	replies_cnt = 0;
+	replies_cnt = send_queue = 0;
 	timeout_ = 10.;
 	packets_count = 32;
 	setPacketSize(4096);
@@ -67,6 +67,11 @@ void PIBaseTransfer::received(PIByteArray data) {
 	case pt_ReplySuccess:
 	case pt_ReplyInvalid:
 		if (h.session_id != header.session_id) return;
+		if (pt == pt_ReplySuccess) {
+			send_queue--;
+			if (send_queue < 0) send_queue = 0;
+			send_tm.reset();
+		}
 		replies_cnt++;
 		if (is_sending) {
 			if (h.id >= 0 && h.id < replies.size())
@@ -113,7 +118,7 @@ void PIBaseTransfer::received(PIByteArray data) {
 			is_receiving = true;
 			break_ = false;
 			receiveStarted();
-			replies_cnt = 0;
+			replies_cnt = send_queue = 0;
 			state_string = "receiving";
 			replies[0] = pt_ReplySuccess;
 			sendReply(pt_ReplySuccess);
@@ -133,43 +138,52 @@ bool PIBaseTransfer::send_process() {
 	replies.fill(pt_Unknown);
 	PIByteArray ba;
 	if (!getStartRequest()) return finish_send(false);
-	replies_cnt = 0;
+	replies_cnt = send_queue = 0;
 	state_string = "sending";
 	PITimeMeasurer stm;
+	send_tm.reset();
 	int ltm = 0;
 	for (int i = 0; i < session.size_s(); i++) {
-		if (i - replies_cnt >= packets_count) {
+		if (send_queue >= packets_count) {
 			--i;
-			piMSleep(1);
+			piMSleep(10);
-			if (ltm > 10) return finish_send(false);
+			if (send_tm.elapsed_s() > timeout_) return finish_send(false);
-			if (stm.elapsed_s() > timeout_/10) {
+			if (stm.elapsed_s() > timeout_ / 10.)
-				replies_cnt = i;
+				send_queue = 0;
-				ltm++;
-			}
 			continue;
 		}
 		stm.reset();
 		ba = build_packet(i);
 		sendRequest(ba);
+		send_queue++;
 		if (break_) return finish_send(false);
 	}
 	PITimeMeasurer tm2, tm;
 	int prev_chk = 0;
+	send_queue = 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) {
+		if (chk != prev_chk) tm.reset();
-			if (tm2.elapsed_s() > timeout_/10) {
+		else if (tm.elapsed_m() < 100) {
-	// 			piCoutObj << "recovery packet" << chk;
 			piMSleep(1);
+			continue;
+		}
+		prev_chk = chk;
+		if (chk > 0) {
+			if (tm2.elapsed_s() > timeout_ / 10.) {
+				//piCoutObj << "recovery packet" << chk;
+				if (send_queue >= packets_count) {
+					piMSleep(10);
+					continue;
+				}
 				ba = build_packet(chk - 1);
 				sendRequest(ba);
+				send_queue++;
 			}
 		}
 //		if (chk == -1) return finish_send(false);
 		if (break_) return finish_send(false);
-		prev_chk = chk;
 		piMSleep(1);
 	}
 	return finish_send(false);

src/io/pibasetransfer.h

@@ -86,9 +86,10 @@ private:
 	double timeout_;
 	PIVector<PIVector<Part> > session;
 	PIVector<PacketType> replies;
+	PITimeMeasurer send_tm;
 	PacketHeader header;
 	CRC_16 crc;
-    int replies_cnt;
+    int replies_cnt, send_queue;
 
 	void processData(int id, PIByteArray &data);
 	PIByteArray build_packet(int id);

src/io/pipeer.cpp

@@ -424,7 +424,7 @@ bool PIPeer::dataRead(uchar * readed, int size) {
 		cnt++;
 		if (cnt > _PIPEER_MSG_TTL || from == dp->name) return true;
 		sba << type << from << to << cnt << pba;
-		//piCoutObj << "Translate data packet" << type << from << to << cnt << rec_size;
+		piCout << "translate packet" << from << "->" << to << ", ttl =" << cnt;
 		sendToNeighbour(dp, sba);
 	}
 	return true;