feat: cancel request bundle when socket is closed

bp_socket/af_bp.c

@@ -29,7 +29,10 @@ static struct sock* bp_alloc_socket(struct net* net, int kern)
 	bp = bp_sk(sk);
 	skb_queue_head_init(&bp->queue);
 	init_waitqueue_head(&bp->wait_queue);
-
+	bp->bp_node_id = 0;
+	bp->bp_service_id = 0;
+	bp->recv_aborted = false;
+	bp->has_bound = false;
 out:
 	return sk;
 }
@@ -132,8 +135,7 @@ int bp_bind(struct socket* sock, struct sockaddr* uaddr, int addr_len)
 		if (iter_bp->bp_service_id == service_id
 		    && iter_bp->bp_node_id == node_id) {
 			read_unlock_bh(&bp_list_lock);
-			err = -EADDRINUSE;
-			goto out;
+			return -EADDRINUSE;
 		}
 	}
 	read_unlock_bh(&bp_list_lock);
@@ -142,14 +144,21 @@ int bp_bind(struct socket* sock, struct sockaddr* uaddr, int addr_len)
 	bp = bp_sk(sk);
 	bp->bp_service_id = service_id;
 	bp->bp_node_id = node_id;
+
+	err = open_endpoint_doit(bp->bp_node_id, bp->bp_service_id, 8443);
+	if (err < 0) {
+		pr_err("bp_bind: open_endpoint_doit failed (%d)\n", err);
+		release_sock(sk);
+		return err;
+	}
+
 	write_lock_bh(&bp_list_lock);
 	sk_add_node(sk, &bp_list);
 	write_unlock_bh(&bp_list_lock);
 
-out:
+	bp->has_bound = true;
 	release_sock(sk);
-
-	return err;
+	return 0;
 }
 
 int bp_release(struct socket* sock)
@@ -164,6 +173,14 @@ int bp_release(struct socket* sock)
 	sock_orphan(sk);
 	bp = bp_sk(sk);
 
+	if (bp->has_bound && bp->recv_aborted) {
+		abort_endpoint_doit(bp->bp_node_id, bp->bp_service_id, 8443);
+	} else {
+		if (bp->has_bound)
+			close_endpoint_doit(
+			    bp->bp_node_id, bp->bp_service_id, 8443);
+	}
+
 	write_lock_bh(&bp_list_lock);
 	sk_del_node_init(sk);
 	write_unlock_bh(&bp_list_lock);
@@ -247,13 +264,12 @@ int bp_sendmsg(struct socket* sock, struct msghdr* msg, size_t size)
 
 	ret = send_bundle_doit(
 	    sockid, (char*)payload, size, node_id, service_id, 8443);
-	kfree(payload);
-
 	if (ret < 0) {
 		pr_err("bp_sendmsg: send_bundle_doit failed (%d)\n", ret);
 		return ret;
 	}
 
+	kfree(payload);
 	return size;
 }
 
@@ -274,12 +290,7 @@ int bp_recvmsg(struct socket* sock, struct msghdr* msg, size_t size, int flags)
 	    bp->wait_queue, !skb_queue_empty(&bp->queue));
 	if (ret < 0) {
 		pr_err("bp_recvmsg: interrupted while waiting\n");
-		goto out_unlock;
-	}
-
-	if (sock_flag(sk, SOCK_DEAD)) {
-		pr_err("bp_recvmsg: socket closed while waiting\n");
-		ret = -ECONNRESET;
+		bp->recv_aborted = true;
 		goto out_unlock;
 	}
 

bp_socket/af_bp.h

@@ -17,6 +17,8 @@ struct bp_sock {
 	u_int8_t bp_service_id;
 	struct sk_buff_head queue;
 	wait_queue_head_t wait_queue;
+	bool has_bound;
+	bool recv_aborted;
 };
 
 int bp_bind(struct socket* sock, struct sockaddr* addr, int addr_len);

bp_socket/bp_genl.c

@@ -43,6 +43,142 @@ int fail_doit(struct sk_buff* skb, struct genl_info* info)
 	return -1;
 }
 
+int open_endpoint_doit(u32 node_id, u32 service_id, int port_id)
+{
+	int ret = 0;
+	void* hdr;
+	struct sk_buff* msg;
+	int msg_size;
+
+	/* Taille du message avec deux attributs */
+	msg_size = nla_total_size(sizeof(u32)) + nla_total_size(sizeof(u32));
+	msg = genlmsg_new(msg_size + GENL_HDRLEN, GFP_KERNEL);
+	if (!msg) {
+		pr_err("bp_open_request: failed to allocate message buffer\n");
+		return -ENOMEM;
+	}
+
+	hdr = genlmsg_put(msg, 0, 0, &genl_fam, 0, BP_GENL_CMD_OPEN_ENDPOINT);
+	if (!hdr) {
+		pr_err("bp_open_request: failed to create genetlink header\n");
+		nlmsg_free(msg);
+		return -EMSGSIZE;
+	}
+
+	ret = nla_put_u32(msg, BP_GENL_A_NODE_ID, node_id);
+	if (ret) {
+		pr_err("bp_open_request: failed to add NODE_ID\n");
+		goto fail;
+	}
+
+	ret = nla_put_u32(msg, BP_GENL_A_SERVICE_ID, service_id);
+	if (ret) {
+		pr_err("bp_open_request: failed to add SERVICE_ID\n");
+		goto fail;
+	}
+
+	genlmsg_end(msg, hdr);
+	ret = genlmsg_unicast(&init_net, msg, port_id);
+	if (ret != 0) {
+		pr_err("bp_open_request: genlmsg_unicast failed (%d)\n", ret);
+	}
+	return ret;
+
+fail:
+	genlmsg_cancel(msg, hdr);
+	nlmsg_free(msg);
+	return ret;
+}
+
+int close_endpoint_doit(u32 node_id, u32 service_id, int port_id)
+{
+	int ret = 0;
+	void* hdr;
+	struct sk_buff* msg;
+	int msg_size;
+
+	/* Compute total size of Netlink attributes */
+	msg_size = nla_total_size(sizeof(u32)) + nla_total_size(sizeof(u32));
+	/* Allocate a new buffer */
+	msg = genlmsg_new(msg_size + GENL_HDRLEN, GFP_KERNEL);
+	if (!msg) {
+		pr_err("bp_close_request: failed to allocate message buffer\n");
+		return -ENOMEM;
+	}
+	/* Generic Netlink header */
+	hdr = genlmsg_put(msg, 0, 0, &genl_fam, 0, BP_GENL_CMD_CLOSE_ENDPOINT);
+	if (!hdr) {
+		pr_err("bp_close_request: failed to create genetlink header\n");
+		nlmsg_free(msg);
+		return -EMSGSIZE;
+	}
+	/* And the message */
+	ret = nla_put_u32(msg, BP_GENL_A_NODE_ID, node_id);
+	if (ret) {
+		pr_err("bp_close_request: failed to put NODE_ID (%d)\n", ret);
+		goto fail;
+	}
+	ret = nla_put_u32(msg, BP_GENL_A_SERVICE_ID, service_id);
+	if (ret) {
+		pr_err(
+		    "bp_close_request: failed to put SERVICE_ID (%d)\n", ret);
+		goto fail;
+	}
+	genlmsg_end(msg, hdr);
+	ret = genlmsg_unicast(&init_net, msg, port_id);
+	if (ret != 0) {
+		pr_err("bp_close_request: genlmsg_unicast failed (%d)\n", ret);
+	}
+	return ret;
+fail:
+	genlmsg_cancel(msg, hdr);
+	nlmsg_free(msg);
+	return ret;
+}
+
+int abort_endpoint_doit(u32 node_id, u32 service_id, int port_id)
+{
+	int ret = 0;
+	void* hdr;
+	struct sk_buff* msg;
+	int msg_size;
+
+	msg_size = 2 * nla_total_size(sizeof(u32));
+	msg = genlmsg_new(msg_size + GENL_HDRLEN, GFP_KERNEL);
+	if (!msg) {
+		pr_err("cancel_bundle: failed to allocate message buffer\n");
+		return -ENOMEM;
+	}
+
+	hdr = genlmsg_put(msg, 0, 0, &genl_fam, 0, BP_GENL_CMD_ABORT_ENDPOINT);
+	if (!hdr) {
+		pr_err("cancel_bundle: failed to create genetlink header\n");
+		nlmsg_free(msg);
+		return -EMSGSIZE;
+	}
+
+	if ((ret = nla_put_u32(msg, BP_GENL_A_SERVICE_ID, service_id))) {
+		pr_err("cancel_bundle: failed to put service_id\n");
+		genlmsg_cancel(msg, hdr);
+		nlmsg_free(msg);
+		return ret;
+	}
+
+	if ((ret = nla_put_u32(msg, BP_GENL_A_NODE_ID, node_id))) {
+		pr_err("cancel_bundle: failed to put node_id\n");
+		genlmsg_cancel(msg, hdr);
+		nlmsg_free(msg);
+		return ret;
+	}
+
+	genlmsg_end(msg, hdr);
+	ret = genlmsg_unicast(&init_net, msg, port_id);
+	if (ret != 0) {
+		pr_err("cancel_bundle: unicast failed (%d)\n", ret);
+	}
+	return ret;
+}
+
 int send_bundle_doit(u64 sockid, const char* payload, int payload_size,
     u32 node_id, u32 service_id, int port_id)
 {
@@ -160,23 +296,16 @@ int deliver_bundle_doit(struct sk_buff* skb, struct genl_info* info)
 	size_t payload_len;
 	struct sk_buff* new_skb;
 
-	pr_info("TRIGGER: received message\n");
-
-	if (!info->attrs[BP_GENL_A_SERVICE_ID]) {
-		pr_err("attribute missing from message\n");
+	if (!info->attrs[BP_GENL_A_SERVICE_ID]
+	    || !info->attrs[BP_GENL_A_PAYLOAD]) {
+		pr_err("deliver_bundle: missing required attributes\n");
 		return -EINVAL;
 	}
-	service_id = nla_get_u32(info->attrs[BP_GENL_A_SERVICE_ID]);
 
-	if (!info->attrs[BP_GENL_A_PAYLOAD]) {
-		pr_err("empty message received\n");
-		return -EINVAL;
-	}
+	service_id = nla_get_u32(info->attrs[BP_GENL_A_SERVICE_ID]);
 	payload = nla_data(info->attrs[BP_GENL_A_PAYLOAD]);
 	payload_len = nla_len(info->attrs[BP_GENL_A_PAYLOAD]);
 
-	pr_info("Message for service %d: %s\n", service_id, payload);
-
 	new_skb = alloc_skb(payload_len, GFP_KERNEL);
 	if (!new_skb) {
 		pr_err("Failed to allocate sk_buff for payload\n");
@@ -187,18 +316,18 @@ int deliver_bundle_doit(struct sk_buff* skb, struct genl_info* info)
 	read_lock_bh(&bp_list_lock);
 	sk_for_each(sk, &bp_list)
 	{
-		lock_sock(sk);
+		bh_lock_sock(sk);
 		bp = bp_sk(sk);
 
 		if (bp->bp_service_id == service_id) {
 
 			skb_queue_tail(&bp->queue, new_skb);
 			if (waitqueue_active(&bp->wait_queue))
 				wake_up_interruptible(&bp->wait_queue);
-			release_sock(sk);
+			bh_unlock_sock(sk);
 			break;
 		}
-		release_sock(sk);
+		bh_unlock_sock(sk);
 	}
 	read_unlock_bh(&bp_list_lock);
 

bp_socket/bp_genl.h

@@ -6,6 +6,9 @@
 extern struct genl_family genl_fam;
 
 int fail_doit(struct sk_buff* skb, struct genl_info* info);
+int open_endpoint_doit(u32 node_id, u32 service_id, int port_id);
+int close_endpoint_doit(u32 node_id, u32 service_id, int port_id);
+int abort_endpoint_doit(u32 node_id, u32 service_id, int port_id);
 int send_bundle_doit(u64 sockid, const char* payload, int payload_size,
     u32 node_id, u32 service_id, int port_id);
 int deliver_bundle_doit(struct sk_buff* skb, struct genl_info* info);

daemon/bp_genl.c

@@ -66,6 +66,12 @@ int genl_bp_sock_recvmsg_cb(struct nl_msg *msg, void *arg) {
     }
 
     switch (genlhdr->cmd) {
+    case BP_GENL_CMD_OPEN_ENDPOINT:
+        return handle_open_endpoint(daemon, attrs);
+    case BP_GENL_CMD_CLOSE_ENDPOINT:
+        return handle_close_endpoint(daemon, attrs);
+    case BP_GENL_CMD_ABORT_ENDPOINT:
+        return handle_abort_endpoint(daemon, attrs);
     case BP_GENL_CMD_SEND_BUNDLE:
         return handle_send_bundle(daemon, attrs);
     case BP_GENL_CMD_REQUEST_BUNDLE:

daemon/bp_genl.h

@@ -6,8 +6,6 @@
 
 struct nl_sock *genl_bp_sock_init(Daemon *daemon);
 void genl_bp_sock_close(Daemon *daemon);
-int genl_bp_sock_sendmsg(Daemon *self, void *payload, size_t len);
-int genl_bp_sock_recvmsg(Daemon *self, void *payload, size_t len);
 int genl_bp_sock_recvmsg_cb(struct nl_msg *msg, void *arg);
 
 #endif