Merge branch 'net-smc-enhancements-2018-05-15'

Ursula Braun says:

====================
net/smc: enhancements 2018/05/15

here are smc patches for net-next. The first one is a fix for net-next
commit 01d2f7e2cdd3 "net/smc: sockopts TCP_NODELAY and TCP_CORK".
Patch 7 improves Connection Layer Control error handling, patch 10
improves abnormal termination of link groups. The remaining patches
from Karsten improve Link Layer Control code.
====================

Signed-off-by: David S. Miller <davem@davemloft.net>

6 files changed, 196 insertions, 65 deletions

diff --git a/net/smc/af_smc.c b/net/smc/af_smc.c
index 17688a02035b..d15762b057c0 100644
--- 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 */
@@ -455,6 +463,8 @@ static int smc_connect_rdma(struct smc_sock *smc)
 			reason_code = SMC_CLC_DECL_MEM;/* insufficient memory*/
 		else if (rc == -ENOLINK)
 			reason_code = SMC_CLC_DECL_SYNCERR; /* synchr. error */
+		else
+			reason_code = SMC_CLC_DECL_INTERR; /* other error */
 		goto decline_rdma_unlock;
 	}
 	link = &smc->conn.lgr->lnk[SMC_SINGLE_LINK];
@@ -488,7 +498,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 +739,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 +876,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;
 			}
@@ -1353,14 +1363,14 @@ static int smc_setsockopt(struct socket *sock, int level, int optname,
 		break;
 	case TCP_NODELAY:
 		if (sk->sk_state != SMC_INIT && sk->sk_state != SMC_LISTEN) {
-			if (val)
+			if (val && !smc->use_fallback)
 				mod_delayed_work(system_wq, &smc->conn.tx_work,
 						 0);
 		}
 		break;
 	case TCP_CORK:
 		if (sk->sk_state != SMC_INIT && sk->sk_state != SMC_LISTEN) {
-			if (!val)
+			if (!val && !smc->use_fallback)
 				mod_delayed_work(system_wq, &smc->conn.tx_work,
 						 0);
 		}
@@ -1634,6 +1644,7 @@ static void __exit smc_exit(void)
 	spin_unlock_bh(&smc_lgr_list.lock);
 	list_for_each_entry_safe(lgr, lg, &lgr_freeing_list, list) {
 		list_del_init(&lgr->list);
+		smc_llc_link_inactive(&lgr->lnk[SMC_SINGLE_LINK]);
 		cancel_delayed_work_sync(&lgr->free_work);
 		smc_lgr_free(lgr); /* free link group */
 	}
diff --git a/net/smc/smc_clc.c b/net/smc/smc_clc.c
index 3a988c22f627..236cb3f12c71 100644
--- a/net/smc/smc_clc.c
+++ b/net/smc/smc_clc.c
@@ -316,7 +316,7 @@ int smc_clc_wait_msg(struct smc_sock *smc, void *buf, int buflen,
 	if (clcm->type == SMC_CLC_DECLINE) {
 		reason_code = SMC_CLC_DECL_REPLY;
 		if (((struct smc_clc_msg_decline *)buf)->hdr.flag) {
-			smc->conn.lgr->sync_err = true;
+			smc->conn.lgr->sync_err = 1;
 			smc_lgr_terminate(smc->conn.lgr);
 		}
 	}
diff --git a/net/smc/smc_core.c b/net/smc/smc_core.c
index 9c74844e2008..08c05cd0bbae 100644
--- a/net/smc/smc_core.c
+++ b/net/smc/smc_core.c
@@ -148,8 +148,11 @@ static void smc_lgr_free_work(struct work_struct *work)
 	list_del_init(&lgr->list); /* remove from smc_lgr_list */
 free:
 	spin_unlock_bh(&smc_lgr_list.lock);
-	if (!delayed_work_pending(&lgr->free_work))
+	if (!delayed_work_pending(&lgr->free_work)) {
+		if (lgr->lnk[SMC_SINGLE_LINK].state != SMC_LNK_INACTIVE)
+			smc_llc_link_inactive(&lgr->lnk[SMC_SINGLE_LINK]);
 		smc_lgr_free(lgr);
+	}
 }
 
 /* create a new SMC link group */
@@ -169,7 +172,7 @@ static int smc_lgr_create(struct smc_sock *smc,
 		goto out;
 	}
 	lgr->role = smc->listen_smc ? SMC_SERV : SMC_CLNT;
-	lgr->sync_err = false;
+	lgr->sync_err = 0;
 	memcpy(lgr->peer_systemid, peer_systemid, SMC_SYSTEMID_LEN);
 	lgr->vlan_id = vlan_id;
 	rwlock_init(&lgr->sndbufs_lock);
@@ -194,9 +197,12 @@ static int smc_lgr_create(struct smc_sock *smc,
 		smc_ib_setup_per_ibdev(smcibdev);
 	get_random_bytes(rndvec, sizeof(rndvec));
 	lnk->psn_initial = rndvec[0] + (rndvec[1] << 8) + (rndvec[2] << 16);
-	rc = smc_wr_alloc_link_mem(lnk);
+	rc = smc_llc_link_init(lnk);
 	if (rc)
 		goto free_lgr;
+	rc = smc_wr_alloc_link_mem(lnk);
+	if (rc)
+		goto clear_llc_lnk;
 	rc = smc_ib_create_protection_domain(lnk);
 	if (rc)
 		goto free_link_mem;
@@ -206,10 +212,6 @@ static int smc_lgr_create(struct smc_sock *smc,
 	rc = smc_wr_create_link(lnk);
 	if (rc)
 		goto destroy_qp;
-	init_completion(&lnk->llc_confirm);
-	init_completion(&lnk->llc_confirm_resp);
-	init_completion(&lnk->llc_add);
-	init_completion(&lnk->llc_add_resp);
 
 	smc->conn.lgr = lgr;
 	rwlock_init(&lgr->conns_lock);
@@ -224,6 +226,8 @@ dealloc_pd:
 	smc_ib_dealloc_protection_domain(lnk);
 free_link_mem:
 	smc_wr_free_link_mem(lnk);
+clear_llc_lnk:
+	smc_llc_link_clear(lnk);
 free_lgr:
 	kfree(lgr);
 out:
@@ -269,6 +273,7 @@ void smc_conn_free(struct smc_connection *conn)
 static void smc_link_clear(struct smc_link *lnk)
 {
 	lnk->peer_qpn = 0;
+	smc_llc_link_clear(lnk);
 	smc_ib_modify_qp_reset(lnk);
 	smc_wr_free_link(lnk);
 	smc_ib_destroy_queue_pair(lnk);
@@ -326,7 +331,6 @@ static void smc_lgr_free_bufs(struct smc_link_group *lgr)
 /* remove a link group */
 void smc_lgr_free(struct smc_link_group *lgr)
 {
-	smc_llc_link_flush(&lgr->lnk[SMC_SINGLE_LINK]);
 	smc_lgr_free_bufs(lgr);
 	smc_link_clear(&lgr->lnk[SMC_SINGLE_LINK]);
 	kfree(lgr);
@@ -348,6 +352,9 @@ void smc_lgr_terminate(struct smc_link_group *lgr)
 	struct smc_sock *smc;
 	struct rb_node *node;
 
+	if (lgr->terminating)
+		return;	/* lgr already terminating */
+	lgr->terminating = 1;
 	smc_lgr_forget(lgr);
 	smc_llc_link_inactive(&lgr->lnk[SMC_SINGLE_LINK]);
 
diff --git a/net/smc/smc_core.h b/net/smc/smc_core.h
index fca8624e5e71..845dc073de13 100644
--- a/net/smc/smc_core.h
+++ b/net/smc/smc_core.h
@@ -96,6 +96,7 @@ struct smc_link {
 	u8			link_id;	/* unique # within link group */
 
 	enum smc_link_state	state;		/* state of link */
+	struct workqueue_struct *llc_wq;	/* single thread work queue */
 	struct completion	llc_confirm;	/* wait for rx of conf link */
 	struct completion	llc_confirm_resp; /* wait 4 rx of cnf lnk rsp */
 	int			llc_confirm_rc; /* rc from confirm link msg */
@@ -105,6 +106,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
@@ -163,7 +166,8 @@ struct smc_link_group {
 
 	u8			id[SMC_LGR_ID_SIZE];	/* unique lgr id */
 	struct delayed_work	free_work;	/* delayed freeing of an lgr */
-	bool			sync_err;	/* lgr no longer fits to peer */
+	u8			sync_err : 1;	/* lgr no longer fits to peer */
+	u8			terminating : 1;/* lgr is terminating */
 };
 
 /* Find the connection associated with the given alert token in the link group.
diff --git a/net/smc/smc_llc.c b/net/smc/smc_llc.c
index 33b4d856f4c6..5800a6b43d83 100644
--- a/net/smc/smc_llc.c
+++ b/net/smc/smc_llc.c
@@ -214,12 +214,11 @@ int smc_llc_send_confirm_link(struct smc_link *link, u8 mac[],
 	return rc;
 }
 
-/* send ADD LINK request or response */
-int smc_llc_send_add_link(struct smc_link *link, u8 mac[],
-			  union ib_gid *gid,
-			  enum smc_llc_reqresp reqresp)
+/* 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_add_link *addllc;
+	struct smc_llc_msg_confirm_rkey *rkeyllc;
 	struct smc_wr_tx_pend_priv *pend;
 	struct smc_wr_buf *wr_buf;
 	int rc;
@@ -227,7 +226,25 @@ int smc_llc_send_add_link(struct smc_link *link, u8 mac[],
 	rc = smc_llc_add_pending_send(link, &wr_buf, &pend);
 	if (rc)
 		return rc;
-	addllc = (struct smc_llc_msg_add_link *)wr_buf;
+	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;
+}
+
+/* prepare an add link message */
+static void smc_llc_prep_add_link(struct smc_llc_msg_add_link *addllc,
+				  struct smc_link *link, u8 mac[],
+				  union ib_gid *gid,
+				  enum smc_llc_reqresp reqresp)
+{
 	memset(addllc, 0, sizeof(*addllc));
 	addllc->hd.common.type = SMC_LLC_ADD_LINK;
 	addllc->hd.length = sizeof(struct smc_llc_msg_add_link);
@@ -239,16 +256,14 @@ int smc_llc_send_add_link(struct smc_link *link, u8 mac[],
 	}
 	memcpy(addllc->sender_mac, mac, ETH_ALEN);
 	memcpy(addllc->sender_gid, gid, SMC_GID_SIZE);
-	/* send llc message */
-	rc = smc_wr_tx_send(link, pend);
-	return rc;
 }
 
-/* send DELETE LINK request or response */
-int smc_llc_send_delete_link(struct smc_link *link,
-			     enum smc_llc_reqresp reqresp)
+/* send ADD LINK request or response */
+int smc_llc_send_add_link(struct smc_link *link, u8 mac[],
+			  union ib_gid *gid,
+			  enum smc_llc_reqresp reqresp)
 {
-	struct smc_llc_msg_del_link *delllc;
+	struct smc_llc_msg_add_link *addllc;
 	struct smc_wr_tx_pend_priv *pend;
 	struct smc_wr_buf *wr_buf;
 	int rc;
@@ -256,7 +271,18 @@ int smc_llc_send_delete_link(struct smc_link *link,
 	rc = smc_llc_add_pending_send(link, &wr_buf, &pend);
 	if (rc)
 		return rc;
-	delllc = (struct smc_llc_msg_del_link *)wr_buf;
+	addllc = (struct smc_llc_msg_add_link *)wr_buf;
+	smc_llc_prep_add_link(addllc, link, mac, gid, reqresp);
+	/* send llc message */
+	rc = smc_wr_tx_send(link, pend);
+	return rc;
+}
+
+/* prepare a delete link message */
+static void smc_llc_prep_delete_link(struct smc_llc_msg_del_link *delllc,
+				     struct smc_link *link,
+				     enum smc_llc_reqresp reqresp)
+{
 	memset(delllc, 0, sizeof(*delllc));
 	delllc->hd.common.type = SMC_LLC_DELETE_LINK;
 	delllc->hd.length = sizeof(struct smc_llc_msg_add_link);
@@ -266,14 +292,29 @@ int smc_llc_send_delete_link(struct smc_link *link,
 	delllc->hd.flags |= SMC_LLC_FLAG_DEL_LINK_ALL;
 	delllc->hd.flags |= SMC_LLC_FLAG_DEL_LINK_ORDERLY;
 	delllc->link_num = link->link_id;
+}
+
+/* send DELETE LINK request or response */
+int smc_llc_send_delete_link(struct smc_link *link,
+			     enum smc_llc_reqresp reqresp)
+{
+	struct smc_llc_msg_del_link *delllc;
+	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;
+	delllc = (struct smc_llc_msg_del_link *)wr_buf;
+	smc_llc_prep_delete_link(delllc, link, reqresp);
 	/* send llc message */
 	rc = smc_wr_tx_send(link, pend);
 	return rc;
 }
 
-/* send LLC test link request or response */
-int smc_llc_send_test_link(struct smc_link *link, u8 user_data[16],
-			   enum smc_llc_reqresp reqresp)
+/* send LLC test link request */
+static int smc_llc_send_test_link(struct smc_link *link, u8 user_data[16])
 {
 	struct smc_llc_msg_test_link *testllc;
 	struct smc_wr_tx_pend_priv *pend;
@@ -287,28 +328,52 @@ int smc_llc_send_test_link(struct smc_link *link, u8 user_data[16],
 	memset(testllc, 0, sizeof(*testllc));
 	testllc->hd.common.type = SMC_LLC_TEST_LINK;
 	testllc->hd.length = sizeof(struct smc_llc_msg_test_link);
-	if (reqresp == SMC_LLC_RESP)
-		testllc->hd.flags |= SMC_LLC_FLAG_RESP;
 	memcpy(testllc->user_data, user_data, sizeof(testllc->user_data));
 	/* send llc message */
 	rc = smc_wr_tx_send(link, pend);
 	return rc;
 }
 
-/* send a prepared message */
-static int smc_llc_send_message(struct smc_link *link, void *llcbuf, int llclen)
+struct smc_llc_send_work {
+	struct work_struct work;
+	struct smc_link *link;
+	int llclen;
+	union smc_llc_msg llcbuf;
+};
+
+/* worker that sends a prepared message */
+static void smc_llc_send_message_work(struct work_struct *work)
 {
+	struct smc_llc_send_work *llcwrk = container_of(work,
+						struct smc_llc_send_work, work);
 	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 (llcwrk->link->state == SMC_LNK_INACTIVE)
+		goto out;
+	rc = smc_llc_add_pending_send(llcwrk->link, &wr_buf, &pend);
 	if (rc)
-		return rc;
-	memcpy(wr_buf, llcbuf, llclen);
-	/* send llc message */
-	rc = smc_wr_tx_send(link, pend);
-	return rc;
+		goto out;
+	memcpy(wr_buf, &llcwrk->llcbuf, llcwrk->llclen);
+	smc_wr_tx_send(llcwrk->link, pend);
+out:
+	kfree(llcwrk);
+}
+
+/* copy llcbuf and schedule an llc send on link */
+static int smc_llc_send_message(struct smc_link *link, void *llcbuf, int llclen)
+{
+	struct smc_llc_send_work *wrk = kmalloc(sizeof(*wrk), GFP_ATOMIC);
+
+	if (!wrk)
+		return -ENOMEM;
+	INIT_WORK(&wrk->work, smc_llc_send_message_work);
+	wrk->link = link;
+	wrk->llclen = llclen;
+	memcpy(&wrk->llcbuf, llcbuf, llclen);
+	queue_work(link->llc_wq, &wrk->work);
+	return 0;
 }
 
 /********************************* receive ***********************************/
@@ -359,17 +424,18 @@ static void smc_llc_rx_add_link(struct smc_link *link,
 		}
 
 		if (lgr->role == SMC_SERV) {
-			smc_llc_send_add_link(link,
+			smc_llc_prep_add_link(llc, link,
 					link->smcibdev->mac[link->ibport - 1],
 					&link->smcibdev->gid[link->ibport - 1],
 					SMC_LLC_REQ);
 
 		} else {
-			smc_llc_send_add_link(link,
+			smc_llc_prep_add_link(llc, link,
 					link->smcibdev->mac[link->ibport - 1],
 					&link->smcibdev->gid[link->ibport - 1],
 					SMC_LLC_RESP);
 		}
+		smc_llc_send_message(link, llc, sizeof(*llc));
 	}
 }
 
@@ -385,9 +451,11 @@ static void smc_llc_rx_delete_link(struct smc_link *link,
 	} else {
 		if (lgr->role == SMC_SERV) {
 			smc_lgr_forget(lgr);
-			smc_llc_send_delete_link(link, SMC_LLC_REQ);
+			smc_llc_prep_delete_link(llc, link, SMC_LLC_REQ);
+			smc_llc_send_message(link, llc, sizeof(*llc));
 		} else {
-			smc_llc_send_delete_link(link, SMC_LLC_RESP);
+			smc_llc_prep_delete_link(llc, link, SMC_LLC_RESP);
+			smc_llc_send_message(link, llc, sizeof(*llc));
 			smc_lgr_terminate(lgr);
 		}
 	}
@@ -400,7 +468,8 @@ static void smc_llc_rx_test_link(struct smc_link *link,
 		if (link->state == SMC_LNK_ACTIVE)
 			complete(&link->llc_testlink_resp);
 	} else {
-		smc_llc_send_test_link(link, llc->user_data, SMC_LLC_RESP);
+		llc->hd.flags |= SMC_LLC_FLAG_RESP;
+		smc_llc_send_message(link, llc, sizeof(*llc));
 	}
 }
 
@@ -413,7 +482,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,
@@ -424,7 +495,7 @@ static void smc_llc_rx_confirm_rkey(struct smc_link *link,
 		llc->hd.flags |= SMC_LLC_FLAG_RESP;
 		if (rc < 0)
 			llc->hd.flags |= SMC_LLC_FLAG_RKEY_NEG;
-		smc_llc_send_message(link, (void *)llc, sizeof(*llc));
+		smc_llc_send_message(link, llc, sizeof(*llc));
 	}
 }
 
@@ -436,7 +507,7 @@ static void smc_llc_rx_confirm_rkey_cont(struct smc_link *link,
 	} else {
 		/* ignore rtokens for other links, we have only one link */
 		llc->hd.flags |= SMC_LLC_FLAG_RESP;
-		smc_llc_send_message(link, (void *)llc, sizeof(*llc));
+		smc_llc_send_message(link, llc, sizeof(*llc));
 	}
 }
 
@@ -464,7 +535,7 @@ static void smc_llc_rx_delete_rkey(struct smc_link *link,
 		}
 
 		llc->hd.flags |= SMC_LLC_FLAG_RESP;
-		smc_llc_send_message(link, (void *)llc, sizeof(*llc));
+		smc_llc_send_message(link, llc, sizeof(*llc));
 	}
 }
 
@@ -477,6 +548,8 @@ static void smc_llc_rx_handler(struct ib_wc *wc, void *buf)
 		return; /* short message */
 	if (llc->raw.hdr.length != sizeof(*llc))
 		return; /* invalid message */
+	if (link->state == SMC_LNK_INACTIVE)
+		return; /* link not active, drop msg */
 
 	switch (llc->raw.hdr.common.type) {
 	case SMC_LLC_TEST_LINK:
@@ -503,7 +576,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)
 {
@@ -524,7 +597,7 @@ static void smc_llc_testlink_work(struct work_struct *work)
 		goto out;
 	}
 	reinit_completion(&link->llc_testlink_resp);
-	smc_llc_send_test_link(link, user_data, SMC_LLC_REQ);
+	smc_llc_send_test_link(link, user_data);
 	/* receive TEST LINK response over RoCE fabric */
 	rc = wait_for_completion_interruptible_timeout(&link->llc_testlink_resp,
 						       SMC_LLC_WAIT_TIME);
@@ -534,18 +607,36 @@ static void smc_llc_testlink_work(struct work_struct *work)
 	}
 	next_interval = link->llc_testlink_time;
 out:
-	schedule_delayed_work(&link->llc_testlink_wrk, next_interval);
+	queue_delayed_work(link->llc_wq, &link->llc_testlink_wrk,
+			   next_interval);
 }
 
-void smc_llc_link_active(struct smc_link *link, int testlink_time)
+int smc_llc_link_init(struct smc_link *link)
 {
+	struct smc_link_group *lgr = container_of(link, struct smc_link_group,
+						  lnk[SMC_SINGLE_LINK]);
+	link->llc_wq = alloc_ordered_workqueue("llc_wq-%x:%x)", WQ_MEM_RECLAIM,
+					       *((u32 *)lgr->id),
+					       link->link_id);
+	if (!link->llc_wq)
+		return -ENOMEM;
+	init_completion(&link->llc_confirm);
+	init_completion(&link->llc_confirm_resp);
+	init_completion(&link->llc_add);
+	init_completion(&link->llc_add_resp);
+	init_completion(&link->llc_confirm_rkey);
 	init_completion(&link->llc_testlink_resp);
 	INIT_DELAYED_WORK(&link->llc_testlink_wrk, smc_llc_testlink_work);
+	return 0;
+}
+
+void smc_llc_link_active(struct smc_link *link, int testlink_time)
+{
 	link->state = SMC_LNK_ACTIVE;
 	if (testlink_time) {
 		link->llc_testlink_time = testlink_time * HZ;
-		schedule_delayed_work(&link->llc_testlink_wrk,
-				      link->llc_testlink_time);
+		queue_delayed_work(link->llc_wq, &link->llc_testlink_wrk,
+				   link->llc_testlink_time);
 	}
 }
 
@@ -557,9 +648,26 @@ void smc_llc_link_inactive(struct smc_link *link)
 }
 
 /* called in worker context */
-void smc_llc_link_flush(struct smc_link *link)
+void smc_llc_link_clear(struct smc_link *link)
+{
+	flush_workqueue(link->llc_wq);
+	destroy_workqueue(link->llc_wq);
+}
+
+/* register a new rtoken at the remote peer */
+int smc_llc_do_confirm_rkey(struct smc_link *link,
+			    struct smc_buf_desc *rmb_desc)
 {
-	cancel_delayed_work_sync(&link->llc_testlink_wrk);
+	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 ******************************/
diff --git a/net/smc/smc_llc.h b/net/smc/smc_llc.h
index d6e42116485e..65c8645e96a1 100644
--- a/net/smc/smc_llc.h
+++ b/net/smc/smc_llc.h
@@ -42,11 +42,12 @@ int smc_llc_send_add_link(struct smc_link *link, u8 mac[], union ib_gid *gid,
 			  enum smc_llc_reqresp reqresp);
 int smc_llc_send_delete_link(struct smc_link *link,
 			     enum smc_llc_reqresp reqresp);
-int smc_llc_send_test_link(struct smc_link *lnk, u8 user_data[16],
-			   enum smc_llc_reqresp reqresp);
+int smc_llc_link_init(struct smc_link *link);
 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);
+void smc_llc_link_clear(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 */