net/smc: register new rmbs with the peer

Register new rmb buffers with the remote peer by exchanging a
confirm_rkey llc message.

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

diff --git a/net/smc/af_smc.c b/net/smc/af_smc.c
index 83403be..397ba21 100644 (file)
--- a/net/smc/af_smc.c
+++ b/net/smc/af_smc.c
@@ -293,14 +293,22 @@ static void smc_copy_sock_settings_to_smc(struct smc_sock *smc)
        smc_copy_sock_settings(&smc->sk, smc->clcsock->sk, SK_FLAGS_CLC_TO_SMC);
 }
 
-/* register a new rmb */
-static int smc_reg_rmb(struct smc_link *link, struct smc_buf_desc *rmb_desc)
+/* register a new rmb, optionally send confirm_rkey msg to register with peer */
+static int smc_reg_rmb(struct smc_link *link, struct smc_buf_desc *rmb_desc,
+                      bool conf_rkey)
 {
        /* register memory region for new rmb */
        if (smc_wr_reg_send(link, rmb_desc->mr_rx[SMC_SINGLE_LINK])) {
                rmb_desc->regerr = 1;
                return -EFAULT;
        }
+       if (!conf_rkey)
+               return 0;
+       /* exchange confirm_rkey msg with peer */
+       if (smc_llc_do_confirm_rkey(link, rmb_desc)) {
+               rmb_desc->regerr = 1;
+               return -EFAULT;
+       }
        return 0;
 }
 
@@ -334,7 +342,7 @@ static int smc_clnt_conf_first_link(struct smc_sock *smc)
 
        smc_wr_remember_qp_attr(link);
 
-       if (smc_reg_rmb(link, smc->conn.rmb_desc))
+       if (smc_reg_rmb(link, smc->conn.rmb_desc, false))
                return SMC_CLC_DECL_INTERR;
 
        /* send CONFIRM LINK response over RoCE fabric */
@@ -488,7 +496,7 @@ static int smc_connect_rdma(struct smc_sock *smc)
                }
        } else {
                if (!smc->conn.rmb_desc->reused) {
-                       if (smc_reg_rmb(link, smc->conn.rmb_desc)) {
+                       if (smc_reg_rmb(link, smc->conn.rmb_desc, true)) {
                                reason_code = SMC_CLC_DECL_INTERR;
                                goto decline_rdma_unlock;
                        }
@@ -729,7 +737,7 @@ static int smc_serv_conf_first_link(struct smc_sock *smc)
 
        link = &lgr->lnk[SMC_SINGLE_LINK];
 
-       if (smc_reg_rmb(link, smc->conn.rmb_desc))
+       if (smc_reg_rmb(link, smc->conn.rmb_desc, false))
                return SMC_CLC_DECL_INTERR;
 
        /* send CONFIRM LINK request to client over the RoCE fabric */
@@ -866,7 +874,7 @@ static void smc_listen_work(struct work_struct *work)
 
        if (local_contact != SMC_FIRST_CONTACT) {
                if (!new_smc->conn.rmb_desc->reused) {
-                       if (smc_reg_rmb(link, new_smc->conn.rmb_desc)) {
+                       if (smc_reg_rmb(link, new_smc->conn.rmb_desc, true)) {
                                reason_code = SMC_CLC_DECL_INTERR;
                                goto decline_rdma_unlock;
                        }

diff --git a/net/smc/smc_core.c b/net/smc/smc_core.c
index 9c74844..29ae077 100644 (file)
--- a/net/smc/smc_core.c
+++ b/net/smc/smc_core.c
@@ -210,6 +210,7 @@ static int smc_lgr_create(struct smc_sock *smc,
        init_completion(&lnk->llc_confirm_resp);
        init_completion(&lnk->llc_add);
        init_completion(&lnk->llc_add_resp);
+       init_completion(&lnk->llc_confirm_rkey);
 
        smc->conn.lgr = lgr;
        rwlock_init(&lgr->conns_lock);

diff --git a/net/smc/smc_core.h b/net/smc/smc_core.h
index fca8624..9398e23 100644 (file)
--- a/net/smc/smc_core.h
+++ b/net/smc/smc_core.h
@@ -105,6 +105,8 @@ struct smc_link {
        struct delayed_work     llc_testlink_wrk; /* testlink worker */
        struct completion       llc_testlink_resp; /* wait for rx of testlink */
        int                     llc_testlink_time; /* testlink interval */
+       struct completion       llc_confirm_rkey; /* wait 4 rx of cnf rkey */
+       int                     llc_confirm_rkey_rc; /* rc from cnf rkey msg */
 };
 
 /* For now we just allow one parallel link per link group. The SMC protocol

diff --git a/net/smc/smc_llc.c b/net/smc/smc_llc.c
index 33b4d85..b118f80 100644 (file)
--- a/net/smc/smc_llc.c
+++ b/net/smc/smc_llc.c
@@ -214,6 +214,31 @@ int smc_llc_send_confirm_link(struct smc_link *link, u8 mac[],
        return rc;
 }
 
+/* send LLC confirm rkey request */
+static int smc_llc_send_confirm_rkey(struct smc_link *link,
+                                    struct smc_buf_desc *rmb_desc)
+{
+       struct smc_llc_msg_confirm_rkey *rkeyllc;
+       struct smc_wr_tx_pend_priv *pend;
+       struct smc_wr_buf *wr_buf;
+       int rc;
+
+       rc = smc_llc_add_pending_send(link, &wr_buf, &pend);
+       if (rc)
+               return rc;
+       rkeyllc = (struct smc_llc_msg_confirm_rkey *)wr_buf;
+       memset(rkeyllc, 0, sizeof(*rkeyllc));
+       rkeyllc->hd.common.type = SMC_LLC_CONFIRM_RKEY;
+       rkeyllc->hd.length = sizeof(struct smc_llc_msg_confirm_rkey);
+       rkeyllc->rtoken[0].rmb_key =
+               htonl(rmb_desc->mr_rx[SMC_SINGLE_LINK]->rkey);
+       rkeyllc->rtoken[0].rmb_vaddr = cpu_to_be64(
+               (u64)sg_dma_address(rmb_desc->sgt[SMC_SINGLE_LINK].sgl));
+       /* send llc message */
+       rc = smc_wr_tx_send(link, pend);
+       return rc;
+}
+
 /* send ADD LINK request or response */
 int smc_llc_send_add_link(struct smc_link *link, u8 mac[],
                          union ib_gid *gid,
@@ -413,7 +438,9 @@ static void smc_llc_rx_confirm_rkey(struct smc_link *link,
        lgr = container_of(link, struct smc_link_group, lnk[SMC_SINGLE_LINK]);
 
        if (llc->hd.flags & SMC_LLC_FLAG_RESP) {
-               /* unused as long as we don't send this type of msg */
+               link->llc_confirm_rkey_rc = llc->hd.flags &
+                                           SMC_LLC_FLAG_RKEY_NEG;
+               complete(&link->llc_confirm_rkey);
        } else {
                rc = smc_rtoken_add(lgr,
                                    llc->rtoken[0].rmb_vaddr,
@@ -503,7 +530,7 @@ static void smc_llc_rx_handler(struct ib_wc *wc, void *buf)
        }
 }
 
-/***************************** worker ****************************************/
+/***************************** worker, utils *********************************/
 
 static void smc_llc_testlink_work(struct work_struct *work)
 {
@@ -562,6 +589,22 @@ void smc_llc_link_flush(struct smc_link *link)
        cancel_delayed_work_sync(&link->llc_testlink_wrk);
 }
 
+/* register a new rtoken at the remote peer */
+int smc_llc_do_confirm_rkey(struct smc_link *link,
+                           struct smc_buf_desc *rmb_desc)
+{
+       int rc;
+
+       reinit_completion(&link->llc_confirm_rkey);
+       smc_llc_send_confirm_rkey(link, rmb_desc);
+       /* receive CONFIRM RKEY response from server over RoCE fabric */
+       rc = wait_for_completion_interruptible_timeout(&link->llc_confirm_rkey,
+                                                      SMC_LLC_WAIT_TIME);
+       if (rc <= 0 || link->llc_confirm_rkey_rc)
+               return -EFAULT;
+       return 0;
+}
+
 /***************************** init, exit, misc ******************************/
 
 static struct smc_wr_rx_handler smc_llc_rx_handlers[] = {

diff --git a/net/smc/smc_llc.h b/net/smc/smc_llc.h
index d6e4211..728823d 100644 (file)
--- a/net/smc/smc_llc.h
+++ b/net/smc/smc_llc.h
@@ -47,6 +47,8 @@ int smc_llc_send_test_link(struct smc_link *lnk, u8 user_data[16],
 void smc_llc_link_active(struct smc_link *link, int testlink_time);
 void smc_llc_link_inactive(struct smc_link *link);
 void smc_llc_link_flush(struct smc_link *link);
+int smc_llc_do_confirm_rkey(struct smc_link *link,
+                           struct smc_buf_desc *rmb_desc);
 int smc_llc_init(void) __init;
 
 #endif /* SMC_LLC_H */