feat: get source eid with bp_recvmsg in struct msghdr* msg

bp_socket/af_bp.c

@@ -28,8 +28,8 @@ static struct sock* bp_alloc_socket(struct net* net, int kern)
 		sock_init_data(NULL, sk);
 
 		bp = bp_sk(sk);
-		skb_queue_head_init(&bp->queue);
-		init_waitqueue_head(&bp->wait_queue);
+		skb_queue_head_init(&bp->rx_queue);
+		init_waitqueue_head(&bp->rx_waitq);
 		bp->bp_node_id = 0;
 		bp->bp_service_id = 0;
 	}
@@ -178,7 +178,7 @@ int bp_release(struct socket* sock)
 		write_lock_bh(&bp_list_lock);
 		sk_del_node_init(sk);
 		write_unlock_bh(&bp_list_lock);
-		skb_queue_purge(&bp->queue);
+		skb_queue_purge(&bp->rx_queue);
 
 		sock->sk = NULL;
 		release_sock(sk);
@@ -192,9 +192,15 @@ int bp_sendmsg(struct socket* sock, struct msghdr* msg, size_t size)
 {
 	struct sockaddr_bp* addr;
 	void* payload;
-	u_int32_t service_id;
-	u_int32_t node_id;
+	u_int32_t dest_node_id, dest_service_id;
 	int ret;
+	struct bp_sock* bp = bp_sk(sock->sk);
+
+	if (bp->bp_node_id == 0 || bp->bp_service_id == 0) {
+		pr_err("bp_sendmsg: socket must be bound before sending\n");
+		ret = -EADDRNOTAVAIL;
+		goto out;
+	}
 
 	if (!msg->msg_name) {
 		pr_err("bp_sendmsg: no destination address provided\n");
@@ -226,21 +232,21 @@ int bp_sendmsg(struct socket* sock, struct msghdr* msg, size_t size)
 		goto out;
 	}
 
-	service_id = addr->bp_addr.ipn.service_id;
-	node_id = addr->bp_addr.ipn.node_id;
+	dest_node_id = addr->bp_addr.ipn.node_id;
+	dest_service_id = addr->bp_addr.ipn.service_id;
 
 	// https://www.rfc-editor.org/rfc/rfc9758.html#name-node-numbers
-	if (node_id > 0xFFFFFFFF) {
+	if (dest_node_id > 0xFFFFFFFF) {
 		pr_err("bp_bind: invalid node ID (must be in [0;2^31])\n");
 		ret = -EINVAL;
 		goto out;
 	}
 
 	// https://www.rfc-editor.org/rfc/rfc9758.html#name-service-numbers
-	if (service_id < 1 || service_id > 0xFFFFFFFF) {
+	if (dest_service_id < 1 || dest_service_id > 0xFFFFFFFF) {
 		pr_err("bp_bind: invalid service ID %d (must be in "
 		       "[1;2^31])\n",
-		    service_id);
+		    dest_service_id);
 		ret = -EINVAL;
 		goto out;
 	}
@@ -265,8 +271,8 @@ int bp_sendmsg(struct socket* sock, struct msghdr* msg, size_t size)
 			goto err_free;
 		}
 
-		ret = send_bundle_doit(
-		    payload, size, node_id, service_id, 8443);
+		ret = send_bundle_doit(payload, size, dest_node_id,
+		    dest_service_id, bp->bp_node_id, bp->bp_service_id, 8443);
 		if (ret < 0) {
 			pr_err(
 			    "bp_sendmsg: send_bundle_doit failed (%d)\n", ret);
@@ -289,31 +295,45 @@ int bp_recvmsg(struct socket* sock, struct msghdr* msg, size_t size, int flags)
 	struct sock* sk;
 	struct bp_sock* bp;
 	struct sk_buff* skb;
+	struct sockaddr_bp* src_addr;
 	int ret;
 
 	sk = sock->sk;
 	lock_sock(sk);
 	bp = bp_sk(sk);
+
+	if (bp->bp_node_id == 0 || bp->bp_service_id == 0) {
+		pr_err("bp_recvmsg: socket must be bound before receiving\n");
+		ret = -EADDRNOTAVAIL;
+		goto out;
+	}
+
 	ret = request_bundle_doit(bp->bp_node_id, bp->bp_service_id, 8443);
 	if (ret < 0) {
 		pr_err("bp_recvmsg: request_bundle_doit failed (%d)\n", ret);
 		goto out;
 	}
 
 	ret = wait_event_interruptible(
-	    bp->wait_queue, !skb_queue_empty(&bp->queue));
+	    bp->rx_waitq, !skb_queue_empty(&bp->rx_queue) || bp->rx_canceled);
 	if (ret < 0) {
 		pr_err("bp_recvmsg: interrupted while waiting\n");
 		goto out;
 	}
 
+	if (bp->rx_canceled) {
+		pr_info("bp_recvmsg: bundle request canceled\n");
+		ret = -ECANCELED;
+		goto out;
+	}
+
 	if (sock_flag(sk, SOCK_DEAD)) {
 		pr_err("bp_recvmsg: socket closed while waiting\n");
 		ret = -ESHUTDOWN;
 		goto out;
 	}
 
-	skb = skb_dequeue(&bp->queue);
+	skb = skb_dequeue(&bp->rx_queue);
 	if (!skb) {
 		pr_info("bp_recvmsg: no messages in the queue for service %d\n",
 		    bp->bp_service_id);
@@ -329,6 +349,16 @@ int bp_recvmsg(struct socket* sock, struct msghdr* msg, size_t size, int flags)
 		goto out;
 	}
 
+	if (msg->msg_name) {
+		src_addr = (struct sockaddr_bp*)msg->msg_name;
+		src_addr->bp_family = AF_BP;
+		src_addr->bp_scheme = BP_SCHEME_IPN;
+		src_addr->bp_addr.ipn.node_id = BP_SKB_CB(skb)->src_node_id;
+		src_addr->bp_addr.ipn.service_id
+		    = BP_SKB_CB(skb)->src_service_id;
+		msg->msg_namelen = sizeof(struct sockaddr_bp);
+	}
+
 	if (copy_to_iter(skb->data, skb->len, &msg->msg_iter) != skb->len) {
 		pr_err("bp_recvmsg: failed to copy data to user space\n");
 		ret = -EFAULT;

bp_socket/af_bp.h

@@ -4,7 +4,13 @@
 #include <linux/net.h>
 #include <net/sock.h>
 
+struct bp_skb_cb {
+	u_int32_t src_node_id;
+	u_int32_t src_service_id;
+};
+
 #define bp_sk(ptr) container_of(ptr, struct bp_sock, sk)
+#define BP_SKB_CB(skb) ((struct bp_skb_cb*)((skb)->cb))
 
 extern struct hlist_head bp_list;
 extern rwlock_t bp_list_lock;
@@ -15,8 +21,9 @@ struct bp_sock {
 	struct sock sk;
 	u_int32_t bp_node_id;
 	u_int8_t bp_service_id;
-	struct sk_buff_head queue;
-	wait_queue_head_t wait_queue;
+	struct sk_buff_head rx_queue;
+	wait_queue_head_t rx_waitq;
+	bool rx_canceled;
 };
 
 int bp_bind(struct socket* sock, struct sockaddr* addr, int addr_len);