net/rds: Refactor sendmsg ancillary data processing

Add a process_send_cmsg() to struct rds_transport for transport
specific sendmsg ancillary data processing.  This separates
ancillary data processing for each transport for better error
handling.

Orabug: 31979196

Signed-off-by: Ka-Cheong Poon <ka-cheong.poon@oracle.com>
Tested-by: Gerald Gibson <gerald.gibson@oracle.com>
Acked-by: Håkon Bugge <haakon.bugge@oracle.com>
Reviewed-by: Hans Westgaard Ry <hans.westgaard.ry@oracle.com>

Author: kcp-git

net/rds/ib.c

@@ -1342,6 +1342,8 @@ struct rds_transport rds_ib_transport = {
 	.free_mr		= rds_ib_free_mr,
 	.flush_mrs		= rds_ib_flush_mrs,
 	.sock_release		= rds_rdma_sock_release,
+	.process_send_cmsg	= rds_rdma_process_send_cmsg,
+
 	.t_owner		= THIS_MODULE,
 	.t_name			= "infiniband",
 	.t_type			= RDS_TRANS_IB,

net/rds/rdma.c

@@ -550,8 +550,8 @@ int rds_rdma_extra_size(struct rds_rdma_args *args)
  * The application asks for a RDMA transfer.
  * Extract all arguments and set up the rdma_op
  */
-int rds_cmsg_rdma_args(struct rds_sock *rs, struct rds_message *rm,
-			  struct cmsghdr *cmsg)
+static int rds_cmsg_rdma_args(struct rds_sock *rs, struct rds_message *rm,
+			      struct cmsghdr *cmsg)
 {
 	struct rds_rdma_args *args;
 	struct rds_iovec vec;
@@ -705,8 +705,8 @@ int rds_cmsg_rdma_args(struct rds_sock *rs, struct rds_message *rm,
  * The application wants us to pass an RDMA destination (aka MR)
  * to the remote
  */
-int rds_cmsg_rdma_dest(struct rds_sock *rs, struct rds_message *rm,
-			  struct cmsghdr *cmsg)
+static int rds_cmsg_rdma_dest(struct rds_sock *rs, struct rds_message *rm,
+			      struct cmsghdr *cmsg)
 {
 	unsigned long flags;
 	struct rds_mr *mr;
@@ -761,8 +761,8 @@ static void dec_rdma_map_pending(struct rds_conn_path *cp)
  * in rm->m_rdma_cookie. This causes it to be sent along to the peer
  * in an extension header.
  */
-int rds_cmsg_rdma_map(struct rds_sock *rs, struct rds_message *rm,
-			  struct cmsghdr *cmsg)
+static int rds_cmsg_rdma_map(struct rds_sock *rs, struct rds_message *rm,
+			     struct cmsghdr *cmsg)
 {
 	int ret;
 	if (cmsg->cmsg_len < CMSG_LEN(sizeof(struct rds_get_mr_args))
@@ -788,8 +788,8 @@ int rds_cmsg_rdma_map(struct rds_sock *rs, struct rds_message *rm,
 /*
  * Fill in rds_message for an atomic request.
  */
-int rds_cmsg_atomic(struct rds_sock *rs, struct rds_message *rm,
-		    struct cmsghdr *cmsg)
+static int rds_cmsg_atomic(struct rds_sock *rs, struct rds_message *rm,
+			   struct cmsghdr *cmsg)
 {
 	struct page *page = NULL;
 	struct rds_atomic_args *args;
@@ -861,3 +861,68 @@ int rds_cmsg_atomic(struct rds_sock *rs, struct rds_message *rm,
 
 	return ret;
 }
+
+static int rds_cmsg_asend(struct rds_sock *rs, struct rds_message *rm,
+			  struct cmsghdr *cmsg)
+{
+	struct rds_asend_args *args;
+
+	if (!rds_async_send_enabled)
+		return -EINVAL;
+
+	if (cmsg->cmsg_len < CMSG_LEN(sizeof(struct rds_asend_args)))
+		return -EINVAL;
+
+	args = CMSG_DATA(cmsg);
+	rm->data.op_notifier = kzalloc(sizeof(*rm->data.op_notifier),
+				       GFP_KERNEL);
+	if (!rm->data.op_notifier)
+		return -ENOMEM;
+
+	rm->data.op_notify = !!(args->flags & RDS_SEND_NOTIFY_ME);
+	rm->data.op_notifier->n_user_token = args->user_token;
+	rm->data.op_notifier->n_status = RDS_RDMA_SEND_SUCCESS;
+	rm->data.op_async = 1;
+
+	return 0;
+}
+
+int rds_rdma_process_send_cmsg(struct rds_sock *rs, struct rds_message *rm,
+			       struct cmsghdr *cmsg)
+{
+	int ret;
+
+	switch (cmsg->cmsg_type) {
+	case RDS_CMSG_RDMA_ARGS:
+		ret = rds_cmsg_rdma_args(rs, rm, cmsg);
+		break;
+
+	case RDS_CMSG_RDMA_DEST:
+		ret = rds_cmsg_rdma_dest(rs, rm, cmsg);
+		break;
+
+	case RDS_CMSG_RDMA_MAP:
+		ret = rds_cmsg_rdma_map(rs, rm, cmsg);
+		if (ret == -ENODEV)
+			/* Accommodate the get_mr() case which can fail
+			 * if connection isn't established yet.
+			 */
+			ret = -EAGAIN;
+		break;
+	case RDS_CMSG_ATOMIC_CSWP:
+	case RDS_CMSG_ATOMIC_FADD:
+		ret = rds_cmsg_atomic(rs, rm, cmsg);
+		break;
+
+	case RDS_CMSG_ASYNC_SEND:
+		ret = rds_cmsg_asend(rs, rm, cmsg);
+		break;
+
+	default:
+		ret = -EINVAL;
+		break;
+	}
+
+	return ret;
+}
+EXPORT_SYMBOL_GPL(rds_rdma_process_send_cmsg);

net/rds/rds.h

@@ -775,6 +775,8 @@ struct rds_transport {
 	void (*check_migration)(struct rds_connection *conn,
 				struct rdma_cm_event *event);
 	void (*sock_release)(struct rds_sock *rs);
+	int (*process_send_cmsg)(struct rds_sock *rs, struct rds_message *rm,
+				 struct cmsghdr *cmsg);
 
 	atomic64_t rds_avg_conn_jf;
 };
@@ -1283,21 +1285,14 @@ int rds_get_mr_for_dest(struct rds_sock *rs, char __user *optval, int optlen);
 int rds_free_mr(struct rds_sock *rs, char __user *optval, int optlen);
 void rds_rdma_drop_keys(struct rds_sock *rs);
 int rds_rdma_extra_size(struct rds_rdma_args *args);
-int rds_cmsg_rdma_args(struct rds_sock *rs, struct rds_message *rm,
-			  struct cmsghdr *cmsg);
-int rds_cmsg_rdma_dest(struct rds_sock *rs, struct rds_message *rm,
-			  struct cmsghdr *cmsg);
-int rds_cmsg_rdma_args(struct rds_sock *rs, struct rds_message *rm,
-			  struct cmsghdr *cmsg);
-int rds_cmsg_rdma_map(struct rds_sock *rs, struct rds_message *rm,
-			  struct cmsghdr *cmsg);
+
+int rds_rdma_process_send_cmsg(struct rds_sock *rs, struct rds_message *rm,
+			       struct cmsghdr *cmsg);
 void rds_rdma_free_op(struct rm_rdma_op *ro);
 void rds_atomic_free_op(struct rm_atomic_op *ao);
 void rds_rdma_send_complete(struct rds_message *rm, int wc_status);
 void rds_atomic_send_complete(struct rds_message *rm, int wc_status);
 void rds_asend_complete(struct rds_message *rm, int wc_status);
-int rds_cmsg_atomic(struct rds_sock *rs, struct rds_message *rm,
-		    struct cmsghdr *cmsg);
 
 void __rds_put_mr_final(struct kref *kref);
 

net/rds/send.c

@@ -1149,32 +1149,8 @@ static int rds_rm_size(struct msghdr *msg, int data_len)
 	return size;
 }
 
-static int rds_cmsg_asend(struct rds_sock *rs, struct rds_message *rm,
-		struct cmsghdr *cmsg)
-{
-	struct rds_asend_args *args;
-
-	if (!rds_async_send_enabled)
-		return -EINVAL;
-
-	if (cmsg->cmsg_len < CMSG_LEN(sizeof(struct rds_asend_args)))
-		return -EINVAL;
-
-	args = CMSG_DATA(cmsg);
-	rm->data.op_notifier = kzalloc(sizeof(*rm->data.op_notifier), GFP_KERNEL);
-	if (!rm->data.op_notifier)
-		return -ENOMEM;
-
-	rm->data.op_notify = !!(args->flags & RDS_SEND_NOTIFY_ME);
-	rm->data.op_notifier->n_user_token = args->user_token;
-	rm->data.op_notifier->n_status = RDS_RDMA_SEND_SUCCESS;
-	rm->data.op_async = 1;
-
-	return 0;
-}
-
-static int rds_cmsg_send(struct rds_sock *rs, struct rds_message *rm,
-			 struct msghdr *msg, int *allocated_mr)
+static int rds_cmsg_send(struct rds_connection *conn, struct rds_sock *rs,
+			 struct rds_message *rm, struct msghdr *msg)
 {
 	struct cmsghdr *cmsg;
 	int ret = 0;
@@ -1186,41 +1162,13 @@ static int rds_cmsg_send(struct rds_sock *rs, struct rds_message *rm,
 		if (cmsg->cmsg_level != SOL_RDS)
 			continue;
 
-		/* As a side effect, RDMA_DEST and RDMA_MAP will set
-		 * rm->rdma.m_rdma_cookie and rm->rdma.m_rdma_mr.
+		/* Non-transport specific ancillary data should be handled
+		 * here before calling transport specific ancillary data
+		 * handler.
 		 */
-		switch (cmsg->cmsg_type) {
-		case RDS_CMSG_RDMA_ARGS:
-			ret = rds_cmsg_rdma_args(rs, rm, cmsg);
-			break;
-
-		case RDS_CMSG_RDMA_DEST:
-			ret = rds_cmsg_rdma_dest(rs, rm, cmsg);
-			break;
-
-		case RDS_CMSG_RDMA_MAP:
-			ret = rds_cmsg_rdma_map(rs, rm, cmsg);
-			if (!ret)
-				*allocated_mr = 1;
-			else if (ret == -ENODEV)
-				/* Accomodate the get_mr() case which can fail
-				 * if connection isn't established yet.
-				 */
-				ret = -EAGAIN;
-			break;
-		case RDS_CMSG_ATOMIC_CSWP:
-		case RDS_CMSG_ATOMIC_FADD:
-			ret = rds_cmsg_atomic(rs, rm, cmsg);
-			break;
-
-		case RDS_CMSG_ASYNC_SEND:
-			ret = rds_cmsg_asend(rs, rm, cmsg);
-			break;
-
-		default:
+		if (!conn->c_trans->process_send_cmsg)
 			return -EINVAL;
-		}
-
+		ret = conn->c_trans->process_send_cmsg(rs, rm, cmsg);
 		if (ret)
 			break;
 	}
@@ -1299,7 +1247,7 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 	struct rds_message *rm = NULL;
 	struct rds_connection *conn;
 	int ret = 0;
-	int queued = 0, allocated_mr = 0;
+	int queued = 0;
 	int nonblock = msg->msg_flags & MSG_DONTWAIT;
 	long timeo = sock_sndtimeo(sk, nonblock);
 	size_t total_payload_len = payload_len, rdma_payload_len = 0;
@@ -1542,10 +1490,10 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 	rm->m_conn_path = cpath;
 
 	/* Parse any control messages the user may have included. */
-	ret = rds_cmsg_send(rs, rm, msg, &allocated_mr);
+	ret = rds_cmsg_send(conn, rs, rm, msg);
 	if (ret) {
 		/* Trigger connection so that its ready for the next retry */
-		if ( ret ==  -EAGAIN)
+		if (ret == -EAGAIN)
 			rds_conn_connect_if_down(conn);
 		goto out;
 	}
@@ -1643,11 +1591,13 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)
 	/* If the user included a RDMA_MAP cmsg, we allocated a MR on the fly.
 	 * If the sendmsg goes through, we keep the MR. If it fails with EAGAIN
 	 * or in any other way, we need to destroy the MR again */
-	if (allocated_mr)
-		rds_rdma_unuse(rs, rds_rdma_cookie_key(rm->m_rdma_cookie), 1);
-
-	if (rm)
+	if (rm) {
+		if (rm->rdma.op_implicit_mr)
+			rds_rdma_unuse(rs,
+				       rds_rdma_cookie_key(rm->m_rdma_cookie),
+				       1);
 		rds_message_put(rm);
+	}
 	return ret;
 }