smc: send data (through RDMA)

copy data to kernel send buffer, and trigger RDMA write

Signed-off-by: Ursula Braun <ubraun@linux.vnet.ibm.com>
Signed-off-by: David S. Miller <davem@davemloft.net>

net/smc/Makefile

@@ -1,3 +1,3 @@
 obj-$(CONFIG_SMC)	+= smc.o
 smc-y := af_smc.o smc_pnet.o smc_ib.o smc_clc.o smc_core.o smc_wr.o smc_llc.o
-smc-y += smc_cdc.o
+smc-y += smc_cdc.o smc_tx.o

net/smc/af_smc.c

@@ -37,6 +37,7 @@
 #include "smc_core.h"
 #include "smc_ib.h"
 #include "smc_pnet.h"
+#include "smc_tx.h"
 
 static DEFINE_MUTEX(smc_create_lgr_pending);	/* serialize link group
 						 * creation
@@ -410,6 +411,8 @@ static int smc_connect_rdma(struct smc_sock *smc)
 	}
 
 	mutex_unlock(&smc_create_lgr_pending);
+	smc_tx_init(smc);
+
 out_connected:
 	smc_copy_sock_settings_to_clc(smc);
 	smc->sk.sk_state = SMC_ACTIVE;
@@ -751,6 +754,8 @@ static void smc_listen_work(struct work_struct *work)
 			goto decline_rdma;
 	}
 
+	smc_tx_init(new_smc);
+
 out_connected:
 	sk_refcnt_debug_inc(newsmcsk);
 	newsmcsk->sk_state = SMC_ACTIVE;
@@ -924,7 +929,7 @@ static int smc_sendmsg(struct socket *sock, struct msghdr *msg, size_t len)
 	if (smc->use_fallback)
 		rc = smc->clcsock->ops->sendmsg(smc->clcsock, msg, len);
 	else
-		rc = sock_no_sendmsg(sock, msg, len);
+		rc = smc_tx_sendmsg(smc, msg, len);
 out:
 	release_sock(sk);
 	return rc;
@@ -1005,6 +1010,12 @@ static unsigned int smc_poll(struct file *file, struct socket *sock,
 			mask |= smc_accept_poll(sk);
 		if (sk->sk_err)
 			mask |= POLLERR;
+		if (atomic_read(&smc->conn.sndbuf_space)) {
+			mask |= POLLOUT | POLLWRNORM;
+		} else {
+			sk_set_bit(SOCKWQ_ASYNC_NOSPACE, sk);
+			set_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
+		}
 		/* for now - to be enhanced in follow-on patch */
 	}
 

net/smc/smc.h

@@ -133,6 +133,7 @@ struct smc_connection {
 	atomic_t		sndbuf_space;	/* remaining space in sndbuf */
 	u16			tx_cdc_seq;	/* sequence # for CDC send */
 	spinlock_t		send_lock;	/* protect wr_sends */
+	struct work_struct	tx_work;	/* retry of smc_cdc_msg_send */
 
 	struct smc_host_cdc_msg	local_rx_ctrl;	/* filled during event_handl.
 						 * .prod cf. TCP rcv_nxt

net/smc/smc_cdc.c

@@ -14,6 +14,7 @@
 #include "smc.h"
 #include "smc_wr.h"
 #include "smc_cdc.h"
+#include "smc_tx.h"
 
 /********************************** send *************************************/
 
@@ -52,7 +53,7 @@ static void smc_cdc_tx_handler(struct smc_wr_tx_pend_priv *pnd_snd,
 			       smc_curs_read(&cdcpend->cursor, cdcpend->conn),
 			       cdcpend->conn);
 	}
-	/* subsequent patch: wake if send buffer space available */
+	smc_tx_sndbuf_nonfull(smc);
 	bh_unlock_sock(&smc->sk);
 }
 
@@ -204,7 +205,9 @@ static void smc_cdc_msg_recv_action(struct smc_sock *smc,
 		/* subsequent patch: terminate connection */
 
 	/* piggy backed tx info */
-	/* subsequent patch: wake receivers if receive buffer space available */
+	/* trigger sndbuf consumer: RDMA write into peer RMBE and CDC */
+	if (diff_cons && smc_tx_prepared_sends(conn))
+		smc_tx_sndbuf_nonempty(conn);
 
 	/* subsequent patch: trigger socket release if connection closed */