From e9c43f1f8a391997d5a8041fdd3bd86b43bdf767 Mon Sep 17 00:00:00 2001
From: Frederic Robra
Date: Tue, 27 Aug 2019 17:00:40 +0200
Subject: optimized behaviour with failures

---
 src/kernel/dnbd3.h |  34 +++------------
 src/kernel/mq.c    |  34 +++++----------
 src/kernel/mq.h    |   4 +-
 src/kernel/net.c   | 123 ++++++++++++++++++++++++++++++++++-------------------
 4 files changed, 99 insertions(+), 96 deletions(-)

diff --git a/src/kernel/dnbd3.h b/src/kernel/dnbd3.h
index 4a2302d..01e8c15 100644
--- a/src/kernel/dnbd3.h
+++ b/src/kernel/dnbd3.h
@@ -74,25 +74,7 @@ struct dnbd3_server {
 	uint64_t rtts[4];
 	uint64_t avg_rtt;
 	uint16_t protocol_version;
-	uint16_t failures; // TODO failures runterzaehlen wenn wieder verbindung moeglich, seltener messen bei hohem failure count
-
-	/*
-	 * TODO would this help?
-	 * uint8_t failures[4]
-	 * uint8_t avg_failures;
-	 *
-	 * in timer, calculate each x iterations average over failure array
-	 * then move elements in array to left
-	 * for (i = 3; i > 0; i--) {
-	 *     server->failures[i] = server->failures[i - 1]
-	 * }
-	 * server->failures[0] = 0;
-	 * failures always go to failures[0] / *failures
-	 *
-	 * - is this to much effort?
-	 * - what are the benefits?
-	 * - could increase the rtt e.g. (rtt = rtt * failures)
-	 */
+	uint16_t failures;
 };
 
 /**
@@ -145,7 +127,7 @@ struct dnbd3_sock {
  * @update_available: 'true' if the rid has changes
  * @use_server_provided_alts: 'true' if the alt_servers array is upated by the
  *                             alternatives provided by the server
- * @rid: the revision ID? TODO
+ * @rid: the revision ID
  * @reported_size: the size of the image
  * @panic_worker: worker to handle panics and to connect if all connections are
  *                down
@@ -195,18 +177,16 @@ struct dnbd3_device {
  * @requed: 'true' if the command is requed
  */
 struct dnbd3_cmd {
-	//TODO do we want the socket here (index)?
 	struct dnbd3_device *dnbd3;
 	struct mutex lock;
 	uint32_t cookie;
-	blk_status_t status;
+	uint32_t index;
 	bool requed;
 };
 
 
-#define dnbd3_avg_rtt(server) \
-	(( (server)->rtts[0] + (server)->rtts[1] \
-			+ (server)->rtts[2] + (server)->rtts[3] ) / 4 )
+
+
 
 #define dnbd3_set_rtt_unreachable(server) \
 	(server)->rtts[0] = (server)->rtts[1] = (server)->rtts[2] \
@@ -216,8 +196,8 @@ struct dnbd3_cmd {
 
 #define dnbd3_set_rtt_unknown(server) \
 	(server)->rtts[0] = (server)->rtts[1] = (server)->rtts[2] \
-			= (server)->rtts[3] = (server)->avg_rtt = \
-			RTT_UNKNOWN;
+			= (server)->rtts[3] = 0;\
+	(server)->avg_rtt = RTT_UNKNOWN;
 
 
 /**
diff --git a/src/kernel/mq.c b/src/kernel/mq.c
index 946e395..b1c1e99 100644
--- a/src/kernel/mq.c
+++ b/src/kernel/mq.c
@@ -32,21 +32,24 @@
  */
 static void dnbd3_busy_iter(struct request *req, void *priv, bool arg2)
 {
-	bool *is_busy = (bool *) priv;
-	*is_busy = true;
+	struct dnbd3_cmd *cmd = blk_mq_rq_to_pdu(req);
+	unsigned long *busy = (unsigned long *) priv;
+	set_bit(cmd->index, busy);
 }
 
 
 /**
  * dnbd3_is_mq_busy - check if mq is busy
  * @dev: the device
+ *
+ * sets bit to 1, where socket is busy
  */
-bool dnbd3_is_mq_busy(struct dnbd3_device *dev)
+unsigned long dnbd3_is_mq_busy(struct dnbd3_device *dev)
 {
 	struct blk_mq_tag_set *set = &dev->tag_set;
-	bool is_busy = false;
+	unsigned long busy = 0;
 
-	blk_mq_tagset_busy_iter(set, dnbd3_busy_iter, &is_busy);
+	blk_mq_tagset_busy_iter(set, dnbd3_busy_iter, &busy);
 
 	/*
 	 * just for demonstration
@@ -64,7 +67,7 @@ bool dnbd3_is_mq_busy(struct dnbd3_device *dev)
 	}
 	*/
 
-	return is_busy;
+	return busy;
 }
 
 
@@ -92,20 +95,6 @@ void dnbd3_end_cmd(struct dnbd3_cmd *cmd, blk_status_t error)
 	blk_mq_end_request(req, error);
 }
 
-/**
- * dnbd3_is_any_sock_alive - check if any socket is alive
- * @cmd: the command
- */
-static bool dnbd3_is_any_sock_alive(struct dnbd3_cmd *cmd) {
-	int i;
-	for (i = 0; i < NUMBER_CONNECTIONS; i++) {
-		if (dnbd3_is_sock_alive(cmd->dnbd3->socks[i]) &&
-			!cmd->dnbd3->socks->panic) {
-			return true;
-		}
-	}
-	return false;
-}
 
 /**
  * dnbd3_handle_cmd - handles a mq command
@@ -140,8 +129,6 @@ static int dnbd3_handle_cmd(struct dnbd3_cmd *cmd, int index)
 		}
 	}
 
-	cmd->status = BLK_STS_OK;
-
 	mutex_lock(&sock->tx_lock);
 	if (unlikely(!sock->sock)) {
 		mutex_unlock(&sock->tx_lock);
@@ -181,6 +168,7 @@ static blk_status_t dnbd3_queue_rq(struct blk_mq_hw_ctx *hctx,
 	mutex_lock(&cmd->lock);
 
 	cmd->requed = false;
+	cmd->index = hctx->queue_num;
 
 	ret = dnbd3_handle_cmd(cmd, hctx->queue_num);
 	if (ret < 0) {
@@ -234,8 +222,6 @@ static enum blk_eh_timer_return dnbd3_xmit_timeout(struct request *req,
 	}
 
 	error_dev(dev, "connection timed out");
-	cmd->status = BLK_STS_IOERR;
-//	blk_mq_complete_request(req);
 	dnbd3_end_cmd(cmd, BLK_STS_TIMEOUT);
 	mutex_unlock(&cmd->lock);
 	return BLK_EH_DONE;
diff --git a/src/kernel/mq.h b/src/kernel/mq.h
index 74f4f68..821fd91 100644
--- a/src/kernel/mq.h
+++ b/src/kernel/mq.h
@@ -26,8 +26,10 @@
 /**
  * dnbd3_is_mq_busy - check if mq is busy
  * @dev: the device
+ *
+ * sets bit to 1, where socket is busy
  */
-bool dnbd3_is_mq_busy(struct dnbd3_device *dev);
+unsigned long dnbd3_is_mq_busy(struct dnbd3_device *dev);
 
 /**
  * dnbd3_requeue_cmd - requeue a command once
diff --git a/src/kernel/net.c b/src/kernel/net.c
index 79c860f..1207d0a 100644
--- a/src/kernel/net.c
+++ b/src/kernel/net.c
@@ -73,6 +73,22 @@ static int dnbd3_socket_connect(struct dnbd3_sock *sock,
 		struct dnbd3_server * server);
 static int dnbd3_socket_disconnect(struct dnbd3_sock *sock);
 
+static uint64_t dnbd3_average_rtt(struct dnbd3_server *server)
+{
+	int i, j = 0;
+	uint64_t avg = 0;
+	for (i = 0; i < 4; i++) {
+		avg += server->rtts[i];
+		j += server->rtts[i] == 0 ? 0 : 1;
+	}
+	if (avg == 0) {
+		return RTT_UNKNOWN;
+	} else {
+		avg = avg / j;
+		avg += server->failures * avg / 10;
+		return avg;
+	}
+}
 
 /*
  * Methods for request and receive commands
@@ -267,6 +283,42 @@ static int dnbd3_receive_cmd(struct dnbd3_sock *sock, dnbd3_reply_t *reply)
 	return result;
 }
 
+static int dnbd3_clear_socket(struct dnbd3_sock *sock, dnbd3_reply_t *reply,
+		int remaining)
+{
+	int result = 0;
+	char *buf;
+	struct kvec iov;
+	struct msghdr msg;
+	dnbd3_init_msghdr(msg);
+	warn_sock(sock, "clearing socket %d bytes", remaining);
+	buf = kmalloc(RTT_BLOCK_SIZE, GFP_KERNEL);
+	if (!buf) {
+		error_sock(sock, "kmalloc failed");
+		return -EIO;
+	}
+	/* hold the tx_lock so no new requests are send */
+	mutex_lock(&sock->tx_lock);
+	iov.iov_base = buf;
+	iov.iov_len = RTT_BLOCK_SIZE;
+	while (remaining > 0) {
+		result = kernel_recvmsg(sock->sock, &msg, &iov, 1, iov.iov_len,
+				msg.msg_flags);
+		if (result <= 0) {
+			goto error;
+		}
+		remaining -= result;
+	}
+
+	debug_sock(sock, "cleared socket");
+error:
+	mutex_unlock(&sock->tx_lock);
+	if (buf) {
+		kfree(buf);
+	}
+	return result;
+}
+
 /**
  * dnbd3_receive_cmd_get_block_mq - receive a block for mq
  * @sock: the socket where the request is received
@@ -312,8 +364,8 @@ static int dnbd3_receive_cmd_get_block_mq(struct dnbd3_sock *sock,
 			debug_sock(sock, "requeue request");
 			dnbd3_requeue_cmd(blk_mq_rq_to_pdu(req));
 		}
-//		return -EIO;
-		goto clear_socket;
+		dnbd3_clear_socket(sock, reply, remaining);
+		return -EIO;
 	}
 	cmd = blk_mq_rq_to_pdu(req);
 
@@ -322,8 +374,8 @@ static int dnbd3_receive_cmd_get_block_mq(struct dnbd3_sock *sock,
 		error_sock(sock, "double reply on req %p, cookie %u, handle cookie %u",
 				req, cmd->cookie, cookie);
 		mutex_unlock(&cmd->lock);
-//		return -EIO;
-		goto clear_socket;
+		dnbd3_clear_socket(sock, reply, remaining);
+		return -EIO;
 	}
 
 	rq_for_each_segment(bvec_inst, req, iter) {
@@ -343,7 +395,8 @@ static int dnbd3_receive_cmd_get_block_mq(struct dnbd3_sock *sock,
 			dnbd3_requeue_cmd(cmd);
 			mutex_unlock(&cmd->lock);
 			if (result >= 0) {
-				goto clear_socket;
+				dnbd3_clear_socket(sock, reply, remaining);
+				return -EIO;
 			} else {
 				return result;
 			}
@@ -355,35 +408,10 @@ static int dnbd3_receive_cmd_get_block_mq(struct dnbd3_sock *sock,
 	mutex_unlock(&cmd->lock);
 	dnbd3_end_cmd(cmd, 0);
 	return result;
-clear_socket:
-	warn_sock(sock, "caught an error while receiving block, clearing buffer");
-	char *buf = kmalloc(RTT_BLOCK_SIZE, GFP_KERNEL);
-	if (!buf) {
-		error_sock(sock, "kmalloc failed");
-		return -EIO;
-	}
-
-	iov.iov_base = buf;
-	iov.iov_len = RTT_BLOCK_SIZE;
-	while (remaining > 0) {
-		result = kernel_recvmsg(sock->sock, &msg, &iov, 1, iov.iov_len,
-				msg.msg_flags);
-		if (result <= 0) {
-			goto error;
-		}
-		remaining -= result;
-	}
-
-	debug_sock(sock, "cleared buffer %d bytes, reply size is %d", result,
-			reply->size);
-error:
-	if (buf) {
-		kfree(buf);
-	}
-	return -EIO;
 }
 
 
+
 /**
  * dnbd3_receive_cmd_get_block_test - receive a test block
  * @sock: the socket where the request is received
@@ -620,26 +648,29 @@ static int dnbd3_receive_cmd_select_image(struct dnbd3_sock *sock,
 static void dnbd3_timer(struct timer_list *arg)
 {
 	struct dnbd3_device *dev;
+	unsigned long busy;
 	int i;
 	dev = container_of(arg, struct dnbd3_device, timer);
 	queue_work(dnbd3_wq, &dev->panic_worker);
+	busy = dnbd3_is_mq_busy(dev);
 
-	if (!dnbd3_is_mq_busy(dev)) {
-		if (dev->timer_count % TIMER_INTERVAL_KEEPALIVE_PACKET == 0) {
-			for (i = 0; i < NUMBER_CONNECTIONS; i++) {
-				if (dnbd3_is_sock_alive(dev->socks[i])) {
-					queue_work(dnbd3_wq, &dev->socks[i].keepalive_worker);
-				}
+	if (dev->timer_count % TIMER_INTERVAL_KEEPALIVE_PACKET == 0) {
+		for (i = 0; i < NUMBER_CONNECTIONS; i++) {
+			if (!test_bit(i, &busy) && dnbd3_is_sock_alive(dev->socks[i])) {
+				queue_work(dnbd3_wq, &dev->socks[i].keepalive_worker);
 			}
 		}
+	}
+
+	if (!busy) {
 		/* start after 4 seconds */
 		if (dev->timer_count % TIMER_INTERVAL_PROBE_NORMAL == 4) {
 			queue_work(dnbd3_wq, &dev->discovery_worker);
 		}
 
 
-		dev->timer_count++;
 	}
+	dev->timer_count++;
 	dev->timer.expires = jiffies + HZ;
 	add_timer(&dev->timer);
 }
@@ -739,8 +770,10 @@ static void dnbd3_keepalive_worker(struct work_struct *work)
 {
 	struct dnbd3_sock *sock;
 	sock = container_of(work, struct dnbd3_sock, keepalive_worker);
-	debug_sock(sock, "starting keepalive worker");
-	dnbd3_send_request_cmd(sock, CMD_KEEPALIVE);
+	if (sock->server != NULL && !sock->panic) {
+		debug_sock(sock, "starting keepalive worker");
+		dnbd3_send_request_cmd(sock, CMD_KEEPALIVE);
+	}
 }
 
 /**
@@ -754,9 +787,9 @@ static int dnbd3_compare_servers(const void *lhs, const void *rhs) {
 	struct dnbd3_server *rhs_server = *((struct dnbd3_server **) rhs);
 
 	l = lhs_server->host.type != 0 ? lhs_server->avg_rtt
-			: RTT_UNREACHABLE + 1;
+			: RTT_UNREACHABLE;
 	r = rhs_server->host.type != 0 ? rhs_server->avg_rtt
-			: RTT_UNREACHABLE + 1;
+			: RTT_UNREACHABLE;
 	if (l < r) {
 		return -1;
 	} else if (l > r) {
@@ -998,7 +1031,7 @@ static int dnbd3_meassure_rtt(struct dnbd3_device *dev,
 
 error:
 	sock.server->rtts[dev->discovery_count % 4] = rtt;
-	sock.server->avg_rtt = dnbd3_avg_rtt(sock.server);
+	sock.server->avg_rtt = dnbd3_average_rtt(sock.server);
 	if (result <= 0) {
 		server->failures++;
 	}
@@ -1279,10 +1312,12 @@ static int dnbd3_socket_connect(struct dnbd3_sock *sock,
 
 	INIT_WORK(&sock->keepalive_worker, dnbd3_keepalive_worker);
 
-	/* TODO not on every connect? request alternative servers receiver will handle this */
+	/* request alternative servers receiver will handle this */
 	if (dnbd3_send_request_cmd(sock, CMD_GET_SERVERS) <= 0) {
 		error_sock(sock, "failed to get servers in discovery");
 	}
+	/* failure count is divided on each successful connect */
+	server->failures = server->failures / 2;
 
 	return 0;
 error:
-- 
cgit v1.2.3-55-g7522