sort the connection plan, locks the sending while changing connection,

discovery now takes place even if mq requests are received

6 files changed, 135 insertions, 94 deletions

diff --git a/src/kernel/core.c b/src/kernel/core.c
index 8582b5c..6aa3b50 100644
--- a/src/kernel/core.c
+++ b/src/kernel/core.c
@@ -250,7 +250,7 @@ int dnbd3_add_device(struct dnbd3_device *dev, int minor)
 	for (i = 0; i < NUMBER_CONNECTIONS; i++) {
 		dev->socks[i].device = dev;
 		dev->socks[i].sock_nr = i;
-		mutex_init(&dev->socks[i]->tx_lock);
+		mutex_init(&dev->socks[i].tx_lock);
 	}
 
 	disk = alloc_disk(1);
@@ -422,7 +422,7 @@ static void dnbd3_dev_remove(struct dnbd3_device *dev)
 	}
 	mutex_destroy(&dev->device_lock);
 	for (i = 0; i < NUMBER_CONNECTIONS; i++) {
-		mutex_destroy(&dev->socks[i]->tx_lock);
+		mutex_destroy(&dev->socks[i].tx_lock);
 	}
 }
 
diff --git a/src/kernel/dnbd3.h b/src/kernel/dnbd3.h
index 01e8c15..79484a9 100644
--- a/src/kernel/dnbd3.h
+++ b/src/kernel/dnbd3.h
@@ -99,7 +99,7 @@ struct dnbd3_sock {
 	struct mutex tx_lock;
 	struct socket *sock;
 
-	bool panic;
+//	bool panic;
 	uint32_t cookie;
 
 	struct work_struct keepalive_worker;
diff --git a/src/kernel/mq.c b/src/kernel/mq.c
index 60673b0..5fa34d5 100644
--- a/src/kernel/mq.c
+++ b/src/kernel/mq.c
@@ -132,7 +132,7 @@ static int dnbd3_handle_cmd(struct dnbd3_cmd *cmd, int index)
 //			blk_rq_pos(req), blk_rq_bytes(req), index);
 
 
-	if (!(sock->server && sock->sock && !sock->panic)) {
+	if (!dnbd3_is_sock_alive(*sock)) {
 //		warn_dev(dev, "attempted send on invalid socket %d", index);
 //		msleep(SOCKET_TIMEOUT_CLIENT_DATA * 1000);
 
@@ -142,7 +142,7 @@ static int dnbd3_handle_cmd(struct dnbd3_cmd *cmd, int index)
 			ret = 0;
 			goto out;
 		} else {
-			error_dev(dev, "ending request, device not connected");
+			error_dev(dev, "ending request device not connected");
 			dnbd3_end_cmd(cmd, BLK_STS_IOERR);
 			ret = -EIO;
 			goto out;
@@ -165,7 +165,7 @@ static int dnbd3_handle_cmd(struct dnbd3_cmd *cmd, int index)
 
 	ret = dnbd3_send_request(sock, blk_mq_rq_from_pdu(cmd), cmd);
 	if (ret == -EAGAIN) {
-		dev_err_ratelimited(disk_to_dev(dev->disk), "request send failed, requeueing\n");
+		error_dev(dev, "request send failed, requeueing");
 		dnbd3_requeue_cmd(cmd);
 		ret = 0;
 	}
diff --git a/src/kernel/mq.h b/src/kernel/mq.h
index 94b09f9..c29e99a 100644
--- a/src/kernel/mq.h
+++ b/src/kernel/mq.h
@@ -23,6 +23,7 @@
 
 #include "dnbd3.h"
 
+
 /**
  * dnbd3_is_mq_busy - check if mq is busy
  * @dev: the device
diff --git a/src/kernel/net-txrx.c b/src/kernel/net-txrx.c
index 7226a3e..3d2cc7c 100644
--- a/src/kernel/net-txrx.c
+++ b/src/kernel/net-txrx.c
@@ -121,7 +121,7 @@ int dnbd3_send_request(struct dnbd3_sock *sock, struct request *req,
 
 	switch (req_op(req)) {
 	case REQ_OP_READ:
-		debug_sock(sock, "request operation read");
+		debug_sock(sock, "request operation read %p", req);
 		request.cmd = CMD_GET_BLOCK;
 		request.offset = blk_rq_pos(req) << 9; // * 512
 		request.size = blk_rq_bytes(req);
@@ -168,7 +168,6 @@ int dnbd3_send_request(struct dnbd3_sock *sock, struct request *req,
 		if (cmd) {
 			dnbd3_requeue_cmd(cmd);
 		}
-		sock->panic = true;
 		sock->server->failures++;
 		goto error;
 	}
@@ -275,8 +274,6 @@ static int dnbd3_clear_socket(struct dnbd3_sock *sock, dnbd3_reply_t *reply,
 		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) {
@@ -290,7 +287,6 @@ static int dnbd3_clear_socket(struct dnbd3_sock *sock, dnbd3_reply_t *reply,
 
 	debug_sock(sock, "cleared socket");
 error:
-	mutex_unlock(&sock->tx_lock);
 	if (buf) {
 		kfree(buf);
 	}
diff --git a/src/kernel/net.c b/src/kernel/net.c
index 67f9c74..f4665bf 100644
--- a/src/kernel/net.c
+++ b/src/kernel/net.c
@@ -35,8 +35,13 @@
 	sock_create_kern(&init_net, (af) == HOST_IP4 ? AF_INET : AF_INET6, \
 			type, proto, sock)
 
-#define dnbd3_is_sock_alive(s) ((s).sock && (s).server)
+#define dnbd3_sock_release(sock) \
+	do { \
+		sock_release((sock)->sock); \
+		sock->sock = NULL; \
+	}while (0)
 
+#define dnbd3_is_sock_alive(s) ((s).sock && (s).server)
 
 
 static int __dnbd3_socket_connect(struct dnbd3_sock *sock,
@@ -157,7 +162,7 @@ static void dnbd3_receive_worker(struct work_struct *work)
 error:
 		if (result == 0) {
 			info_sock(sock, "result is 0, socket seems to be down");
-			sock->panic = true;
+			dnbd3_sock_release(sock);
 			break;
 		} else if (result < 0) {
 			/* discovery takes care of to many failures */
@@ -180,11 +185,12 @@ static void dnbd3_keepalive_worker(struct work_struct *work)
 {
 	struct dnbd3_sock *sock;
 	sock = container_of(work, struct dnbd3_sock, keepalive_worker);
-	if (sock->server != NULL && !sock->panic) {
+	if (dnbd3_is_sock_alive(*sock)) {
 		debug_sock(sock, "starting keepalive worker");
-		mutex_lock(&sock->tx_lock);
-		dnbd3_send_request_cmd(sock, CMD_KEEPALIVE);
-		mutex_unlock(&sock->tx_lock);
+		if (mutex_trylock(&sock->tx_lock)) {
+			dnbd3_send_request_cmd(sock, CMD_KEEPALIVE);
+			mutex_unlock(&sock->tx_lock);
+		}
 	}
 }
 
@@ -199,15 +205,10 @@ 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;
+			: RTT_UNREACHABLE * 2;
 	r = rhs_server->host.type != 0 ? rhs_server->avg_rtt
-			: RTT_UNREACHABLE;
-	if (l < r) {
-		return -1;
-	} else if (l > r) {
-		return 1;
-	}
-	return 0;
+			: RTT_UNREACHABLE * 2;
+	return l - r;
 }
 
 /**
@@ -218,47 +219,103 @@ static int dnbd3_compare_servers(const void *lhs, const void *rhs) {
  */
 static struct dnbd3_server **dnbd3_sort_server(struct dnbd3_device *dev) {
 	int i;
-	struct dnbd3_server **sorted_servers = kmalloc(NUMBER_SERVERS * sizeof(struct dnbd3_device *), GFP_KERNEL);
+	struct dnbd3_server **sorted_servers = kmalloc(NUMBER_SERVERS *
+			sizeof(struct dnbd3_server *), GFP_KERNEL);
 	if (!sorted_servers) {
+		debug_dev(dev, "kmalloc failed");
 		return NULL;
 	}
 
 	for (i = 0; i < NUMBER_SERVERS; i++) {
 		sorted_servers[i] = &dev->alt_servers[i];
 	}
-	sort(sorted_servers, NUMBER_SERVERS, sizeof(struct dnbd3_device *),
+	sort(sorted_servers, NUMBER_SERVERS, sizeof(struct dnbd3_server *),
 			&dnbd3_compare_servers, NULL);
 
 	return sorted_servers;
 }
 
-static int dnbd3_panic_connect(struct dnbd3_device *dev)
+//static int dnbd3_panic_connect(struct dnbd3_device *dev)
+//{
+//	struct dnbd3_server *working = NULL;
+//	int i;
+//	debug_dev(dev, "panic connect");
+//	for (i = 0; i < NUMBER_CONNECTIONS; i++) {
+//		if (dnbd3_is_sock_alive(dev->socks[i])) {
+//			working = dev->socks[i].server;
+//			debug_server(dev, working, "found server for panic");
+//		}
+//	}
+//	if (working == NULL) {
+//		for (i = 0; i < NUMBER_SERVERS; i++) {
+//			if (!dnbd3_socket_connect(&dev->socks[0],
+//					&dev->alt_servers[i])) {
+//				working = &dev->alt_servers[i];
+//				debug_server(dev, working, "found server for panic");
+//			}
+//		}
+//	}
+//	if (working == NULL) {
+//		return -ENOENT;
+//	}
+//	for (i = 0; i < NUMBER_CONNECTIONS; i++) {
+//		if (dev->socks[i].server != working) {
+//			dnbd3_socket_connect(&dev->socks[i], working);
+//		}
+//	}
+//	return 0;
+//}
+
+/**
+ * dnbd3_compare_servers - comparator for the connecion plan
+ * @lhs: left hand sign
+ * @rhs: right hand sign
+ */
+static int dnbd3_compare_plan(const void *lhs, const void *rhs)
+{
+	uint64_t l, r;
+	struct dnbd3_server *lhs_server = *((struct dnbd3_server **) lhs);
+	struct dnbd3_server *rhs_server = *((struct dnbd3_server **) rhs);
+	uint8_t *l_addr = lhs_server->host.addr;
+	uint8_t *r_addr = rhs_server->host.addr;
+	l = l_addr[0] + l_addr[1] + l_addr[2] + l_addr[3] + l_addr[4] +
+			 l_addr[5] + l_addr[6] + l_addr[7] + l_addr[8] +
+			 l_addr[9] + l_addr[10] + l_addr[11] + l_addr[12] +
+			 l_addr[13] + l_addr[14] + l_addr[15] +
+			 lhs_server->host.port;
+	r = r_addr[0] + r_addr[1] + r_addr[2] + r_addr[3] + r_addr[4] +
+			 r_addr[5] + r_addr[6] + r_addr[7] + r_addr[8] +
+			 r_addr[9] + r_addr[10] + r_addr[11] + r_addr[12] +
+			 r_addr[13] + r_addr[14] + r_addr[15] +
+			 rhs_server->host.port;
+	return l - r;
+}
+
+static void dnbd3_lock_all_socks(struct dnbd3_device *dev)
 {
-	struct dnbd3_server *working = NULL;
 	int i;
-	debug_dev(dev, "panic connect");
 	for (i = 0; i < NUMBER_CONNECTIONS; i++) {
-		if (dnbd3_is_sock_alive(dev->socks[i])) {
-			working = dev->socks[i].server;
-		}
-	}
-	if (working == NULL) {
-		for (i = 0; i < NUMBER_SERVERS; i++) {
-			if (!dnbd3_socket_connect(&dev->socks[0],
-					&dev->alt_servers[i])) {
-				working = &dev->alt_servers[i];
-			}
-		}
+		mutex_lock(&dev->socks[i].tx_lock);
 	}
-	if (working == NULL) {
-		return -ENOENT;
+}
+
+static void dnbd3_unlock_all_socks(struct dnbd3_device *dev)
+{
+	int i;
+	for (i = 0; i < NUMBER_CONNECTIONS; i++) {
+		mutex_unlock(&dev->socks[i].tx_lock);
 	}
+}
+
+static void dnbd3_print_conenction_plan(struct dnbd3_device *dev,
+		struct dnbd3_server **plan)
+{
+	int i;
+	debug_dev(dev, "connection plan:");
 	for (i = 0; i < NUMBER_CONNECTIONS; i++) {
-		if (dev->socks[i].server != working) {
-			dnbd3_socket_connect(&dev->socks[i], working);
-		}
+		debug_server(dev, plan[i], "server %d with avg rtt %llu:",
+				i, plan[i]->avg_rtt);
 	}
-	return 0;
 }
 
 /**
@@ -270,7 +327,7 @@ static int dnbd3_panic_connect(struct dnbd3_device *dev)
  * 3. connect the plan
  */
 static int dnbd3_adjust_connections(struct dnbd3_device *dev) {
-	int i, j, fallback;
+	int i, j, fallback, alive;
 	struct dnbd3_server *plan[NUMBER_CONNECTIONS];
 	struct dnbd3_server **servers = dnbd3_sort_server(dev);
 
@@ -278,9 +335,6 @@ static int dnbd3_adjust_connections(struct dnbd3_device *dev) {
 		plan[0] = servers[0];
 		fallback = 0;
 		j = 1;
-		debug_dev(dev, "connection plan:");
-		debug_server(dev, plan[0], "server 0 with rtt %llu:",
-				plan[0]->avg_rtt);
 		for (i = 1; i < NUMBER_CONNECTIONS; i++) {
 			if (servers[j]->host.type != 0 &&
 					servers[j]->avg_rtt < RTT_UNKNOWN) {
@@ -296,25 +350,37 @@ static int dnbd3_adjust_connections(struct dnbd3_device *dev) {
 				plan[i] = plan[fallback];
 				fallback++;
 			}
-			debug_server(dev, plan[i], "server %d with rtt %llu:",
-					i, plan[i]->avg_rtt);
 		}
 		kfree(servers);
-		//TODO connection plan sortieren
 
+		sort(plan, NUMBER_CONNECTIONS, sizeof(struct dnbd3_server *),
+			&dnbd3_compare_plan, NULL);
+		dnbd3_print_conenction_plan(dev, plan);
+
+		dnbd3_lock_all_socks(dev);
+		alive = 0;
 		for (i = 0; i < NUMBER_CONNECTIONS; i++) {
-			if (plan[i] != dev->socks[i].server) {
+			if (plan[i] != dev->socks[i].server ||
+					!dnbd3_is_sock_alive(dev->socks[i])) {
+
 				if (dnbd3_is_sock_alive(dev->socks[i])) {
 					dnbd3_socket_disconnect(&dev->socks[i]);
 				}
-				j = dnbd3_socket_connect(&dev->socks[i], plan[i]);
-				if (j) {
-					return j;
+				if (!dnbd3_socket_connect(&dev->socks[i],
+						plan[i])) {
+					alive++;
 				}
+			} else {
+				alive++;
 			}
 		}
+		dnbd3_unlock_all_socks(dev);
+		if (alive == 0) {
+			return -EIO;
+		}
 		return 0;
 	} else { /* there is nothing to connect */
+		debug_dev(dev, "failed to adjust connections");
 		if (servers) {
 			kfree(servers);
 		}
@@ -340,13 +406,13 @@ static void dnbd3_panic_worker(struct work_struct *work)
 	int sock_alive = 0;
 	dev = container_of(work, struct dnbd3_device, panic_worker);
 	for (i = 0; i < NUMBER_CONNECTIONS; i++) {
-		if (dev->socks[i].panic
+		if (!dnbd3_is_sock_alive(dev->socks[i])
 				|| dev->socks[i].server->failures > 1000) {
 			panic = true;
 			dnbd3_set_rtt_unreachable(dev->socks[i].server);
-			dnbd3_socket_disconnect(&dev->socks[i]);
+//			dnbd3_socket_disconnect(&dev->socks[i]);
 
-		} else if (dnbd3_is_sock_alive(dev->socks[i])) {
+		} else {
 			sock_alive++;
 		}
 	}
@@ -358,10 +424,8 @@ static void dnbd3_panic_worker(struct work_struct *work)
 
 		mutex_lock(&dev->device_lock);
 		if (dnbd3_adjust_connections(dev)) {
-			if (dnbd3_panic_connect(dev)) {
-				error_dev(dev, "failed to connect to any server");
-				dev->connected = false;
-			}
+			error_dev(dev, "failed to connect to any server");
+			dev->connected = false;
 
 		}
 		mutex_unlock(&dev->device_lock);
@@ -565,10 +629,8 @@ static void dnbd3_discovery_worker(struct work_struct *work)
 
 	mutex_lock(&dev->device_lock);
 	if (dnbd3_adjust_connections(dev)) {
-		if (dnbd3_panic_connect(dev)) {
-			error_dev(dev, "failed to connect to any server");
-			dev->connected = false;
-		}
+		error_dev(dev, "failed to connect to any server");
+		dev->connected = false;
 
 	}
 	mutex_unlock(&dev->device_lock);
@@ -645,8 +707,7 @@ static int __dnbd3_socket_connect(struct dnbd3_sock *sock,
 	return 0;
 error:
 	if (sock->sock) {
-		sock_release(sock->sock);
-		sock->sock = NULL;
+		dnbd3_sock_release(sock);
 	}
 	return result;
 }
@@ -670,22 +731,16 @@ static int dnbd3_socket_connect(struct dnbd3_sock *sock,
 
 	debug_sock(sock, "socket connect");
 
-	mutex_init(&sock->tx_lock);
-	mutex_lock(&sock->tx_lock);
 	result = __dnbd3_socket_connect(sock, server);
 	if (result) {
 		error_sock(sock, "connection to socket failed");
-		mutex_unlock(&sock->tx_lock);
 		result = -EIO;
 		goto error;
 	}
 
-	sock->panic = false;
-
 	if (!sock->sock) {
 		error_sock(sock, "socket is not connected");
 		server->failures++;
-		mutex_unlock(&sock->tx_lock);
 		result = -EIO;
 		goto error;
 	}
@@ -694,7 +749,6 @@ static int dnbd3_socket_connect(struct dnbd3_sock *sock,
 	result = dnbd3_send_request_cmd(sock, CMD_SELECT_IMAGE);
 	if (result <= 0) {
 		error_sock(sock, "connection to image %s failed", dev->imgname);
-		mutex_unlock(&sock->tx_lock);
 		result = -EIO;
 		goto error;
 	}
@@ -703,7 +757,6 @@ static int dnbd3_socket_connect(struct dnbd3_sock *sock,
 	if (result <= 0) {
 		error_sock(sock, "receive cmd to image %s failed",
 				dev->imgname);
-		mutex_unlock(&sock->tx_lock);
 		result = -EIO;
 		goto error;
 	}
@@ -711,7 +764,6 @@ static int dnbd3_socket_connect(struct dnbd3_sock *sock,
 			|| reply.size < 4) {
 		error_sock(sock, "receive select image wrong header %s",
 				dev->imgname);
-		mutex_unlock(&sock->tx_lock);
 		result = -EIO;
 		goto error;
 	}
@@ -719,7 +771,6 @@ static int dnbd3_socket_connect(struct dnbd3_sock *sock,
 	if (result <= 0) {
 		error_sock(sock, "receive cmd select image %s failed",
 				dev->imgname);
-		mutex_unlock(&sock->tx_lock);
 		result = -EIO;
 		goto error;
 	}
@@ -736,19 +787,15 @@ static int dnbd3_socket_connect(struct dnbd3_sock *sock,
 	if (dnbd3_send_request_cmd(sock, CMD_GET_SERVERS) <= 0) {
 		error_sock(sock, "failed to get servers in discovery");
 	}
-	mutex_unlock(&sock->tx_lock);
 
 	return 0;
 error:
 	server->failures++;
-	sock->panic = true;
 	if (sock->sock) {
 		kernel_sock_shutdown(sock->sock, SHUT_RDWR);
 		cancel_work_sync(&sock->receive_worker);
-		sock_release(sock->sock);
-		sock->sock = NULL;
+		dnbd3_sock_release(sock);
 	}
-	mutex_destroy(&sock->tx_lock);
 	return result;
 }
 
@@ -762,7 +809,6 @@ static int dnbd3_socket_disconnect(struct dnbd3_sock *sock)
 	cancel_work_sync(&sock->keepalive_worker);
 
 	debug_sock(sock, "socket disconnect");
-	mutex_lock(&sock->tx_lock);
 
 	/*
 	 * Important sequence to shut down socket
@@ -777,20 +823,17 @@ static int dnbd3_socket_disconnect(struct dnbd3_sock *sock)
 	 */
 	if (sock->sock) {
 		kernel_sock_shutdown(sock->sock, SHUT_RDWR);
-		dndb3_reque_busy_requests(sock);
 	}
-	mutex_unlock(&sock->tx_lock);
-	mutex_destroy(&sock->tx_lock);
 
 	cancel_work_sync(&sock->receive_worker);
 
+	dndb3_reque_busy_requests(sock);
+
 	if (sock->sock) {
-		sock_release(sock->sock);
-		sock->sock = NULL;
+		dnbd3_sock_release(sock);
 	}
 
 	sock->server = NULL;
-	sock->panic = false;
 	return 0;
 }
 
@@ -849,12 +892,13 @@ int dnbd3_net_disconnect(struct dnbd3_device *dev)
 	/* be sure it does not recover while disconnecting */
 	cancel_work_sync(&dev->discovery_worker);
 	cancel_work_sync(&dev->panic_worker);
-
 	for (i = 0; i < NUMBER_CONNECTIONS; i++) {
 		if (dev->socks[i].sock) {
+			mutex_lock(&dev->socks[i].tx_lock);
 			if (dnbd3_socket_disconnect(&dev->socks[i])) {
 				result = -EIO;
 			}
+			mutex_unlock(&dev->socks[i].tx_lock);
 		}
 	}
 	dev->connected = false;