transport_rdma.c

   1// SPDX-License-Identifier: GPL-2.0-or-later
   2/*
   3 *   Copyright (C) 2017, Microsoft Corporation.
   4 *   Copyright (C) 2018, LG Electronics.
   5 *
   6 *   Author(s): Long Li <longli@microsoft.com>,
   7 *		Hyunchul Lee <hyc.lee@gmail.com>
   8 */
   9
  10#define SUBMOD_NAME	"smb_direct"
  11
  12#include <linux/kthread.h>
  13#include <linux/list.h>
  14#include <linux/mempool.h>
  15#include <linux/highmem.h>
  16#include <linux/scatterlist.h>
  17#include <linux/string_choices.h>
  18#include <rdma/ib_verbs.h>
  19#include <rdma/rdma_cm.h>
  20#include <rdma/rw.h>
  21
  22#define __SMBDIRECT_SOCKET_DISCONNECT(__sc) smb_direct_disconnect_rdma_connection(__sc)
  23
  24#include "glob.h"
  25#include "connection.h"
  26#include "smb_common.h"
  27#include "../common/smb2status.h"
  28#include "../common/smbdirect/smbdirect.h"
  29#include "../common/smbdirect/smbdirect_pdu.h"
  30#include "../common/smbdirect/smbdirect_socket.h"
  31#include "transport_rdma.h"
  32
  33#define SMB_DIRECT_PORT_IWARP		5445
  34#define SMB_DIRECT_PORT_INFINIBAND	445
  35
  36#define SMB_DIRECT_VERSION_LE		cpu_to_le16(SMBDIRECT_V1)
  37
  38/* SMB_DIRECT negotiation timeout (for the server) in seconds */
  39#define SMB_DIRECT_NEGOTIATE_TIMEOUT		5
  40
  41/* The timeout to wait for a keepalive message from peer in seconds */
  42#define SMB_DIRECT_KEEPALIVE_SEND_INTERVAL	120
  43
  44/* The timeout to wait for a keepalive message from peer in seconds */
  45#define SMB_DIRECT_KEEPALIVE_RECV_TIMEOUT	5
  46
  47/*
  48 * Default maximum number of RDMA read/write outstanding on this connection
  49 * This value is possibly decreased during QP creation on hardware limit
  50 */
  51#define SMB_DIRECT_CM_INITIATOR_DEPTH		8
  52
  53/* Maximum number of retries on data transfer operations */
  54#define SMB_DIRECT_CM_RETRY			6
  55/* No need to retry on Receiver Not Ready since SMB_DIRECT manages credits */
  56#define SMB_DIRECT_CM_RNR_RETRY		0
  57
  58/*
  59 * User configurable initial values per SMB_DIRECT transport connection
  60 * as defined in [MS-SMBD] 3.1.1.1
  61 * Those may change after a SMB_DIRECT negotiation
  62 */
  63
  64/* Set 445 port to SMB Direct port by default */
  65static int smb_direct_port = SMB_DIRECT_PORT_INFINIBAND;
  66
  67/* The local peer's maximum number of credits to grant to the peer */
  68static int smb_direct_receive_credit_max = 255;
  69
  70/* The remote peer's credit request of local peer */
  71static int smb_direct_send_credit_target = 255;
  72
  73/* The maximum single message size can be sent to remote peer */
  74static int smb_direct_max_send_size = 1364;
  75
  76/*  The maximum fragmented upper-layer payload receive size supported */
  77static int smb_direct_max_fragmented_recv_size = 1024 * 1024;
  78
  79/*  The maximum single-message size which can be received */
  80static int smb_direct_max_receive_size = 1364;
  81
  82static int smb_direct_max_read_write_size = SMBD_DEFAULT_IOSIZE;
  83
  84static LIST_HEAD(smb_direct_device_list);
  85static DEFINE_RWLOCK(smb_direct_device_lock);
  86
  87struct smb_direct_device {
  88	struct ib_device	*ib_dev;
  89	struct list_head	list;
  90};
  91
  92static struct smb_direct_listener {
  93	struct rdma_cm_id	*cm_id;
  94} smb_direct_listener;
  95
  96static struct workqueue_struct *smb_direct_wq;
  97
  98struct smb_direct_transport {
  99	struct ksmbd_transport	transport;
 100
 101	struct smbdirect_socket socket;
 102};
 103
 104#define KSMBD_TRANS(t) (&(t)->transport)
 105#define SMBD_TRANS(t)	(container_of(t, \
 106				struct smb_direct_transport, transport))
 107
 108static const struct ksmbd_transport_ops ksmbd_smb_direct_transport_ops;
 109
 110void init_smbd_max_io_size(unsigned int sz)
 111{
 112	sz = clamp_val(sz, SMBD_MIN_IOSIZE, SMBD_MAX_IOSIZE);
 113	smb_direct_max_read_write_size = sz;
 114}
 115
 116unsigned int get_smbd_max_read_write_size(struct ksmbd_transport *kt)
 117{
 118	struct smb_direct_transport *t;
 119	struct smbdirect_socket *sc;
 120	struct smbdirect_socket_parameters *sp;
 121
 122	if (kt->ops != &ksmbd_smb_direct_transport_ops)
 123		return 0;
 124
 125	t = SMBD_TRANS(kt);
 126	sc = &t->socket;
 127	sp = &sc->parameters;
 128
 129	return sp->max_read_write_size;
 130}
 131
 132static inline int get_buf_page_count(void *buf, int size)
 133{
 134	return DIV_ROUND_UP((uintptr_t)buf + size, PAGE_SIZE) -
 135		(uintptr_t)buf / PAGE_SIZE;
 136}
 137
 138static void smb_direct_destroy_pools(struct smbdirect_socket *sc);
 139static void smb_direct_post_recv_credits(struct work_struct *work);
 140static int smb_direct_post_send_data(struct smbdirect_socket *sc,
 141				     struct smbdirect_send_batch *send_ctx,
 142				     struct kvec *iov, int niov,
 143				     int remaining_data_length);
 144
 145static inline void
 146*smbdirect_recv_io_payload(struct smbdirect_recv_io *recvmsg)
 147{
 148	return (void *)recvmsg->packet;
 149}
 150
 151static struct
 152smbdirect_recv_io *get_free_recvmsg(struct smbdirect_socket *sc)
 153{
 154	struct smbdirect_recv_io *recvmsg = NULL;
 155	unsigned long flags;
 156
 157	spin_lock_irqsave(&sc->recv_io.free.lock, flags);
 158	if (!list_empty(&sc->recv_io.free.list)) {
 159		recvmsg = list_first_entry(&sc->recv_io.free.list,
 160					   struct smbdirect_recv_io,
 161					   list);
 162		list_del(&recvmsg->list);
 163	}
 164	spin_unlock_irqrestore(&sc->recv_io.free.lock, flags);
 165	return recvmsg;
 166}
 167
 168static void put_recvmsg(struct smbdirect_socket *sc,
 169			struct smbdirect_recv_io *recvmsg)
 170{
 171	unsigned long flags;
 172
 173	if (likely(recvmsg->sge.length != 0)) {
 174		ib_dma_unmap_single(sc->ib.dev,
 175				    recvmsg->sge.addr,
 176				    recvmsg->sge.length,
 177				    DMA_FROM_DEVICE);
 178		recvmsg->sge.length = 0;
 179	}
 180
 181	spin_lock_irqsave(&sc->recv_io.free.lock, flags);
 182	list_add(&recvmsg->list, &sc->recv_io.free.list);
 183	spin_unlock_irqrestore(&sc->recv_io.free.lock, flags);
 184
 185	queue_work(sc->workqueue, &sc->recv_io.posted.refill_work);
 186}
 187
 188static void enqueue_reassembly(struct smbdirect_socket *sc,
 189			       struct smbdirect_recv_io *recvmsg,
 190			       int data_length)
 191{
 192	unsigned long flags;
 193
 194	spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
 195	list_add_tail(&recvmsg->list, &sc->recv_io.reassembly.list);
 196	sc->recv_io.reassembly.queue_length++;
 197	/*
 198	 * Make sure reassembly_data_length is updated after list and
 199	 * reassembly_queue_length are updated. On the dequeue side
 200	 * reassembly_data_length is checked without a lock to determine
 201	 * if reassembly_queue_length and list is up to date
 202	 */
 203	virt_wmb();
 204	sc->recv_io.reassembly.data_length += data_length;
 205	spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
 206}
 207
 208static struct smbdirect_recv_io *get_first_reassembly(struct smbdirect_socket *sc)
 209{
 210	if (!list_empty(&sc->recv_io.reassembly.list))
 211		return list_first_entry(&sc->recv_io.reassembly.list,
 212				struct smbdirect_recv_io, list);
 213	else
 214		return NULL;
 215}
 216
 217static void smb_direct_disconnect_wake_up_all(struct smbdirect_socket *sc)
 218{
 219	/*
 220	 * Wake up all waiters in all wait queues
 221	 * in order to notice the broken connection.
 222	 */
 223	wake_up_all(&sc->status_wait);
 224	wake_up_all(&sc->send_io.bcredits.wait_queue);
 225	wake_up_all(&sc->send_io.lcredits.wait_queue);
 226	wake_up_all(&sc->send_io.credits.wait_queue);
 227	wake_up_all(&sc->send_io.pending.zero_wait_queue);
 228	wake_up_all(&sc->recv_io.reassembly.wait_queue);
 229	wake_up_all(&sc->rw_io.credits.wait_queue);
 230}
 231
 232static void smb_direct_disconnect_rdma_work(struct work_struct *work)
 233{
 234	struct smbdirect_socket *sc =
 235		container_of(work, struct smbdirect_socket, disconnect_work);
 236
 237	if (sc->first_error == 0)
 238		sc->first_error = -ECONNABORTED;
 239
 240	/*
 241	 * make sure this and other work is not queued again
 242	 * but here we don't block and avoid
 243	 * disable[_delayed]_work_sync()
 244	 */
 245	disable_work(&sc->disconnect_work);
 246	disable_work(&sc->connect.work);
 247	disable_work(&sc->recv_io.posted.refill_work);
 248	disable_delayed_work(&sc->idle.timer_work);
 249	disable_work(&sc->idle.immediate_work);
 250
 251	switch (sc->status) {
 252	case SMBDIRECT_SOCKET_NEGOTIATE_NEEDED:
 253	case SMBDIRECT_SOCKET_NEGOTIATE_RUNNING:
 254	case SMBDIRECT_SOCKET_NEGOTIATE_FAILED:
 255	case SMBDIRECT_SOCKET_CONNECTED:
 256	case SMBDIRECT_SOCKET_ERROR:
 257		sc->status = SMBDIRECT_SOCKET_DISCONNECTING;
 258		rdma_disconnect(sc->rdma.cm_id);
 259		break;
 260
 261	case SMBDIRECT_SOCKET_CREATED:
 262	case SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED:
 263	case SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING:
 264	case SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED:
 265	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED:
 266	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING:
 267	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED:
 268	case SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED:
 269	case SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING:
 270	case SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED:
 271		/*
 272		 * rdma_accept() never reached
 273		 * RDMA_CM_EVENT_ESTABLISHED
 274		 */
 275		sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
 276		break;
 277
 278	case SMBDIRECT_SOCKET_DISCONNECTING:
 279	case SMBDIRECT_SOCKET_DISCONNECTED:
 280	case SMBDIRECT_SOCKET_DESTROYED:
 281		break;
 282	}
 283
 284	/*
 285	 * Wake up all waiters in all wait queues
 286	 * in order to notice the broken connection.
 287	 */
 288	smb_direct_disconnect_wake_up_all(sc);
 289}
 290
 291static void
 292smb_direct_disconnect_rdma_connection(struct smbdirect_socket *sc)
 293{
 294	if (sc->first_error == 0)
 295		sc->first_error = -ECONNABORTED;
 296
 297	/*
 298	 * make sure other work (than disconnect_work) is
 299	 * not queued again but here we don't block and avoid
 300	 * disable[_delayed]_work_sync()
 301	 */
 302	disable_work(&sc->connect.work);
 303	disable_work(&sc->recv_io.posted.refill_work);
 304	disable_work(&sc->idle.immediate_work);
 305	disable_delayed_work(&sc->idle.timer_work);
 306
 307	switch (sc->status) {
 308	case SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED:
 309	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED:
 310	case SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED:
 311	case SMBDIRECT_SOCKET_NEGOTIATE_FAILED:
 312	case SMBDIRECT_SOCKET_ERROR:
 313	case SMBDIRECT_SOCKET_DISCONNECTING:
 314	case SMBDIRECT_SOCKET_DISCONNECTED:
 315	case SMBDIRECT_SOCKET_DESTROYED:
 316		/*
 317		 * Keep the current error status
 318		 */
 319		break;
 320
 321	case SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED:
 322	case SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING:
 323		sc->status = SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED;
 324		break;
 325
 326	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED:
 327	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING:
 328		sc->status = SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED;
 329		break;
 330
 331	case SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED:
 332	case SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING:
 333		sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED;
 334		break;
 335
 336	case SMBDIRECT_SOCKET_NEGOTIATE_NEEDED:
 337	case SMBDIRECT_SOCKET_NEGOTIATE_RUNNING:
 338		sc->status = SMBDIRECT_SOCKET_NEGOTIATE_FAILED;
 339		break;
 340
 341	case SMBDIRECT_SOCKET_CREATED:
 342		sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
 343		break;
 344
 345	case SMBDIRECT_SOCKET_CONNECTED:
 346		sc->status = SMBDIRECT_SOCKET_ERROR;
 347		break;
 348	}
 349
 350	/*
 351	 * Wake up all waiters in all wait queues
 352	 * in order to notice the broken connection.
 353	 */
 354	smb_direct_disconnect_wake_up_all(sc);
 355
 356	queue_work(sc->workqueue, &sc->disconnect_work);
 357}
 358
 359static void smb_direct_send_immediate_work(struct work_struct *work)
 360{
 361	struct smbdirect_socket *sc =
 362		container_of(work, struct smbdirect_socket, idle.immediate_work);
 363
 364	if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
 365		return;
 366
 367	smb_direct_post_send_data(sc, NULL, NULL, 0, 0);
 368}
 369
 370static void smb_direct_idle_connection_timer(struct work_struct *work)
 371{
 372	struct smbdirect_socket *sc =
 373		container_of(work, struct smbdirect_socket, idle.timer_work.work);
 374	struct smbdirect_socket_parameters *sp = &sc->parameters;
 375
 376	if (sc->idle.keepalive != SMBDIRECT_KEEPALIVE_NONE) {
 377		smb_direct_disconnect_rdma_connection(sc);
 378		return;
 379	}
 380
 381	if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
 382		return;
 383
 384	/*
 385	 * Now use the keepalive timeout (instead of keepalive interval)
 386	 * in order to wait for a response
 387	 */
 388	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_PENDING;
 389	mod_delayed_work(sc->workqueue, &sc->idle.timer_work,
 390			 msecs_to_jiffies(sp->keepalive_timeout_msec));
 391	queue_work(sc->workqueue, &sc->idle.immediate_work);
 392}
 393
 394static struct smb_direct_transport *alloc_transport(struct rdma_cm_id *cm_id)
 395{
 396	struct smb_direct_transport *t;
 397	struct smbdirect_socket *sc;
 398	struct smbdirect_socket_parameters *sp;
 399	struct ksmbd_conn *conn;
 400
 401	t = kzalloc(sizeof(*t), KSMBD_DEFAULT_GFP);
 402	if (!t)
 403		return NULL;
 404	sc = &t->socket;
 405	smbdirect_socket_init(sc);
 406	sp = &sc->parameters;
 407
 408	sc->workqueue = smb_direct_wq;
 409
 410	INIT_WORK(&sc->disconnect_work, smb_direct_disconnect_rdma_work);
 411
 412	sp->negotiate_timeout_msec = SMB_DIRECT_NEGOTIATE_TIMEOUT * 1000;
 413	sp->initiator_depth = SMB_DIRECT_CM_INITIATOR_DEPTH;
 414	sp->responder_resources = 1;
 415	sp->recv_credit_max = smb_direct_receive_credit_max;
 416	sp->send_credit_target = smb_direct_send_credit_target;
 417	sp->max_send_size = smb_direct_max_send_size;
 418	sp->max_fragmented_recv_size = smb_direct_max_fragmented_recv_size;
 419	sp->max_recv_size = smb_direct_max_receive_size;
 420	sp->max_read_write_size = smb_direct_max_read_write_size;
 421	sp->keepalive_interval_msec = SMB_DIRECT_KEEPALIVE_SEND_INTERVAL * 1000;
 422	sp->keepalive_timeout_msec = SMB_DIRECT_KEEPALIVE_RECV_TIMEOUT * 1000;
 423
 424	sc->rdma.cm_id = cm_id;
 425	cm_id->context = sc;
 426
 427	sc->ib.dev = sc->rdma.cm_id->device;
 428
 429	INIT_DELAYED_WORK(&sc->idle.timer_work, smb_direct_idle_connection_timer);
 430
 431	conn = ksmbd_conn_alloc();
 432	if (!conn)
 433		goto err;
 434
 435	down_write(&conn_list_lock);
 436	hash_add(conn_list, &conn->hlist, 0);
 437	up_write(&conn_list_lock);
 438
 439	conn->transport = KSMBD_TRANS(t);
 440	KSMBD_TRANS(t)->conn = conn;
 441	KSMBD_TRANS(t)->ops = &ksmbd_smb_direct_transport_ops;
 442	return t;
 443err:
 444	kfree(t);
 445	return NULL;
 446}
 447
 448static void smb_direct_free_transport(struct ksmbd_transport *kt)
 449{
 450	kfree(SMBD_TRANS(kt));
 451}
 452
 453static void free_transport(struct smb_direct_transport *t)
 454{
 455	struct smbdirect_socket *sc = &t->socket;
 456	struct smbdirect_recv_io *recvmsg;
 457
 458	disable_work_sync(&sc->disconnect_work);
 459	if (sc->status < SMBDIRECT_SOCKET_DISCONNECTING)
 460		smb_direct_disconnect_rdma_work(&sc->disconnect_work);
 461	if (sc->status < SMBDIRECT_SOCKET_DISCONNECTED)
 462		wait_event(sc->status_wait, sc->status == SMBDIRECT_SOCKET_DISCONNECTED);
 463
 464	/*
 465	 * Wake up all waiters in all wait queues
 466	 * in order to notice the broken connection.
 467	 *
 468	 * Most likely this was already called via
 469	 * smb_direct_disconnect_rdma_work(), but call it again...
 470	 */
 471	smb_direct_disconnect_wake_up_all(sc);
 472
 473	disable_work_sync(&sc->connect.work);
 474	disable_work_sync(&sc->recv_io.posted.refill_work);
 475	disable_delayed_work_sync(&sc->idle.timer_work);
 476	disable_work_sync(&sc->idle.immediate_work);
 477
 478	if (sc->rdma.cm_id)
 479		rdma_lock_handler(sc->rdma.cm_id);
 480
 481	if (sc->ib.qp) {
 482		ib_drain_qp(sc->ib.qp);
 483		sc->ib.qp = NULL;
 484		rdma_destroy_qp(sc->rdma.cm_id);
 485	}
 486
 487	ksmbd_debug(RDMA, "drain the reassembly queue\n");
 488	do {
 489		unsigned long flags;
 490
 491		spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
 492		recvmsg = get_first_reassembly(sc);
 493		if (recvmsg) {
 494			list_del(&recvmsg->list);
 495			spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
 496			put_recvmsg(sc, recvmsg);
 497		} else {
 498			spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
 499		}
 500	} while (recvmsg);
 501	sc->recv_io.reassembly.data_length = 0;
 502
 503	if (sc->ib.send_cq)
 504		ib_free_cq(sc->ib.send_cq);
 505	if (sc->ib.recv_cq)
 506		ib_free_cq(sc->ib.recv_cq);
 507	if (sc->ib.pd)
 508		ib_dealloc_pd(sc->ib.pd);
 509	if (sc->rdma.cm_id) {
 510		rdma_unlock_handler(sc->rdma.cm_id);
 511		rdma_destroy_id(sc->rdma.cm_id);
 512	}
 513
 514	smb_direct_destroy_pools(sc);
 515	ksmbd_conn_free(KSMBD_TRANS(t)->conn);
 516}
 517
 518static struct smbdirect_send_io
 519*smb_direct_alloc_sendmsg(struct smbdirect_socket *sc)
 520{
 521	struct smbdirect_send_io *msg;
 522
 523	msg = mempool_alloc(sc->send_io.mem.pool, KSMBD_DEFAULT_GFP);
 524	if (!msg)
 525		return ERR_PTR(-ENOMEM);
 526	msg->socket = sc;
 527	INIT_LIST_HEAD(&msg->sibling_list);
 528	msg->num_sge = 0;
 529	return msg;
 530}
 531
 532static void smb_direct_free_sendmsg(struct smbdirect_socket *sc,
 533				    struct smbdirect_send_io *msg)
 534{
 535	int i;
 536
 537	/*
 538	 * The list needs to be empty!
 539	 * The caller should take care of it.
 540	 */
 541	WARN_ON_ONCE(!list_empty(&msg->sibling_list));
 542
 543	if (msg->num_sge > 0) {
 544		ib_dma_unmap_single(sc->ib.dev,
 545				    msg->sge[0].addr, msg->sge[0].length,
 546				    DMA_TO_DEVICE);
 547		for (i = 1; i < msg->num_sge; i++)
 548			ib_dma_unmap_page(sc->ib.dev,
 549					  msg->sge[i].addr, msg->sge[i].length,
 550					  DMA_TO_DEVICE);
 551	}
 552	mempool_free(msg, sc->send_io.mem.pool);
 553}
 554
 555static int smb_direct_check_recvmsg(struct smbdirect_recv_io *recvmsg)
 556{
 557	struct smbdirect_socket *sc = recvmsg->socket;
 558
 559	switch (sc->recv_io.expected) {
 560	case SMBDIRECT_EXPECT_DATA_TRANSFER: {
 561		struct smbdirect_data_transfer *req =
 562			(struct smbdirect_data_transfer *)recvmsg->packet;
 563		struct smb2_hdr *hdr = (struct smb2_hdr *)(recvmsg->packet
 564				+ le32_to_cpu(req->data_offset));
 565		ksmbd_debug(RDMA,
 566			    "CreditGranted: %u, CreditRequested: %u, DataLength: %u, RemainingDataLength: %u, SMB: %x, Command: %u\n",
 567			    le16_to_cpu(req->credits_granted),
 568			    le16_to_cpu(req->credits_requested),
 569			    req->data_length, req->remaining_data_length,
 570			    hdr->ProtocolId, hdr->Command);
 571		return 0;
 572	}
 573	case SMBDIRECT_EXPECT_NEGOTIATE_REQ: {
 574		struct smbdirect_negotiate_req *req =
 575			(struct smbdirect_negotiate_req *)recvmsg->packet;
 576		ksmbd_debug(RDMA,
 577			    "MinVersion: %u, MaxVersion: %u, CreditRequested: %u, MaxSendSize: %u, MaxRecvSize: %u, MaxFragmentedSize: %u\n",
 578			    le16_to_cpu(req->min_version),
 579			    le16_to_cpu(req->max_version),
 580			    le16_to_cpu(req->credits_requested),
 581			    le32_to_cpu(req->preferred_send_size),
 582			    le32_to_cpu(req->max_receive_size),
 583			    le32_to_cpu(req->max_fragmented_size));
 584		if (le16_to_cpu(req->min_version) > 0x0100 ||
 585		    le16_to_cpu(req->max_version) < 0x0100)
 586			return -EOPNOTSUPP;
 587		if (le16_to_cpu(req->credits_requested) <= 0 ||
 588		    le32_to_cpu(req->max_receive_size) <= 128 ||
 589		    le32_to_cpu(req->max_fragmented_size) <=
 590					128 * 1024)
 591			return -ECONNABORTED;
 592
 593		return 0;
 594	}
 595	case SMBDIRECT_EXPECT_NEGOTIATE_REP:
 596		/* client only */
 597		break;
 598	}
 599
 600	/* This is an internal error */
 601	return -EINVAL;
 602}
 603
 604static void recv_done(struct ib_cq *cq, struct ib_wc *wc)
 605{
 606	struct smbdirect_recv_io *recvmsg;
 607	struct smbdirect_socket *sc;
 608	struct smbdirect_socket_parameters *sp;
 609
 610	recvmsg = container_of(wc->wr_cqe, struct smbdirect_recv_io, cqe);
 611	sc = recvmsg->socket;
 612	sp = &sc->parameters;
 613
 614	if (wc->status != IB_WC_SUCCESS || wc->opcode != IB_WC_RECV) {
 615		put_recvmsg(sc, recvmsg);
 616		if (wc->status != IB_WC_WR_FLUSH_ERR) {
 617			pr_err("Recv error. status='%s (%d)' opcode=%d\n",
 618			       ib_wc_status_msg(wc->status), wc->status,
 619			       wc->opcode);
 620			smb_direct_disconnect_rdma_connection(sc);
 621		}
 622		return;
 623	}
 624
 625	ksmbd_debug(RDMA, "Recv completed. status='%s (%d)', opcode=%d\n",
 626		    ib_wc_status_msg(wc->status), wc->status,
 627		    wc->opcode);
 628
 629	ib_dma_sync_single_for_cpu(wc->qp->device, recvmsg->sge.addr,
 630				   recvmsg->sge.length, DMA_FROM_DEVICE);
 631
 632	/*
 633	 * Reset timer to the keepalive interval in
 634	 * order to trigger our next keepalive message.
 635	 */
 636	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_NONE;
 637	mod_delayed_work(sc->workqueue, &sc->idle.timer_work,
 638			 msecs_to_jiffies(sp->keepalive_interval_msec));
 639
 640	switch (sc->recv_io.expected) {
 641	case SMBDIRECT_EXPECT_NEGOTIATE_REQ:
 642		/* see smb_direct_negotiate_recv_done */
 643		break;
 644	case SMBDIRECT_EXPECT_DATA_TRANSFER: {
 645		struct smbdirect_data_transfer *data_transfer =
 646			(struct smbdirect_data_transfer *)recvmsg->packet;
 647		u32 remaining_data_length, data_offset, data_length;
 648		int current_recv_credits;
 649		u16 old_recv_credit_target;
 650
 651		if (wc->byte_len <
 652		    offsetof(struct smbdirect_data_transfer, padding)) {
 653			put_recvmsg(sc, recvmsg);
 654			smb_direct_disconnect_rdma_connection(sc);
 655			return;
 656		}
 657
 658		remaining_data_length = le32_to_cpu(data_transfer->remaining_data_length);
 659		data_length = le32_to_cpu(data_transfer->data_length);
 660		data_offset = le32_to_cpu(data_transfer->data_offset);
 661		if (wc->byte_len < data_offset ||
 662		    wc->byte_len < (u64)data_offset + data_length) {
 663			put_recvmsg(sc, recvmsg);
 664			smb_direct_disconnect_rdma_connection(sc);
 665			return;
 666		}
 667		if (remaining_data_length > sp->max_fragmented_recv_size ||
 668		    data_length > sp->max_fragmented_recv_size ||
 669		    (u64)remaining_data_length + (u64)data_length >
 670		    (u64)sp->max_fragmented_recv_size) {
 671			put_recvmsg(sc, recvmsg);
 672			smb_direct_disconnect_rdma_connection(sc);
 673			return;
 674		}
 675
 676		if (data_length) {
 677			if (sc->recv_io.reassembly.full_packet_received)
 678				recvmsg->first_segment = true;
 679
 680			if (le32_to_cpu(data_transfer->remaining_data_length))
 681				sc->recv_io.reassembly.full_packet_received = false;
 682			else
 683				sc->recv_io.reassembly.full_packet_received = true;
 684		}
 685
 686		atomic_dec(&sc->recv_io.posted.count);
 687		current_recv_credits = atomic_dec_return(&sc->recv_io.credits.count);
 688
 689		old_recv_credit_target = sc->recv_io.credits.target;
 690		sc->recv_io.credits.target =
 691				le16_to_cpu(data_transfer->credits_requested);
 692		sc->recv_io.credits.target =
 693			min_t(u16, sc->recv_io.credits.target, sp->recv_credit_max);
 694		sc->recv_io.credits.target =
 695			max_t(u16, sc->recv_io.credits.target, 1);
 696		atomic_add(le16_to_cpu(data_transfer->credits_granted),
 697			   &sc->send_io.credits.count);
 698
 699		if (le16_to_cpu(data_transfer->flags) &
 700		    SMBDIRECT_FLAG_RESPONSE_REQUESTED)
 701			queue_work(sc->workqueue, &sc->idle.immediate_work);
 702
 703		if (atomic_read(&sc->send_io.credits.count) > 0)
 704			wake_up(&sc->send_io.credits.wait_queue);
 705
 706		if (data_length) {
 707			if (current_recv_credits <= (sc->recv_io.credits.target / 4) ||
 708			    sc->recv_io.credits.target > old_recv_credit_target)
 709				queue_work(sc->workqueue, &sc->recv_io.posted.refill_work);
 710
 711			enqueue_reassembly(sc, recvmsg, (int)data_length);
 712			wake_up(&sc->recv_io.reassembly.wait_queue);
 713		} else
 714			put_recvmsg(sc, recvmsg);
 715
 716		return;
 717	}
 718	case SMBDIRECT_EXPECT_NEGOTIATE_REP:
 719		/* client only */
 720		break;
 721	}
 722
 723	/*
 724	 * This is an internal error!
 725	 */
 726	WARN_ON_ONCE(sc->recv_io.expected != SMBDIRECT_EXPECT_DATA_TRANSFER);
 727	put_recvmsg(sc, recvmsg);
 728	smb_direct_disconnect_rdma_connection(sc);
 729}
 730
 731static void smb_direct_negotiate_recv_work(struct work_struct *work);
 732
 733static void smb_direct_negotiate_recv_done(struct ib_cq *cq, struct ib_wc *wc)
 734{
 735	struct smbdirect_recv_io *recv_io =
 736		container_of(wc->wr_cqe, struct smbdirect_recv_io, cqe);
 737	struct smbdirect_socket *sc = recv_io->socket;
 738	unsigned long flags;
 739
 740	/*
 741	 * reset the common recv_done for later reuse.
 742	 */
 743	recv_io->cqe.done = recv_done;
 744
 745	if (wc->status != IB_WC_SUCCESS || wc->opcode != IB_WC_RECV) {
 746		put_recvmsg(sc, recv_io);
 747		if (wc->status != IB_WC_WR_FLUSH_ERR) {
 748			pr_err("Negotiate Recv error. status='%s (%d)' opcode=%d\n",
 749			       ib_wc_status_msg(wc->status), wc->status,
 750			       wc->opcode);
 751			smb_direct_disconnect_rdma_connection(sc);
 752		}
 753		return;
 754	}
 755
 756	ksmbd_debug(RDMA, "Negotiate Recv completed. status='%s (%d)', opcode=%d\n",
 757		    ib_wc_status_msg(wc->status), wc->status,
 758		    wc->opcode);
 759
 760	ib_dma_sync_single_for_cpu(sc->ib.dev,
 761				   recv_io->sge.addr,
 762				   recv_io->sge.length,
 763				   DMA_FROM_DEVICE);
 764
 765	/*
 766	 * This is an internal error!
 767	 */
 768	if (WARN_ON_ONCE(sc->recv_io.expected != SMBDIRECT_EXPECT_NEGOTIATE_REQ)) {
 769		put_recvmsg(sc, recv_io);
 770		smb_direct_disconnect_rdma_connection(sc);
 771		return;
 772	}
 773
 774	/*
 775	 * Don't reset timer to the keepalive interval in
 776	 * this will be done in smb_direct_negotiate_recv_work.
 777	 */
 778
 779	/*
 780	 * Only remember the recv_io if it has enough bytes,
 781	 * this gives smb_direct_negotiate_recv_work enough
 782	 * information in order to disconnect if it was not
 783	 * valid.
 784	 */
 785	sc->recv_io.reassembly.full_packet_received = true;
 786	if (wc->byte_len >= sizeof(struct smbdirect_negotiate_req))
 787		enqueue_reassembly(sc, recv_io, 0);
 788	else
 789		put_recvmsg(sc, recv_io);
 790
 791	/*
 792	 * Some drivers (at least mlx5_ib and irdma in roce mode)
 793	 * might post a recv completion before RDMA_CM_EVENT_ESTABLISHED,
 794	 * we need to adjust our expectation in that case.
 795	 *
 796	 * So we defer further processing of the negotiation
 797	 * to smb_direct_negotiate_recv_work().
 798	 *
 799	 * If we are already in SMBDIRECT_SOCKET_NEGOTIATE_NEEDED
 800	 * we queue the work directly otherwise
 801	 * smb_direct_cm_handler() will do it, when
 802	 * RDMA_CM_EVENT_ESTABLISHED arrived.
 803	 */
 804	spin_lock_irqsave(&sc->connect.lock, flags);
 805	if (!sc->first_error) {
 806		INIT_WORK(&sc->connect.work, smb_direct_negotiate_recv_work);
 807		if (sc->status == SMBDIRECT_SOCKET_NEGOTIATE_NEEDED)
 808			queue_work(sc->workqueue, &sc->connect.work);
 809	}
 810	spin_unlock_irqrestore(&sc->connect.lock, flags);
 811}
 812
 813static void smb_direct_negotiate_recv_work(struct work_struct *work)
 814{
 815	struct smbdirect_socket *sc =
 816		container_of(work, struct smbdirect_socket, connect.work);
 817	const struct smbdirect_socket_parameters *sp = &sc->parameters;
 818	struct smbdirect_recv_io *recv_io;
 819
 820	if (sc->first_error)
 821		return;
 822
 823	ksmbd_debug(RDMA, "Negotiate Recv Work running\n");
 824
 825	/*
 826	 * Reset timer to the keepalive interval in
 827	 * order to trigger our next keepalive message.
 828	 */
 829	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_NONE;
 830	mod_delayed_work(sc->workqueue, &sc->idle.timer_work,
 831			 msecs_to_jiffies(sp->keepalive_interval_msec));
 832
 833	/*
 834	 * If smb_direct_negotiate_recv_done() detected an
 835	 * invalid request we want to disconnect.
 836	 */
 837	recv_io = get_first_reassembly(sc);
 838	if (!recv_io) {
 839		smb_direct_disconnect_rdma_connection(sc);
 840		return;
 841	}
 842
 843	if (SMBDIRECT_CHECK_STATUS_WARN(sc, SMBDIRECT_SOCKET_NEGOTIATE_NEEDED)) {
 844		smb_direct_disconnect_rdma_connection(sc);
 845		return;
 846	}
 847	sc->status = SMBDIRECT_SOCKET_NEGOTIATE_RUNNING;
 848	wake_up(&sc->status_wait);
 849}
 850
 851static int smb_direct_post_recv(struct smbdirect_socket *sc,
 852				struct smbdirect_recv_io *recvmsg)
 853{
 854	struct smbdirect_socket_parameters *sp = &sc->parameters;
 855	struct ib_recv_wr wr;
 856	int ret;
 857
 858	recvmsg->sge.addr = ib_dma_map_single(sc->ib.dev,
 859					      recvmsg->packet,
 860					      sp->max_recv_size,
 861					      DMA_FROM_DEVICE);
 862	ret = ib_dma_mapping_error(sc->ib.dev, recvmsg->sge.addr);
 863	if (ret)
 864		return ret;
 865	recvmsg->sge.length = sp->max_recv_size;
 866	recvmsg->sge.lkey = sc->ib.pd->local_dma_lkey;
 867
 868	wr.wr_cqe = &recvmsg->cqe;
 869	wr.next = NULL;
 870	wr.sg_list = &recvmsg->sge;
 871	wr.num_sge = 1;
 872
 873	ret = ib_post_recv(sc->ib.qp, &wr, NULL);
 874	if (ret) {
 875		pr_err("Can't post recv: %d\n", ret);
 876		ib_dma_unmap_single(sc->ib.dev,
 877				    recvmsg->sge.addr, recvmsg->sge.length,
 878				    DMA_FROM_DEVICE);
 879		recvmsg->sge.length = 0;
 880		smb_direct_disconnect_rdma_connection(sc);
 881		return ret;
 882	}
 883	return ret;
 884}
 885
 886static int smb_direct_read(struct ksmbd_transport *t, char *buf,
 887			   unsigned int size, int unused)
 888{
 889	struct smbdirect_recv_io *recvmsg;
 890	struct smbdirect_data_transfer *data_transfer;
 891	int to_copy, to_read, data_read, offset;
 892	u32 data_length, remaining_data_length, data_offset;
 893	int rc;
 894	struct smb_direct_transport *st = SMBD_TRANS(t);
 895	struct smbdirect_socket *sc = &st->socket;
 896
 897again:
 898	if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
 899		pr_err("disconnected\n");
 900		return -ENOTCONN;
 901	}
 902
 903	/*
 904	 * No need to hold the reassembly queue lock all the time as we are
 905	 * the only one reading from the front of the queue. The transport
 906	 * may add more entries to the back of the queue at the same time
 907	 */
 908	if (sc->recv_io.reassembly.data_length >= size) {
 909		int queue_length;
 910		int queue_removed = 0;
 911		unsigned long flags;
 912
 913		/*
 914		 * Need to make sure reassembly_data_length is read before
 915		 * reading reassembly_queue_length and calling
 916		 * get_first_reassembly. This call is lock free
 917		 * as we never read at the end of the queue which are being
 918		 * updated in SOFTIRQ as more data is received
 919		 */
 920		virt_rmb();
 921		queue_length = sc->recv_io.reassembly.queue_length;
 922		data_read = 0;
 923		to_read = size;
 924		offset = sc->recv_io.reassembly.first_entry_offset;
 925		while (data_read < size) {
 926			recvmsg = get_first_reassembly(sc);
 927			data_transfer = smbdirect_recv_io_payload(recvmsg);
 928			data_length = le32_to_cpu(data_transfer->data_length);
 929			remaining_data_length =
 930				le32_to_cpu(data_transfer->remaining_data_length);
 931			data_offset = le32_to_cpu(data_transfer->data_offset);
 932
 933			/*
 934			 * The upper layer expects RFC1002 length at the
 935			 * beginning of the payload. Return it to indicate
 936			 * the total length of the packet. This minimize the
 937			 * change to upper layer packet processing logic. This
 938			 * will be eventually remove when an intermediate
 939			 * transport layer is added
 940			 */
 941			if (recvmsg->first_segment && size == 4) {
 942				unsigned int rfc1002_len =
 943					data_length + remaining_data_length;
 944				*((__be32 *)buf) = cpu_to_be32(rfc1002_len);
 945				data_read = 4;
 946				recvmsg->first_segment = false;
 947				ksmbd_debug(RDMA,
 948					    "returning rfc1002 length %d\n",
 949					    rfc1002_len);
 950				goto read_rfc1002_done;
 951			}
 952
 953			to_copy = min_t(int, data_length - offset, to_read);
 954			memcpy(buf + data_read, (char *)data_transfer + data_offset + offset,
 955			       to_copy);
 956
 957			/* move on to the next buffer? */
 958			if (to_copy == data_length - offset) {
 959				queue_length--;
 960				/*
 961				 * No need to lock if we are not at the
 962				 * end of the queue
 963				 */
 964				if (queue_length) {
 965					list_del(&recvmsg->list);
 966				} else {
 967					spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
 968					list_del(&recvmsg->list);
 969					spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
 970				}
 971				queue_removed++;
 972				put_recvmsg(sc, recvmsg);
 973				offset = 0;
 974			} else {
 975				offset += to_copy;
 976			}
 977
 978			to_read -= to_copy;
 979			data_read += to_copy;
 980		}
 981
 982		spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
 983		sc->recv_io.reassembly.data_length -= data_read;
 984		sc->recv_io.reassembly.queue_length -= queue_removed;
 985		spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
 986
 987		sc->recv_io.reassembly.first_entry_offset = offset;
 988		ksmbd_debug(RDMA,
 989			    "returning to thread data_read=%d reassembly_data_length=%d first_entry_offset=%d\n",
 990			    data_read, sc->recv_io.reassembly.data_length,
 991			    sc->recv_io.reassembly.first_entry_offset);
 992read_rfc1002_done:
 993		return data_read;
 994	}
 995
 996	ksmbd_debug(RDMA, "wait_event on more data\n");
 997	rc = wait_event_interruptible(sc->recv_io.reassembly.wait_queue,
 998				      sc->recv_io.reassembly.data_length >= size ||
 999				       sc->status != SMBDIRECT_SOCKET_CONNECTED);
1000	if (rc)
1001		return -EINTR;
1002
1003	goto again;
1004}
1005
1006static void smb_direct_post_recv_credits(struct work_struct *work)
1007{
1008	struct smbdirect_socket *sc =
1009		container_of(work, struct smbdirect_socket, recv_io.posted.refill_work);
1010	struct smbdirect_recv_io *recvmsg;
1011	int credits = 0;
1012	int ret;
1013
1014	if (atomic_read(&sc->recv_io.credits.count) < sc->recv_io.credits.target) {
1015		while (true) {
1016			recvmsg = get_free_recvmsg(sc);
1017			if (!recvmsg)
1018				break;
1019
1020			recvmsg->first_segment = false;
1021
1022			ret = smb_direct_post_recv(sc, recvmsg);
1023			if (ret) {
1024				pr_err("Can't post recv: %d\n", ret);
1025				put_recvmsg(sc, recvmsg);
1026				break;
1027			}
1028			credits++;
1029
1030			atomic_inc(&sc->recv_io.posted.count);
1031		}
1032	}
1033
1034	atomic_add(credits, &sc->recv_io.credits.available);
1035
1036	/*
1037	 * If the last send credit is waiting for credits
1038	 * it can grant we need to wake it up
1039	 */
1040	if (credits &&
1041	    atomic_read(&sc->send_io.bcredits.count) == 0 &&
1042	    atomic_read(&sc->send_io.credits.count) == 0)
1043		wake_up(&sc->send_io.credits.wait_queue);
1044
1045	if (credits)
1046		queue_work(sc->workqueue, &sc->idle.immediate_work);
1047}
1048
1049static void send_done(struct ib_cq *cq, struct ib_wc *wc)
1050{
1051	struct smbdirect_send_io *sendmsg, *sibling, *next;
1052	struct smbdirect_socket *sc;
1053	int lcredits = 0;
1054
1055	sendmsg = container_of(wc->wr_cqe, struct smbdirect_send_io, cqe);
1056	sc = sendmsg->socket;
1057
1058	ksmbd_debug(RDMA, "Send completed. status='%s (%d)', opcode=%d\n",
1059		    ib_wc_status_msg(wc->status), wc->status,
1060		    wc->opcode);
1061
1062	if (unlikely(!(sendmsg->wr.send_flags & IB_SEND_SIGNALED))) {
1063		/*
1064		 * This happens when smbdirect_send_io is a sibling
1065		 * before the final message, it is signaled on
1066		 * error anyway, so we need to skip
1067		 * smbdirect_connection_free_send_io here,
1068		 * otherwise is will destroy the memory
1069		 * of the siblings too, which will cause
1070		 * use after free problems for the others
1071		 * triggered from ib_drain_qp().
1072		 */
1073		if (wc->status != IB_WC_SUCCESS)
1074			goto skip_free;
1075
1076		/*
1077		 * This should not happen!
1078		 * But we better just close the
1079		 * connection...
1080		 */
1081		pr_err("unexpected send completion wc->status=%s (%d) wc->opcode=%d\n",
1082		       ib_wc_status_msg(wc->status), wc->status, wc->opcode);
1083		smb_direct_disconnect_rdma_connection(sc);
1084		return;
1085	}
1086
1087	/*
1088	 * Free possible siblings and then the main send_io
1089	 */
1090	list_for_each_entry_safe(sibling, next, &sendmsg->sibling_list, sibling_list) {
1091		list_del_init(&sibling->sibling_list);
1092		smb_direct_free_sendmsg(sc, sibling);
1093		lcredits += 1;
1094	}
1095	/* Note this frees wc->wr_cqe, but not wc */
1096	smb_direct_free_sendmsg(sc, sendmsg);
1097	lcredits += 1;
1098
1099	if (wc->status != IB_WC_SUCCESS || wc->opcode != IB_WC_SEND) {
1100skip_free:
1101		pr_err("Send error. status='%s (%d)', opcode=%d\n",
1102		       ib_wc_status_msg(wc->status), wc->status,
1103		       wc->opcode);
1104		smb_direct_disconnect_rdma_connection(sc);
1105		return;
1106	}
1107
1108	atomic_add(lcredits, &sc->send_io.lcredits.count);
1109	wake_up(&sc->send_io.lcredits.wait_queue);
1110
1111	if (atomic_dec_and_test(&sc->send_io.pending.count))
1112		wake_up(&sc->send_io.pending.zero_wait_queue);
1113}
1114
1115static int manage_credits_prior_sending(struct smbdirect_socket *sc)
1116{
1117	int missing;
1118	int available;
1119	int new_credits;
1120
1121	if (atomic_read(&sc->recv_io.credits.count) >= sc->recv_io.credits.target)
1122		return 0;
1123
1124	missing = (int)sc->recv_io.credits.target - atomic_read(&sc->recv_io.credits.count);
1125	available = atomic_xchg(&sc->recv_io.credits.available, 0);
1126	new_credits = (u16)min3(U16_MAX, missing, available);
1127	if (new_credits <= 0) {
1128		/*
1129		 * If credits are available, but not granted
1130		 * we need to re-add them again.
1131		 */
1132		if (available)
1133			atomic_add(available, &sc->recv_io.credits.available);
1134		return 0;
1135	}
1136
1137	if (new_credits < available) {
1138		/*
1139		 * Readd the remaining available again.
1140		 */
1141		available -= new_credits;
1142		atomic_add(available, &sc->recv_io.credits.available);
1143	}
1144
1145	/*
1146	 * Remember we granted the credits
1147	 */
1148	atomic_add(new_credits, &sc->recv_io.credits.count);
1149	return new_credits;
1150}
1151
1152static int manage_keep_alive_before_sending(struct smbdirect_socket *sc)
1153{
1154	struct smbdirect_socket_parameters *sp = &sc->parameters;
1155
1156	if (sc->idle.keepalive == SMBDIRECT_KEEPALIVE_PENDING) {
1157		sc->idle.keepalive = SMBDIRECT_KEEPALIVE_SENT;
1158		/*
1159		 * Now use the keepalive timeout (instead of keepalive interval)
1160		 * in order to wait for a response
1161		 */
1162		mod_delayed_work(sc->workqueue, &sc->idle.timer_work,
1163				 msecs_to_jiffies(sp->keepalive_timeout_msec));
1164		return 1;
1165	}
1166	return 0;
1167}
1168
1169static int smb_direct_post_send(struct smbdirect_socket *sc,
1170				struct ib_send_wr *wr)
1171{
1172	int ret;
1173
1174	atomic_inc(&sc->send_io.pending.count);
1175	ret = ib_post_send(sc->ib.qp, wr, NULL);
1176	if (ret) {
1177		pr_err("failed to post send: %d\n", ret);
1178		smb_direct_disconnect_rdma_connection(sc);
1179	}
1180	return ret;
1181}
1182
1183static void smb_direct_send_ctx_init(struct smbdirect_send_batch *send_ctx,
1184				     bool need_invalidate_rkey,
1185				     unsigned int remote_key)
1186{
1187	INIT_LIST_HEAD(&send_ctx->msg_list);
1188	send_ctx->wr_cnt = 0;
1189	send_ctx->need_invalidate_rkey = need_invalidate_rkey;
1190	send_ctx->remote_key = remote_key;
1191	send_ctx->credit = 0;
1192}
1193
1194static int smb_direct_flush_send_list(struct smbdirect_socket *sc,
1195				      struct smbdirect_send_batch *send_ctx,
1196				      bool is_last)
1197{
1198	struct smbdirect_send_io *first, *last;
1199	int ret = 0;
1200
1201	if (list_empty(&send_ctx->msg_list))
1202		goto release_credit;
1203
1204	first = list_first_entry(&send_ctx->msg_list,
1205				 struct smbdirect_send_io,
1206				 sibling_list);
1207	last = list_last_entry(&send_ctx->msg_list,
1208			       struct smbdirect_send_io,
1209			       sibling_list);
1210
1211	if (send_ctx->need_invalidate_rkey) {
1212		first->wr.opcode = IB_WR_SEND_WITH_INV;
1213		first->wr.ex.invalidate_rkey = send_ctx->remote_key;
1214		send_ctx->need_invalidate_rkey = false;
1215		send_ctx->remote_key = 0;
1216	}
1217
1218	last->wr.send_flags = IB_SEND_SIGNALED;
1219	last->wr.wr_cqe = &last->cqe;
1220
1221	/*
1222	 * Remove last from send_ctx->msg_list
1223	 * and splice the rest of send_ctx->msg_list
1224	 * to last->sibling_list.
1225	 *
1226	 * send_ctx->msg_list is a valid empty list
1227	 * at the end.
1228	 */
1229	list_del_init(&last->sibling_list);
1230	list_splice_tail_init(&send_ctx->msg_list, &last->sibling_list);
1231	send_ctx->wr_cnt = 0;
1232
1233	ret = smb_direct_post_send(sc, &first->wr);
1234	if (ret) {
1235		struct smbdirect_send_io *sibling, *next;
1236
1237		list_for_each_entry_safe(sibling, next, &last->sibling_list, sibling_list) {
1238			list_del_init(&sibling->sibling_list);
1239			smb_direct_free_sendmsg(sc, sibling);
1240		}
1241		smb_direct_free_sendmsg(sc, last);
1242	}
1243
1244release_credit:
1245	if (is_last && !ret && send_ctx->credit) {
1246		atomic_add(send_ctx->credit, &sc->send_io.bcredits.count);
1247		send_ctx->credit = 0;
1248		wake_up(&sc->send_io.bcredits.wait_queue);
1249	}
1250
1251	return ret;
1252}
1253
1254static int wait_for_credits(struct smbdirect_socket *sc,
1255			    wait_queue_head_t *waitq, atomic_t *total_credits,
1256			    int needed)
1257{
1258	int ret;
1259
1260	do {
1261		if (atomic_sub_return(needed, total_credits) >= 0)
1262			return 0;
1263
1264		atomic_add(needed, total_credits);
1265		ret = wait_event_interruptible(*waitq,
1266					       atomic_read(total_credits) >= needed ||
1267					       sc->status != SMBDIRECT_SOCKET_CONNECTED);
1268
1269		if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
1270			return -ENOTCONN;
1271		else if (ret < 0)
1272			return ret;
1273	} while (true);
1274}
1275
1276static int wait_for_send_bcredit(struct smbdirect_socket *sc,
1277				 struct smbdirect_send_batch *send_ctx)
1278{
1279	int ret;
1280
1281	if (send_ctx->credit)
1282		return 0;
1283
1284	ret = wait_for_credits(sc,
1285			       &sc->send_io.bcredits.wait_queue,
1286			       &sc->send_io.bcredits.count,
1287			       1);
1288	if (ret)
1289		return ret;
1290
1291	send_ctx->credit = 1;
1292	return 0;
1293}
1294
1295static int wait_for_send_lcredit(struct smbdirect_socket *sc,
1296				 struct smbdirect_send_batch *send_ctx)
1297{
1298	if (send_ctx && (atomic_read(&sc->send_io.lcredits.count) <= 1)) {
1299		int ret;
1300
1301		ret = smb_direct_flush_send_list(sc, send_ctx, false);
1302		if (ret)
1303			return ret;
1304	}
1305
1306	return wait_for_credits(sc,
1307				&sc->send_io.lcredits.wait_queue,
1308				&sc->send_io.lcredits.count,
1309				1);
1310}
1311
1312static int wait_for_send_credits(struct smbdirect_socket *sc,
1313				 struct smbdirect_send_batch *send_ctx)
1314{
1315	int ret;
1316
1317	if (send_ctx &&
1318	    (send_ctx->wr_cnt >= 16 || atomic_read(&sc->send_io.credits.count) <= 1)) {
1319		ret = smb_direct_flush_send_list(sc, send_ctx, false);
1320		if (ret)
1321			return ret;
1322	}
1323
1324	return wait_for_credits(sc, &sc->send_io.credits.wait_queue, &sc->send_io.credits.count, 1);
1325}
1326
1327static int wait_for_rw_credits(struct smbdirect_socket *sc, int credits)
1328{
1329	return wait_for_credits(sc,
1330				&sc->rw_io.credits.wait_queue,
1331				&sc->rw_io.credits.count,
1332				credits);
1333}
1334
1335static int calc_rw_credits(struct smbdirect_socket *sc,
1336			   char *buf, unsigned int len)
1337{
1338	return DIV_ROUND_UP(get_buf_page_count(buf, len),
1339			    sc->rw_io.credits.num_pages);
1340}
1341
1342static int smb_direct_create_header(struct smbdirect_socket *sc,
1343				    int size, int remaining_data_length,
1344				    int new_credits,
1345				    struct smbdirect_send_io **sendmsg_out)
1346{
1347	struct smbdirect_socket_parameters *sp = &sc->parameters;
1348	struct smbdirect_send_io *sendmsg;
1349	struct smbdirect_data_transfer *packet;
1350	int header_length;
1351	int ret;
1352
1353	sendmsg = smb_direct_alloc_sendmsg(sc);
1354	if (IS_ERR(sendmsg))
1355		return PTR_ERR(sendmsg);
1356
1357	/* Fill in the packet header */
1358	packet = (struct smbdirect_data_transfer *)sendmsg->packet;
1359	packet->credits_requested = cpu_to_le16(sp->send_credit_target);
1360	packet->credits_granted = cpu_to_le16(new_credits);
1361
1362	packet->flags = 0;
1363	if (manage_keep_alive_before_sending(sc))
1364		packet->flags |= cpu_to_le16(SMBDIRECT_FLAG_RESPONSE_REQUESTED);
1365
1366	packet->reserved = 0;
1367	if (!size)
1368		packet->data_offset = 0;
1369	else
1370		packet->data_offset = cpu_to_le32(24);
1371	packet->data_length = cpu_to_le32(size);
1372	packet->remaining_data_length = cpu_to_le32(remaining_data_length);
1373	packet->padding = 0;
1374
1375	ksmbd_debug(RDMA,
1376		    "credits_requested=%d credits_granted=%d data_offset=%d data_length=%d remaining_data_length=%d\n",
1377		    le16_to_cpu(packet->credits_requested),
1378		    le16_to_cpu(packet->credits_granted),
1379		    le32_to_cpu(packet->data_offset),
1380		    le32_to_cpu(packet->data_length),
1381		    le32_to_cpu(packet->remaining_data_length));
1382
1383	/* Map the packet to DMA */
1384	header_length = sizeof(struct smbdirect_data_transfer);
1385	/* If this is a packet without payload, don't send padding */
1386	if (!size)
1387		header_length =
1388			offsetof(struct smbdirect_data_transfer, padding);
1389
1390	sendmsg->sge[0].addr = ib_dma_map_single(sc->ib.dev,
1391						 (void *)packet,
1392						 header_length,
1393						 DMA_TO_DEVICE);
1394	ret = ib_dma_mapping_error(sc->ib.dev, sendmsg->sge[0].addr);
1395	if (ret) {
1396		smb_direct_free_sendmsg(sc, sendmsg);
1397		return ret;
1398	}
1399
1400	sendmsg->num_sge = 1;
1401	sendmsg->sge[0].length = header_length;
1402	sendmsg->sge[0].lkey = sc->ib.pd->local_dma_lkey;
1403
1404	*sendmsg_out = sendmsg;
1405	return 0;
1406}
1407
1408static int get_sg_list(void *buf, int size, struct scatterlist *sg_list, int nentries)
1409{
1410	bool high = is_vmalloc_addr(buf);
1411	struct page *page;
1412	int offset, len;
1413	int i = 0;
1414
1415	if (size <= 0 || nentries < get_buf_page_count(buf, size))
1416		return -EINVAL;
1417
1418	offset = offset_in_page(buf);
1419	buf -= offset;
1420	while (size > 0) {
1421		len = min_t(int, PAGE_SIZE - offset, size);
1422		if (high)
1423			page = vmalloc_to_page(buf);
1424		else
1425			page = kmap_to_page(buf);
1426
1427		if (!sg_list)
1428			return -EINVAL;
1429		sg_set_page(sg_list, page, len, offset);
1430		sg_list = sg_next(sg_list);
1431
1432		buf += PAGE_SIZE;
1433		size -= len;
1434		offset = 0;
1435		i++;
1436	}
1437	return i;
1438}
1439
1440static int get_mapped_sg_list(struct ib_device *device, void *buf, int size,
1441			      struct scatterlist *sg_list, int nentries,
1442			      enum dma_data_direction dir, int *npages)
1443{
1444	*npages = get_sg_list(buf, size, sg_list, nentries);
1445	if (*npages < 0)
1446		return -EINVAL;
1447	return ib_dma_map_sg(device, sg_list, *npages, dir);
1448}
1449
1450static int post_sendmsg(struct smbdirect_socket *sc,
1451			struct smbdirect_send_batch *send_ctx,
1452			struct smbdirect_send_io *msg)
1453{
1454	int i;
1455
1456	for (i = 0; i < msg->num_sge; i++)
1457		ib_dma_sync_single_for_device(sc->ib.dev,
1458					      msg->sge[i].addr, msg->sge[i].length,
1459					      DMA_TO_DEVICE);
1460
1461	msg->cqe.done = send_done;
1462	msg->wr.opcode = IB_WR_SEND;
1463	msg->wr.sg_list = &msg->sge[0];
1464	msg->wr.num_sge = msg->num_sge;
1465	msg->wr.next = NULL;
1466
1467	if (send_ctx) {
1468		msg->wr.wr_cqe = NULL;
1469		msg->wr.send_flags = 0;
1470		if (!list_empty(&send_ctx->msg_list)) {
1471			struct smbdirect_send_io *last;
1472
1473			last = list_last_entry(&send_ctx->msg_list,
1474					       struct smbdirect_send_io,
1475					       sibling_list);
1476			last->wr.next = &msg->wr;
1477		}
1478		list_add_tail(&msg->sibling_list, &send_ctx->msg_list);
1479		send_ctx->wr_cnt++;
1480		return 0;
1481	}
1482
1483	msg->wr.wr_cqe = &msg->cqe;
1484	msg->wr.send_flags = IB_SEND_SIGNALED;
1485	return smb_direct_post_send(sc, &msg->wr);
1486}
1487
1488static int smb_direct_post_send_data(struct smbdirect_socket *sc,
1489				     struct smbdirect_send_batch *send_ctx,
1490				     struct kvec *iov, int niov,
1491				     int remaining_data_length)
1492{
1493	int i, j, ret;
1494	struct smbdirect_send_io *msg;
1495	int data_length;
1496	struct scatterlist sg[SMBDIRECT_SEND_IO_MAX_SGE - 1];
1497	struct smbdirect_send_batch _send_ctx;
1498	int new_credits;
1499
1500	if (!send_ctx) {
1501		smb_direct_send_ctx_init(&_send_ctx, false, 0);
1502		send_ctx = &_send_ctx;
1503	}
1504
1505	ret = wait_for_send_bcredit(sc, send_ctx);
1506	if (ret)
1507		goto bcredit_failed;
1508
1509	ret = wait_for_send_lcredit(sc, send_ctx);
1510	if (ret)
1511		goto lcredit_failed;
1512
1513	ret = wait_for_send_credits(sc, send_ctx);
1514	if (ret)
1515		goto credit_failed;
1516
1517	new_credits = manage_credits_prior_sending(sc);
1518	if (new_credits == 0 &&
1519	    atomic_read(&sc->send_io.credits.count) == 0 &&
1520	    atomic_read(&sc->recv_io.credits.count) == 0) {
1521		queue_work(sc->workqueue, &sc->recv_io.posted.refill_work);
1522		ret = wait_event_interruptible(sc->send_io.credits.wait_queue,
1523					       atomic_read(&sc->send_io.credits.count) >= 1 ||
1524					       atomic_read(&sc->recv_io.credits.available) >= 1 ||
1525					       sc->status != SMBDIRECT_SOCKET_CONNECTED);
1526		if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
1527			ret = -ENOTCONN;
1528		if (ret < 0)
1529			goto credit_failed;
1530
1531		new_credits = manage_credits_prior_sending(sc);
1532	}
1533
1534	data_length = 0;
1535	for (i = 0; i < niov; i++)
1536		data_length += iov[i].iov_len;
1537
1538	ret = smb_direct_create_header(sc, data_length, remaining_data_length,
1539				       new_credits, &msg);
1540	if (ret)
1541		goto header_failed;
1542
1543	for (i = 0; i < niov; i++) {
1544		struct ib_sge *sge;
1545		int sg_cnt;
1546		int npages;
1547
1548		sg_init_table(sg, SMBDIRECT_SEND_IO_MAX_SGE - 1);
1549		sg_cnt = get_mapped_sg_list(sc->ib.dev,
1550					    iov[i].iov_base, iov[i].iov_len,
1551					    sg, SMBDIRECT_SEND_IO_MAX_SGE - 1,
1552					    DMA_TO_DEVICE, &npages);
1553		if (sg_cnt <= 0) {
1554			pr_err("failed to map buffer\n");
1555			ret = -ENOMEM;
1556			goto err;
1557		} else if (sg_cnt + msg->num_sge > SMBDIRECT_SEND_IO_MAX_SGE) {
1558			pr_err("buffer not fitted into sges\n");
1559			ret = -E2BIG;
1560			ib_dma_unmap_sg(sc->ib.dev, sg, npages,
1561					DMA_TO_DEVICE);
1562			goto err;
1563		}
1564
1565		for (j = 0; j < sg_cnt; j++) {
1566			sge = &msg->sge[msg->num_sge];
1567			sge->addr = sg_dma_address(&sg[j]);
1568			sge->length = sg_dma_len(&sg[j]);
1569			sge->lkey  = sc->ib.pd->local_dma_lkey;
1570			msg->num_sge++;
1571		}
1572	}
1573
1574	ret = post_sendmsg(sc, send_ctx, msg);
1575	if (ret)
1576		goto err;
1577
1578	/*
1579	 * From here msg is moved to send_ctx
1580	 * and we should not free it explicitly.
1581	 */
1582
1583	if (send_ctx == &_send_ctx) {
1584		ret = smb_direct_flush_send_list(sc, send_ctx, true);
1585		if (ret)
1586			goto flush_failed;
1587	}
1588
1589	return 0;
1590err:
1591	smb_direct_free_sendmsg(sc, msg);
1592flush_failed:
1593header_failed:
1594	atomic_inc(&sc->send_io.credits.count);
1595credit_failed:
1596	atomic_inc(&sc->send_io.lcredits.count);
1597lcredit_failed:
1598	atomic_add(send_ctx->credit, &sc->send_io.bcredits.count);
1599	send_ctx->credit = 0;
1600bcredit_failed:
1601	return ret;
1602}
1603
1604static int smb_direct_writev(struct ksmbd_transport *t,
1605			     struct kvec *iov, int niovs, int buflen,
1606			     bool need_invalidate, unsigned int remote_key)
1607{
1608	struct smb_direct_transport *st = SMBD_TRANS(t);
1609	struct smbdirect_socket *sc = &st->socket;
1610	struct smbdirect_socket_parameters *sp = &sc->parameters;
1611	size_t remaining_data_length;
1612	size_t iov_idx;
1613	size_t iov_ofs;
1614	size_t max_iov_size = sp->max_send_size -
1615			sizeof(struct smbdirect_data_transfer);
1616	int ret;
1617	struct smbdirect_send_batch send_ctx;
1618	int error = 0;
1619
1620	if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
1621		return -ENOTCONN;
1622
1623	//FIXME: skip RFC1002 header..
1624	if (WARN_ON_ONCE(niovs <= 1 || iov[0].iov_len != 4))
1625		return -EINVAL;
1626	buflen -= 4;
1627	iov_idx = 1;
1628	iov_ofs = 0;
1629
1630	remaining_data_length = buflen;
1631	ksmbd_debug(RDMA, "Sending smb (RDMA): smb_len=%u\n", buflen);
1632
1633	smb_direct_send_ctx_init(&send_ctx, need_invalidate, remote_key);
1634	while (remaining_data_length) {
1635		struct kvec vecs[SMBDIRECT_SEND_IO_MAX_SGE - 1]; /* minus smbdirect hdr */
1636		size_t possible_bytes = max_iov_size;
1637		size_t possible_vecs;
1638		size_t bytes = 0;
1639		size_t nvecs = 0;
1640
1641		/*
1642		 * For the last message remaining_data_length should be
1643		 * have been 0 already!
1644		 */
1645		if (WARN_ON_ONCE(iov_idx >= niovs)) {
1646			error = -EINVAL;
1647			goto done;
1648		}
1649
1650		/*
1651		 * We have 2 factors which limit the arguments we pass
1652		 * to smb_direct_post_send_data():
1653		 *
1654		 * 1. The number of supported sges for the send,
1655		 *    while one is reserved for the smbdirect header.
1656		 *    And we currently need one SGE per page.
1657		 * 2. The number of negotiated payload bytes per send.
1658		 */
1659		possible_vecs = min_t(size_t, ARRAY_SIZE(vecs), niovs - iov_idx);
1660
1661		while (iov_idx < niovs && possible_vecs && possible_bytes) {
1662			struct kvec *v = &vecs[nvecs];
1663			int page_count;
1664
1665			v->iov_base = ((u8 *)iov[iov_idx].iov_base) + iov_ofs;
1666			v->iov_len = min_t(size_t,
1667					   iov[iov_idx].iov_len - iov_ofs,
1668					   possible_bytes);
1669			page_count = get_buf_page_count(v->iov_base, v->iov_len);
1670			if (page_count > possible_vecs) {
1671				/*
1672				 * If the number of pages in the buffer
1673				 * is to much (because we currently require
1674				 * one SGE per page), we need to limit the
1675				 * length.
1676				 *
1677				 * We know possible_vecs is at least 1,
1678				 * so we always keep the first page.
1679				 *
1680				 * We need to calculate the number extra
1681				 * pages (epages) we can also keep.
1682				 *
1683				 * We calculate the number of bytes in the
1684				 * first page (fplen), this should never be
1685				 * larger than v->iov_len because page_count is
1686				 * at least 2, but adding a limitation feels
1687				 * better.
1688				 *
1689				 * Then we calculate the number of bytes (elen)
1690				 * we can keep for the extra pages.
1691				 */
1692				size_t epages = possible_vecs - 1;
1693				size_t fpofs = offset_in_page(v->iov_base);
1694				size_t fplen = min_t(size_t, PAGE_SIZE - fpofs, v->iov_len);
1695				size_t elen = min_t(size_t, v->iov_len - fplen, epages*PAGE_SIZE);
1696
1697				v->iov_len = fplen + elen;
1698				page_count = get_buf_page_count(v->iov_base, v->iov_len);
1699				if (WARN_ON_ONCE(page_count > possible_vecs)) {
1700					/*
1701					 * Something went wrong in the above
1702					 * logic...
1703					 */
1704					error = -EINVAL;
1705					goto done;
1706				}
1707			}
1708			possible_vecs -= page_count;
1709			nvecs += 1;
1710			possible_bytes -= v->iov_len;
1711			bytes += v->iov_len;
1712
1713			iov_ofs += v->iov_len;
1714			if (iov_ofs >= iov[iov_idx].iov_len) {
1715				iov_idx += 1;
1716				iov_ofs = 0;
1717			}
1718		}
1719
1720		remaining_data_length -= bytes;
1721
1722		ret = smb_direct_post_send_data(sc, &send_ctx,
1723						vecs, nvecs,
1724						remaining_data_length);
1725		if (unlikely(ret)) {
1726			error = ret;
1727			goto done;
1728		}
1729	}
1730
1731done:
1732	ret = smb_direct_flush_send_list(sc, &send_ctx, true);
1733	if (unlikely(!ret && error))
1734		ret = error;
1735
1736	/*
1737	 * As an optimization, we don't wait for individual I/O to finish
1738	 * before sending the next one.
1739	 * Send them all and wait for pending send count to get to 0
1740	 * that means all the I/Os have been out and we are good to return
1741	 */
1742
1743	wait_event(sc->send_io.pending.zero_wait_queue,
1744		   atomic_read(&sc->send_io.pending.count) == 0 ||
1745		   sc->status != SMBDIRECT_SOCKET_CONNECTED);
1746	if (sc->status != SMBDIRECT_SOCKET_CONNECTED && ret == 0)
1747		ret = -ENOTCONN;
1748
1749	return ret;
1750}
1751
1752static void smb_direct_free_rdma_rw_msg(struct smb_direct_transport *t,
1753					struct smbdirect_rw_io *msg,
1754					enum dma_data_direction dir)
1755{
1756	struct smbdirect_socket *sc = &t->socket;
1757
1758	rdma_rw_ctx_destroy(&msg->rdma_ctx, sc->ib.qp, sc->ib.qp->port,
1759			    msg->sgt.sgl, msg->sgt.nents, dir);
1760	sg_free_table_chained(&msg->sgt, SG_CHUNK_SIZE);
1761	kfree(msg);
1762}
1763
1764static void read_write_done(struct ib_cq *cq, struct ib_wc *wc,
1765			    enum dma_data_direction dir)
1766{
1767	struct smbdirect_rw_io *msg =
1768		container_of(wc->wr_cqe, struct smbdirect_rw_io, cqe);
1769	struct smbdirect_socket *sc = msg->socket;
1770
1771	if (wc->status != IB_WC_SUCCESS) {
1772		msg->error = -EIO;
1773		pr_err("read/write error. opcode = %d, status = %s(%d)\n",
1774		       wc->opcode, ib_wc_status_msg(wc->status), wc->status);
1775		if (wc->status != IB_WC_WR_FLUSH_ERR)
1776			smb_direct_disconnect_rdma_connection(sc);
1777	}
1778
1779	complete(msg->completion);
1780}
1781
1782static void read_done(struct ib_cq *cq, struct ib_wc *wc)
1783{
1784	read_write_done(cq, wc, DMA_FROM_DEVICE);
1785}
1786
1787static void write_done(struct ib_cq *cq, struct ib_wc *wc)
1788{
1789	read_write_done(cq, wc, DMA_TO_DEVICE);
1790}
1791
1792static int smb_direct_rdma_xmit(struct smb_direct_transport *t,
1793				void *buf, int buf_len,
1794				struct smbdirect_buffer_descriptor_v1 *desc,
1795				unsigned int desc_len,
1796				bool is_read)
1797{
1798	struct smbdirect_socket *sc = &t->socket;
1799	struct smbdirect_socket_parameters *sp = &sc->parameters;
1800	struct smbdirect_rw_io *msg, *next_msg;
1801	int i, ret;
1802	DECLARE_COMPLETION_ONSTACK(completion);
1803	struct ib_send_wr *first_wr;
1804	LIST_HEAD(msg_list);
1805	char *desc_buf;
1806	int credits_needed;
1807	unsigned int desc_buf_len, desc_num = 0;
1808
1809	if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
1810		return -ENOTCONN;
1811
1812	if (buf_len > sp->max_read_write_size)
1813		return -EINVAL;
1814
1815	/* calculate needed credits */
1816	credits_needed = 0;
1817	desc_buf = buf;
1818	for (i = 0; i < desc_len / sizeof(*desc); i++) {
1819		if (!buf_len)
1820			break;
1821
1822		desc_buf_len = le32_to_cpu(desc[i].length);
1823		if (!desc_buf_len)
1824			return -EINVAL;
1825
1826		if (desc_buf_len > buf_len) {
1827			desc_buf_len = buf_len;
1828			desc[i].length = cpu_to_le32(desc_buf_len);
1829			buf_len = 0;
1830		}
1831
1832		credits_needed += calc_rw_credits(sc, desc_buf, desc_buf_len);
1833		desc_buf += desc_buf_len;
1834		buf_len -= desc_buf_len;
1835		desc_num++;
1836	}
1837
1838	ksmbd_debug(RDMA, "RDMA %s, len %#x, needed credits %#x\n",
1839		    str_read_write(is_read), buf_len, credits_needed);
1840
1841	ret = wait_for_rw_credits(sc, credits_needed);
1842	if (ret < 0)
1843		return ret;
1844
1845	/* build rdma_rw_ctx for each descriptor */
1846	desc_buf = buf;
1847	for (i = 0; i < desc_num; i++) {
1848		msg = kzalloc(struct_size(msg, sg_list, SG_CHUNK_SIZE),
1849			      KSMBD_DEFAULT_GFP);
1850		if (!msg) {
1851			ret = -ENOMEM;
1852			goto out;
1853		}
1854
1855		desc_buf_len = le32_to_cpu(desc[i].length);
1856
1857		msg->socket = sc;
1858		msg->cqe.done = is_read ? read_done : write_done;
1859		msg->completion = &completion;
1860
1861		msg->sgt.sgl = &msg->sg_list[0];
1862		ret = sg_alloc_table_chained(&msg->sgt,
1863					     get_buf_page_count(desc_buf, desc_buf_len),
1864					     msg->sg_list, SG_CHUNK_SIZE);
1865		if (ret) {
1866			ret = -ENOMEM;
1867			goto free_msg;
1868		}
1869
1870		ret = get_sg_list(desc_buf, desc_buf_len,
1871				  msg->sgt.sgl, msg->sgt.orig_nents);
1872		if (ret < 0)
1873			goto free_table;
1874
1875		ret = rdma_rw_ctx_init(&msg->rdma_ctx, sc->ib.qp, sc->ib.qp->port,
1876				       msg->sgt.sgl,
1877				       get_buf_page_count(desc_buf, desc_buf_len),
1878				       0,
1879				       le64_to_cpu(desc[i].offset),
1880				       le32_to_cpu(desc[i].token),
1881				       is_read ? DMA_FROM_DEVICE : DMA_TO_DEVICE);
1882		if (ret < 0) {
1883			pr_err("failed to init rdma_rw_ctx: %d\n", ret);
1884			goto free_table;
1885		}
1886
1887		list_add_tail(&msg->list, &msg_list);
1888		desc_buf += desc_buf_len;
1889	}
1890
1891	/* concatenate work requests of rdma_rw_ctxs */
1892	first_wr = NULL;
1893	list_for_each_entry_reverse(msg, &msg_list, list) {
1894		first_wr = rdma_rw_ctx_wrs(&msg->rdma_ctx, sc->ib.qp, sc->ib.qp->port,
1895					   &msg->cqe, first_wr);
1896	}
1897
1898	ret = ib_post_send(sc->ib.qp, first_wr, NULL);
1899	if (ret) {
1900		pr_err("failed to post send wr for RDMA R/W: %d\n", ret);
1901		goto out;
1902	}
1903
1904	msg = list_last_entry(&msg_list, struct smbdirect_rw_io, list);
1905	wait_for_completion(&completion);
1906	ret = msg->error;
1907out:
1908	list_for_each_entry_safe(msg, next_msg, &msg_list, list) {
1909		list_del(&msg->list);
1910		smb_direct_free_rdma_rw_msg(t, msg,
1911					    is_read ? DMA_FROM_DEVICE : DMA_TO_DEVICE);
1912	}
1913	atomic_add(credits_needed, &sc->rw_io.credits.count);
1914	wake_up(&sc->rw_io.credits.wait_queue);
1915	return ret;
1916
1917free_table:
1918	sg_free_table_chained(&msg->sgt, SG_CHUNK_SIZE);
1919free_msg:
1920	kfree(msg);
1921	goto out;
1922}
1923
1924static int smb_direct_rdma_write(struct ksmbd_transport *t,
1925				 void *buf, unsigned int buflen,
1926				 struct smbdirect_buffer_descriptor_v1 *desc,
1927				 unsigned int desc_len)
1928{
1929	return smb_direct_rdma_xmit(SMBD_TRANS(t), buf, buflen,
1930				    desc, desc_len, false);
1931}
1932
1933static int smb_direct_rdma_read(struct ksmbd_transport *t,
1934				void *buf, unsigned int buflen,
1935				struct smbdirect_buffer_descriptor_v1 *desc,
1936				unsigned int desc_len)
1937{
1938	return smb_direct_rdma_xmit(SMBD_TRANS(t), buf, buflen,
1939				    desc, desc_len, true);
1940}
1941
1942static void smb_direct_disconnect(struct ksmbd_transport *t)
1943{
1944	struct smb_direct_transport *st = SMBD_TRANS(t);
1945	struct smbdirect_socket *sc = &st->socket;
1946
1947	ksmbd_debug(RDMA, "Disconnecting cm_id=%p\n", sc->rdma.cm_id);
1948
1949	free_transport(st);
1950}
1951
1952static void smb_direct_shutdown(struct ksmbd_transport *t)
1953{
1954	struct smb_direct_transport *st = SMBD_TRANS(t);
1955	struct smbdirect_socket *sc = &st->socket;
1956
1957	ksmbd_debug(RDMA, "smb-direct shutdown cm_id=%p\n", sc->rdma.cm_id);
1958
1959	smb_direct_disconnect_rdma_work(&sc->disconnect_work);
1960}
1961
1962static int smb_direct_cm_handler(struct rdma_cm_id *cm_id,
1963				 struct rdma_cm_event *event)
1964{
1965	struct smbdirect_socket *sc = cm_id->context;
1966	unsigned long flags;
1967
1968	ksmbd_debug(RDMA, "RDMA CM event. cm_id=%p event=%s (%d)\n",
1969		    cm_id, rdma_event_msg(event->event), event->event);
1970
1971	switch (event->event) {
1972	case RDMA_CM_EVENT_ESTABLISHED: {
1973		/*
1974		 * Some drivers (at least mlx5_ib and irdma in roce mode)
1975		 * might post a recv completion before RDMA_CM_EVENT_ESTABLISHED,
1976		 * we need to adjust our expectation in that case.
1977		 *
1978		 * If smb_direct_negotiate_recv_done was called first
1979		 * it initialized sc->connect.work only for us to
1980		 * start, so that we turned into
1981		 * SMBDIRECT_SOCKET_NEGOTIATE_NEEDED, before
1982		 * smb_direct_negotiate_recv_work() runs.
1983		 *
1984		 * If smb_direct_negotiate_recv_done didn't happen
1985		 * yet. sc->connect.work is still be disabled and
1986		 * queue_work() is a no-op.
1987		 */
1988		if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING))
1989			break;
1990		sc->status = SMBDIRECT_SOCKET_NEGOTIATE_NEEDED;
1991		spin_lock_irqsave(&sc->connect.lock, flags);
1992		if (!sc->first_error)
1993			queue_work(sc->workqueue, &sc->connect.work);
1994		spin_unlock_irqrestore(&sc->connect.lock, flags);
1995		wake_up(&sc->status_wait);
1996		break;
1997	}
1998	case RDMA_CM_EVENT_DEVICE_REMOVAL:
1999	case RDMA_CM_EVENT_DISCONNECTED: {
2000		sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
2001		smb_direct_disconnect_rdma_work(&sc->disconnect_work);
2002		if (sc->ib.qp)
2003			ib_drain_qp(sc->ib.qp);
2004		break;
2005	}
2006	case RDMA_CM_EVENT_CONNECT_ERROR: {
2007		sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
2008		smb_direct_disconnect_rdma_work(&sc->disconnect_work);
2009		break;
2010	}
2011	default:
2012		pr_err("Unexpected RDMA CM event. cm_id=%p, event=%s (%d)\n",
2013		       cm_id, rdma_event_msg(event->event),
2014		       event->event);
2015		break;
2016	}
2017	return 0;
2018}
2019
2020static void smb_direct_qpair_handler(struct ib_event *event, void *context)
2021{
2022	struct smbdirect_socket *sc = context;
2023
2024	ksmbd_debug(RDMA, "Received QP event. cm_id=%p, event=%s (%d)\n",
2025		    sc->rdma.cm_id, ib_event_msg(event->event), event->event);
2026
2027	switch (event->event) {
2028	case IB_EVENT_CQ_ERR:
2029	case IB_EVENT_QP_FATAL:
2030		smb_direct_disconnect_rdma_connection(sc);
2031		break;
2032	default:
2033		break;
2034	}
2035}
2036
2037static int smb_direct_send_negotiate_response(struct smbdirect_socket *sc,
2038					      int failed)
2039{
2040	struct smbdirect_socket_parameters *sp = &sc->parameters;
2041	struct smbdirect_send_io *sendmsg;
2042	struct smbdirect_negotiate_resp *resp;
2043	int ret;
2044
2045	sendmsg = smb_direct_alloc_sendmsg(sc);
2046	if (IS_ERR(sendmsg))
2047		return -ENOMEM;
2048
2049	resp = (struct smbdirect_negotiate_resp *)sendmsg->packet;
2050	if (failed) {
2051		memset(resp, 0, sizeof(*resp));
2052		resp->min_version = SMB_DIRECT_VERSION_LE;
2053		resp->max_version = SMB_DIRECT_VERSION_LE;
2054		resp->status = STATUS_NOT_SUPPORTED;
2055
2056		sc->status = SMBDIRECT_SOCKET_NEGOTIATE_FAILED;
2057	} else {
2058		resp->status = STATUS_SUCCESS;
2059		resp->min_version = SMB_DIRECT_VERSION_LE;
2060		resp->max_version = SMB_DIRECT_VERSION_LE;
2061		resp->negotiated_version = SMB_DIRECT_VERSION_LE;
2062		resp->reserved = 0;
2063		resp->credits_requested =
2064				cpu_to_le16(sp->send_credit_target);
2065		resp->credits_granted = cpu_to_le16(manage_credits_prior_sending(sc));
2066		resp->max_readwrite_size = cpu_to_le32(sp->max_read_write_size);
2067		resp->preferred_send_size = cpu_to_le32(sp->max_send_size);
2068		resp->max_receive_size = cpu_to_le32(sp->max_recv_size);
2069		resp->max_fragmented_size =
2070				cpu_to_le32(sp->max_fragmented_recv_size);
2071
2072		atomic_set(&sc->send_io.bcredits.count, 1);
2073		sc->recv_io.expected = SMBDIRECT_EXPECT_DATA_TRANSFER;
2074		sc->status = SMBDIRECT_SOCKET_CONNECTED;
2075	}
2076
2077	sendmsg->sge[0].addr = ib_dma_map_single(sc->ib.dev,
2078						 (void *)resp, sizeof(*resp),
2079						 DMA_TO_DEVICE);
2080	ret = ib_dma_mapping_error(sc->ib.dev, sendmsg->sge[0].addr);
2081	if (ret) {
2082		smb_direct_free_sendmsg(sc, sendmsg);
2083		return ret;
2084	}
2085
2086	sendmsg->num_sge = 1;
2087	sendmsg->sge[0].length = sizeof(*resp);
2088	sendmsg->sge[0].lkey = sc->ib.pd->local_dma_lkey;
2089
2090	ret = post_sendmsg(sc, NULL, sendmsg);
2091	if (ret) {
2092		smb_direct_free_sendmsg(sc, sendmsg);
2093		return ret;
2094	}
2095
2096	wait_event(sc->send_io.pending.zero_wait_queue,
2097		   atomic_read(&sc->send_io.pending.count) == 0 ||
2098		   sc->status != SMBDIRECT_SOCKET_CONNECTED);
2099	if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
2100		return -ENOTCONN;
2101
2102	return 0;
2103}
2104
2105static int smb_direct_accept_client(struct smbdirect_socket *sc)
2106{
2107	struct smbdirect_socket_parameters *sp = &sc->parameters;
2108	struct rdma_conn_param conn_param;
2109	__be32 ird_ord_hdr[2];
2110	int ret;
2111
2112	/*
2113	 * smb_direct_handle_connect_request()
2114	 * already negotiated sp->initiator_depth
2115	 * and sp->responder_resources
2116	 */
2117	memset(&conn_param, 0, sizeof(conn_param));
2118	conn_param.initiator_depth = sp->initiator_depth;
2119	conn_param.responder_resources = sp->responder_resources;
2120
2121	if (sc->rdma.legacy_iwarp) {
2122		ird_ord_hdr[0] = cpu_to_be32(conn_param.responder_resources);
2123		ird_ord_hdr[1] = cpu_to_be32(conn_param.initiator_depth);
2124		conn_param.private_data = ird_ord_hdr;
2125		conn_param.private_data_len = sizeof(ird_ord_hdr);
2126	} else {
2127		conn_param.private_data = NULL;
2128		conn_param.private_data_len = 0;
2129	}
2130	conn_param.retry_count = SMB_DIRECT_CM_RETRY;
2131	conn_param.rnr_retry_count = SMB_DIRECT_CM_RNR_RETRY;
2132	conn_param.flow_control = 0;
2133
2134	/*
2135	 * start with the negotiate timeout and SMBDIRECT_KEEPALIVE_PENDING
2136	 * so that the timer will cause a disconnect.
2137	 */
2138	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_PENDING;
2139	mod_delayed_work(sc->workqueue, &sc->idle.timer_work,
2140			 msecs_to_jiffies(sp->negotiate_timeout_msec));
2141
2142	WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED);
2143	sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING;
2144	ret = rdma_accept(sc->rdma.cm_id, &conn_param);
2145	if (ret) {
2146		pr_err("error at rdma_accept: %d\n", ret);
2147		return ret;
2148	}
2149	return 0;
2150}
2151
2152static int smb_direct_prepare_negotiation(struct smbdirect_socket *sc)
2153{
2154	struct smbdirect_recv_io *recvmsg;
2155	bool recv_posted = false;
2156	int ret;
2157
2158	WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_CREATED);
2159	sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED;
2160
2161	sc->recv_io.expected = SMBDIRECT_EXPECT_NEGOTIATE_REQ;
2162
2163	recvmsg = get_free_recvmsg(sc);
2164	if (!recvmsg)
2165		return -ENOMEM;
2166	recvmsg->cqe.done = smb_direct_negotiate_recv_done;
2167
2168	ret = smb_direct_post_recv(sc, recvmsg);
2169	if (ret) {
2170		pr_err("Can't post recv: %d\n", ret);
2171		goto out_err;
2172	}
2173	recv_posted = true;
2174
2175	ret = smb_direct_accept_client(sc);
2176	if (ret) {
2177		pr_err("Can't accept client\n");
2178		goto out_err;
2179	}
2180
2181	return 0;
2182out_err:
2183	/*
2184	 * If the recv was never posted, return it to the free list.
2185	 * If it was posted, leave it alone so disconnect teardown can
2186	 * drain the QP and complete it (flush) and the completion path
2187	 * will unmap it exactly once.
2188	 */
2189	if (!recv_posted)
2190		put_recvmsg(sc, recvmsg);
2191	return ret;
2192}
2193
2194static int smb_direct_init_params(struct smbdirect_socket *sc)
2195{
2196	struct smbdirect_socket_parameters *sp = &sc->parameters;
2197	int max_send_sges;
2198	unsigned int maxpages;
2199
2200	/* need 3 more sge. because a SMB_DIRECT header, SMB2 header,
2201	 * SMB2 response could be mapped.
2202	 */
2203	max_send_sges = DIV_ROUND_UP(sp->max_send_size, PAGE_SIZE) + 3;
2204	if (max_send_sges > SMBDIRECT_SEND_IO_MAX_SGE) {
2205		pr_err("max_send_size %d is too large\n", sp->max_send_size);
2206		return -EINVAL;
2207	}
2208
2209	atomic_set(&sc->send_io.lcredits.count, sp->send_credit_target);
2210
2211	maxpages = DIV_ROUND_UP(sp->max_read_write_size, PAGE_SIZE);
2212	sc->rw_io.credits.max = rdma_rw_mr_factor(sc->ib.dev,
2213						  sc->rdma.cm_id->port_num,
2214						  maxpages);
2215	sc->rw_io.credits.num_pages = DIV_ROUND_UP(maxpages, sc->rw_io.credits.max);
2216	/* add one extra in order to handle unaligned pages */
2217	sc->rw_io.credits.max += 1;
2218
2219	sc->recv_io.credits.target = 1;
2220
2221	atomic_set(&sc->rw_io.credits.count, sc->rw_io.credits.max);
2222
2223	return 0;
2224}
2225
2226static void smb_direct_destroy_pools(struct smbdirect_socket *sc)
2227{
2228	struct smbdirect_recv_io *recvmsg;
2229
2230	while ((recvmsg = get_free_recvmsg(sc)))
2231		mempool_free(recvmsg, sc->recv_io.mem.pool);
2232
2233	mempool_destroy(sc->recv_io.mem.pool);
2234	sc->recv_io.mem.pool = NULL;
2235
2236	kmem_cache_destroy(sc->recv_io.mem.cache);
2237	sc->recv_io.mem.cache = NULL;
2238
2239	mempool_destroy(sc->send_io.mem.pool);
2240	sc->send_io.mem.pool = NULL;
2241
2242	kmem_cache_destroy(sc->send_io.mem.cache);
2243	sc->send_io.mem.cache = NULL;
2244}
2245
2246static int smb_direct_create_pools(struct smbdirect_socket *sc)
2247{
2248	struct smbdirect_socket_parameters *sp = &sc->parameters;
2249	char name[80];
2250	int i;
2251	struct smbdirect_recv_io *recvmsg;
2252
2253	snprintf(name, sizeof(name), "smbdirect_send_io_pool_%p", sc);
2254	sc->send_io.mem.cache = kmem_cache_create(name,
2255					     sizeof(struct smbdirect_send_io) +
2256					      sizeof(struct smbdirect_negotiate_resp),
2257					     0, SLAB_HWCACHE_ALIGN, NULL);
2258	if (!sc->send_io.mem.cache)
2259		return -ENOMEM;
2260
2261	sc->send_io.mem.pool = mempool_create(sp->send_credit_target,
2262					    mempool_alloc_slab, mempool_free_slab,
2263					    sc->send_io.mem.cache);
2264	if (!sc->send_io.mem.pool)
2265		goto err;
2266
2267	snprintf(name, sizeof(name), "smbdirect_recv_io_pool_%p", sc);
2268	sc->recv_io.mem.cache = kmem_cache_create(name,
2269					     sizeof(struct smbdirect_recv_io) +
2270					     sp->max_recv_size,
2271					     0, SLAB_HWCACHE_ALIGN, NULL);
2272	if (!sc->recv_io.mem.cache)
2273		goto err;
2274
2275	sc->recv_io.mem.pool =
2276		mempool_create(sp->recv_credit_max, mempool_alloc_slab,
2277			       mempool_free_slab, sc->recv_io.mem.cache);
2278	if (!sc->recv_io.mem.pool)
2279		goto err;
2280
2281	for (i = 0; i < sp->recv_credit_max; i++) {
2282		recvmsg = mempool_alloc(sc->recv_io.mem.pool, KSMBD_DEFAULT_GFP);
2283		if (!recvmsg)
2284			goto err;
2285		recvmsg->socket = sc;
2286		recvmsg->sge.length = 0;
2287		list_add(&recvmsg->list, &sc->recv_io.free.list);
2288	}
2289
2290	return 0;
2291err:
2292	smb_direct_destroy_pools(sc);
2293	return -ENOMEM;
2294}
2295
2296static u32 smb_direct_rdma_rw_send_wrs(struct ib_device *dev, const struct ib_qp_init_attr *attr)
2297{
2298	/*
2299	 * This could be split out of rdma_rw_init_qp()
2300	 * and be a helper function next to rdma_rw_mr_factor()
2301	 *
2302	 * We can't check unlikely(rdma_rw_force_mr) here,
2303	 * but that is most likely 0 anyway.
2304	 */
2305	u32 factor;
2306
2307	WARN_ON_ONCE(attr->port_num == 0);
2308
2309	/*
2310	 * Each context needs at least one RDMA READ or WRITE WR.
2311	 *
2312	 * For some hardware we might need more, eventually we should ask the
2313	 * HCA driver for a multiplier here.
2314	 */
2315	factor = 1;
2316
2317	/*
2318	 * If the device needs MRs to perform RDMA READ or WRITE operations,
2319	 * we'll need two additional MRs for the registrations and the
2320	 * invalidation.
2321	 */
2322	if (rdma_protocol_iwarp(dev, attr->port_num) || dev->attrs.max_sgl_rd)
2323		factor += 2;	/* inv + reg */
2324
2325	return factor * attr->cap.max_rdma_ctxs;
2326}
2327
2328static int smb_direct_create_qpair(struct smbdirect_socket *sc)
2329{
2330	struct smbdirect_socket_parameters *sp = &sc->parameters;
2331	int ret;
2332	struct ib_qp_cap qp_cap;
2333	struct ib_qp_init_attr qp_attr;
2334	u32 max_send_wr;
2335	u32 rdma_send_wr;
2336
2337	/*
2338	 * Note that {rdma,ib}_create_qp() will call
2339	 * rdma_rw_init_qp() if cap->max_rdma_ctxs is not 0.
2340	 * It will adjust cap->max_send_wr to the required
2341	 * number of additional WRs for the RDMA RW operations.
2342	 * It will cap cap->max_send_wr to the device limit.
2343	 *
2344	 * +1 for ib_drain_qp
2345	 */
2346	qp_cap.max_send_wr = sp->send_credit_target + 1;
2347	qp_cap.max_recv_wr = sp->recv_credit_max + 1;
2348	qp_cap.max_send_sge = SMBDIRECT_SEND_IO_MAX_SGE;
2349	qp_cap.max_recv_sge = SMBDIRECT_RECV_IO_MAX_SGE;
2350	qp_cap.max_inline_data = 0;
2351	qp_cap.max_rdma_ctxs = sc->rw_io.credits.max;
2352
2353	/*
2354	 * Find out the number of max_send_wr
2355	 * after rdma_rw_init_qp() adjusted it.
2356	 *
2357	 * We only do it on a temporary variable,
2358	 * as rdma_create_qp() will trigger
2359	 * rdma_rw_init_qp() again.
2360	 */
2361	memset(&qp_attr, 0, sizeof(qp_attr));
2362	qp_attr.cap = qp_cap;
2363	qp_attr.port_num = sc->rdma.cm_id->port_num;
2364	rdma_send_wr = smb_direct_rdma_rw_send_wrs(sc->ib.dev, &qp_attr);
2365	max_send_wr = qp_cap.max_send_wr + rdma_send_wr;
2366
2367	if (qp_cap.max_send_wr > sc->ib.dev->attrs.max_cqe ||
2368	    qp_cap.max_send_wr > sc->ib.dev->attrs.max_qp_wr) {
2369		pr_err("Possible CQE overrun: max_send_wr %d\n",
2370		       qp_cap.max_send_wr);
2371		pr_err("device %.*s reporting max_cqe %d max_qp_wr %d\n",
2372		       IB_DEVICE_NAME_MAX,
2373		       sc->ib.dev->name,
2374		       sc->ib.dev->attrs.max_cqe,
2375		       sc->ib.dev->attrs.max_qp_wr);
2376		pr_err("consider lowering send_credit_target = %d\n",
2377		       sp->send_credit_target);
2378		return -EINVAL;
2379	}
2380
2381	if (qp_cap.max_rdma_ctxs &&
2382	    (max_send_wr >= sc->ib.dev->attrs.max_cqe ||
2383	     max_send_wr >= sc->ib.dev->attrs.max_qp_wr)) {
2384		pr_err("Possible CQE overrun: rdma_send_wr %d + max_send_wr %d = %d\n",
2385		       rdma_send_wr, qp_cap.max_send_wr, max_send_wr);
2386		pr_err("device %.*s reporting max_cqe %d max_qp_wr %d\n",
2387		       IB_DEVICE_NAME_MAX,
2388		       sc->ib.dev->name,
2389		       sc->ib.dev->attrs.max_cqe,
2390		       sc->ib.dev->attrs.max_qp_wr);
2391		pr_err("consider lowering send_credit_target = %d, max_rdma_ctxs = %d\n",
2392		       sp->send_credit_target, qp_cap.max_rdma_ctxs);
2393		return -EINVAL;
2394	}
2395
2396	if (qp_cap.max_recv_wr > sc->ib.dev->attrs.max_cqe ||
2397	    qp_cap.max_recv_wr > sc->ib.dev->attrs.max_qp_wr) {
2398		pr_err("Possible CQE overrun: max_recv_wr %d\n",
2399		       qp_cap.max_recv_wr);
2400		pr_err("device %.*s reporting max_cqe %d max_qp_wr %d\n",
2401		       IB_DEVICE_NAME_MAX,
2402		       sc->ib.dev->name,
2403		       sc->ib.dev->attrs.max_cqe,
2404		       sc->ib.dev->attrs.max_qp_wr);
2405		pr_err("consider lowering receive_credit_max = %d\n",
2406		       sp->recv_credit_max);
2407		return -EINVAL;
2408	}
2409
2410	if (qp_cap.max_send_sge > sc->ib.dev->attrs.max_send_sge ||
2411	    qp_cap.max_recv_sge > sc->ib.dev->attrs.max_recv_sge) {
2412		pr_err("device %.*s max_send_sge/max_recv_sge = %d/%d too small\n",
2413		       IB_DEVICE_NAME_MAX,
2414		       sc->ib.dev->name,
2415		       sc->ib.dev->attrs.max_send_sge,
2416		       sc->ib.dev->attrs.max_recv_sge);
2417		return -EINVAL;
2418	}
2419
2420	sc->ib.pd = ib_alloc_pd(sc->ib.dev, 0);
2421	if (IS_ERR(sc->ib.pd)) {
2422		pr_err("Can't create RDMA PD\n");
2423		ret = PTR_ERR(sc->ib.pd);
2424		sc->ib.pd = NULL;
2425		return ret;
2426	}
2427
2428	sc->ib.send_cq = ib_alloc_cq_any(sc->ib.dev, sc,
2429					 max_send_wr,
2430					 IB_POLL_WORKQUEUE);
2431	if (IS_ERR(sc->ib.send_cq)) {
2432		pr_err("Can't create RDMA send CQ\n");
2433		ret = PTR_ERR(sc->ib.send_cq);
2434		sc->ib.send_cq = NULL;
2435		goto err;
2436	}
2437
2438	sc->ib.recv_cq = ib_alloc_cq_any(sc->ib.dev, sc,
2439					 qp_cap.max_recv_wr,
2440					 IB_POLL_WORKQUEUE);
2441	if (IS_ERR(sc->ib.recv_cq)) {
2442		pr_err("Can't create RDMA recv CQ\n");
2443		ret = PTR_ERR(sc->ib.recv_cq);
2444		sc->ib.recv_cq = NULL;
2445		goto err;
2446	}
2447
2448	/*
2449	 * We reset completely here!
2450	 * As the above use was just temporary
2451	 * to calc max_send_wr and rdma_send_wr.
2452	 *
2453	 * rdma_create_qp() will trigger rdma_rw_init_qp()
2454	 * again if max_rdma_ctxs is not 0.
2455	 */
2456	memset(&qp_attr, 0, sizeof(qp_attr));
2457	qp_attr.event_handler = smb_direct_qpair_handler;
2458	qp_attr.qp_context = sc;
2459	qp_attr.cap = qp_cap;
2460	qp_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
2461	qp_attr.qp_type = IB_QPT_RC;
2462	qp_attr.send_cq = sc->ib.send_cq;
2463	qp_attr.recv_cq = sc->ib.recv_cq;
2464	qp_attr.port_num = ~0;
2465
2466	ret = rdma_create_qp(sc->rdma.cm_id, sc->ib.pd, &qp_attr);
2467	if (ret) {
2468		pr_err("Can't create RDMA QP: %d\n", ret);
2469		goto err;
2470	}
2471
2472	sc->ib.qp = sc->rdma.cm_id->qp;
2473	sc->rdma.cm_id->event_handler = smb_direct_cm_handler;
2474
2475	return 0;
2476err:
2477	if (sc->ib.qp) {
2478		sc->ib.qp = NULL;
2479		rdma_destroy_qp(sc->rdma.cm_id);
2480	}
2481	if (sc->ib.recv_cq) {
2482		ib_destroy_cq(sc->ib.recv_cq);
2483		sc->ib.recv_cq = NULL;
2484	}
2485	if (sc->ib.send_cq) {
2486		ib_destroy_cq(sc->ib.send_cq);
2487		sc->ib.send_cq = NULL;
2488	}
2489	if (sc->ib.pd) {
2490		ib_dealloc_pd(sc->ib.pd);
2491		sc->ib.pd = NULL;
2492	}
2493	return ret;
2494}
2495
2496static int smb_direct_prepare(struct ksmbd_transport *t)
2497{
2498	struct smb_direct_transport *st = SMBD_TRANS(t);
2499	struct smbdirect_socket *sc = &st->socket;
2500	struct smbdirect_socket_parameters *sp = &sc->parameters;
2501	struct smbdirect_recv_io *recvmsg;
2502	struct smbdirect_negotiate_req *req;
2503	unsigned long flags;
2504	int ret;
2505
2506	/*
2507	 * We are waiting to pass the following states:
2508	 *
2509	 * SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED
2510	 * SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING
2511	 * SMBDIRECT_SOCKET_NEGOTIATE_NEEDED
2512	 *
2513	 * To finally get to SMBDIRECT_SOCKET_NEGOTIATE_RUNNING
2514	 * in order to continue below.
2515	 *
2516	 * Everything else is unexpected and an error.
2517	 */
2518	ksmbd_debug(RDMA, "Waiting for SMB_DIRECT negotiate request\n");
2519	ret = wait_event_interruptible_timeout(sc->status_wait,
2520					sc->status != SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED &&
2521					sc->status != SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING &&
2522					sc->status != SMBDIRECT_SOCKET_NEGOTIATE_NEEDED,
2523					msecs_to_jiffies(sp->negotiate_timeout_msec));
2524	if (ret <= 0 || sc->status != SMBDIRECT_SOCKET_NEGOTIATE_RUNNING)
2525		return ret < 0 ? ret : -ETIMEDOUT;
2526
2527	recvmsg = get_first_reassembly(sc);
2528	if (!recvmsg)
2529		return -ECONNABORTED;
2530
2531	ret = smb_direct_check_recvmsg(recvmsg);
2532	if (ret)
2533		goto put;
2534
2535	req = (struct smbdirect_negotiate_req *)recvmsg->packet;
2536	sp->max_recv_size = min_t(u32, sp->max_recv_size,
2537				  le32_to_cpu(req->preferred_send_size));
2538	sp->max_send_size = min_t(u32, sp->max_send_size,
2539				  le32_to_cpu(req->max_receive_size));
2540	sp->max_fragmented_send_size =
2541		le32_to_cpu(req->max_fragmented_size);
2542	sp->max_fragmented_recv_size =
2543		(sp->recv_credit_max * sp->max_recv_size) / 2;
2544	sc->recv_io.credits.target = le16_to_cpu(req->credits_requested);
2545	sc->recv_io.credits.target = min_t(u16, sc->recv_io.credits.target, sp->recv_credit_max);
2546	sc->recv_io.credits.target = max_t(u16, sc->recv_io.credits.target, 1);
2547
2548put:
2549	spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
2550	sc->recv_io.reassembly.queue_length--;
2551	list_del(&recvmsg->list);
2552	spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
2553	put_recvmsg(sc, recvmsg);
2554
2555	if (ret == -ECONNABORTED)
2556		return ret;
2557
2558	if (ret)
2559		goto respond;
2560
2561	/*
2562	 * We negotiated with success, so we need to refill the recv queue.
2563	 * We do that with sc->idle.immediate_work still being disabled
2564	 * via smbdirect_socket_init(), so that queue_work(sc->workqueue,
2565	 * &sc->idle.immediate_work) in smb_direct_post_recv_credits()
2566	 * is a no-op.
2567	 *
2568	 * The message that grants the credits to the client is
2569	 * the negotiate response.
2570	 */
2571	INIT_WORK(&sc->recv_io.posted.refill_work, smb_direct_post_recv_credits);
2572	smb_direct_post_recv_credits(&sc->recv_io.posted.refill_work);
2573	if (unlikely(sc->first_error))
2574		return sc->first_error;
2575	INIT_WORK(&sc->idle.immediate_work, smb_direct_send_immediate_work);
2576
2577respond:
2578	ret = smb_direct_send_negotiate_response(sc, ret);
2579
2580	return ret;
2581}
2582
2583static int smb_direct_connect(struct smbdirect_socket *sc)
2584{
2585	struct smbdirect_recv_io *recv_io;
2586	int ret;
2587
2588	ret = smb_direct_init_params(sc);
2589	if (ret) {
2590		pr_err("Can't configure RDMA parameters\n");
2591		return ret;
2592	}
2593
2594	ret = smb_direct_create_pools(sc);
2595	if (ret) {
2596		pr_err("Can't init RDMA pool: %d\n", ret);
2597		return ret;
2598	}
2599
2600	list_for_each_entry(recv_io, &sc->recv_io.free.list, list)
2601		recv_io->cqe.done = recv_done;
2602
2603	ret = smb_direct_create_qpair(sc);
2604	if (ret) {
2605		pr_err("Can't accept RDMA client: %d\n", ret);
2606		return ret;
2607	}
2608
2609	ret = smb_direct_prepare_negotiation(sc);
2610	if (ret) {
2611		pr_err("Can't negotiate: %d\n", ret);
2612		return ret;
2613	}
2614	return 0;
2615}
2616
2617static bool rdma_frwr_is_supported(struct ib_device_attr *attrs)
2618{
2619	if (!(attrs->device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS))
2620		return false;
2621	if (attrs->max_fast_reg_page_list_len == 0)
2622		return false;
2623	return true;
2624}
2625
2626static int smb_direct_handle_connect_request(struct rdma_cm_id *new_cm_id,
2627					     struct rdma_cm_event *event)
2628{
2629	struct smb_direct_transport *t;
2630	struct smbdirect_socket *sc;
2631	struct smbdirect_socket_parameters *sp;
2632	struct task_struct *handler;
2633	u8 peer_initiator_depth;
2634	u8 peer_responder_resources;
2635	int ret;
2636
2637	if (!rdma_frwr_is_supported(&new_cm_id->device->attrs)) {
2638		ksmbd_debug(RDMA,
2639			    "Fast Registration Work Requests is not supported. device capabilities=%llx\n",
2640			    new_cm_id->device->attrs.device_cap_flags);
2641		return -EPROTONOSUPPORT;
2642	}
2643
2644	t = alloc_transport(new_cm_id);
2645	if (!t)
2646		return -ENOMEM;
2647	sc = &t->socket;
2648	sp = &sc->parameters;
2649
2650	peer_initiator_depth = event->param.conn.initiator_depth;
2651	peer_responder_resources = event->param.conn.responder_resources;
2652	if (rdma_protocol_iwarp(new_cm_id->device, new_cm_id->port_num) &&
2653	    event->param.conn.private_data_len == 8) {
2654		/*
2655		 * Legacy clients with only iWarp MPA v1 support
2656		 * need a private blob in order to negotiate
2657		 * the IRD/ORD values.
2658		 */
2659		const __be32 *ird_ord_hdr = event->param.conn.private_data;
2660		u32 ird32 = be32_to_cpu(ird_ord_hdr[0]);
2661		u32 ord32 = be32_to_cpu(ird_ord_hdr[1]);
2662
2663		/*
2664		 * cifs.ko sends the legacy IRD/ORD negotiation
2665		 * event if iWarp MPA v2 was used.
2666		 *
2667		 * Here we check that the values match and only
2668		 * mark the client as legacy if they don't match.
2669		 */
2670		if ((u32)event->param.conn.initiator_depth != ird32 ||
2671		    (u32)event->param.conn.responder_resources != ord32) {
2672			/*
2673			 * There are broken clients (old cifs.ko)
2674			 * using little endian and also
2675			 * struct rdma_conn_param only uses u8
2676			 * for initiator_depth and responder_resources,
2677			 * so we truncate the value to U8_MAX.
2678			 *
2679			 * smb_direct_accept_client() will then
2680			 * do the real negotiation in order to
2681			 * select the minimum between client and
2682			 * server.
2683			 */
2684			ird32 = min_t(u32, ird32, U8_MAX);
2685			ord32 = min_t(u32, ord32, U8_MAX);
2686
2687			sc->rdma.legacy_iwarp = true;
2688			peer_initiator_depth = (u8)ird32;
2689			peer_responder_resources = (u8)ord32;
2690		}
2691	}
2692
2693	/*
2694	 * First set what the we as server are able to support
2695	 */
2696	sp->initiator_depth = min_t(u8, sp->initiator_depth,
2697				   new_cm_id->device->attrs.max_qp_rd_atom);
2698
2699	/*
2700	 * negotiate the value by using the minimum
2701	 * between client and server if the client provided
2702	 * non 0 values.
2703	 */
2704	if (peer_initiator_depth != 0)
2705		sp->initiator_depth = min_t(u8, sp->initiator_depth,
2706					   peer_initiator_depth);
2707	if (peer_responder_resources != 0)
2708		sp->responder_resources = min_t(u8, sp->responder_resources,
2709					       peer_responder_resources);
2710
2711	ret = smb_direct_connect(sc);
2712	if (ret)
2713		goto out_err;
2714
2715	handler = kthread_run(ksmbd_conn_handler_loop,
2716			      KSMBD_TRANS(t)->conn, "ksmbd:r%u",
2717			      smb_direct_port);
2718	if (IS_ERR(handler)) {
2719		ret = PTR_ERR(handler);
2720		pr_err("Can't start thread\n");
2721		goto out_err;
2722	}
2723
2724	return 0;
2725out_err:
2726	free_transport(t);
2727	return ret;
2728}
2729
2730static int smb_direct_listen_handler(struct rdma_cm_id *cm_id,
2731				     struct rdma_cm_event *event)
2732{
2733	switch (event->event) {
2734	case RDMA_CM_EVENT_CONNECT_REQUEST: {
2735		int ret = smb_direct_handle_connect_request(cm_id, event);
2736
2737		if (ret) {
2738			pr_err("Can't create transport: %d\n", ret);
2739			return ret;
2740		}
2741
2742		ksmbd_debug(RDMA, "Received connection request. cm_id=%p\n",
2743			    cm_id);
2744		break;
2745	}
2746	default:
2747		pr_err("Unexpected listen event. cm_id=%p, event=%s (%d)\n",
2748		       cm_id, rdma_event_msg(event->event), event->event);
2749		break;
2750	}
2751	return 0;
2752}
2753
2754static int smb_direct_listen(int port)
2755{
2756	int ret;
2757	struct rdma_cm_id *cm_id;
2758	struct sockaddr_in sin = {
2759		.sin_family		= AF_INET,
2760		.sin_addr.s_addr	= htonl(INADDR_ANY),
2761		.sin_port		= htons(port),
2762	};
2763
2764	cm_id = rdma_create_id(&init_net, smb_direct_listen_handler,
2765			       &smb_direct_listener, RDMA_PS_TCP, IB_QPT_RC);
2766	if (IS_ERR(cm_id)) {
2767		pr_err("Can't create cm id: %ld\n", PTR_ERR(cm_id));
2768		return PTR_ERR(cm_id);
2769	}
2770
2771	ret = rdma_bind_addr(cm_id, (struct sockaddr *)&sin);
2772	if (ret) {
2773		pr_err("Can't bind: %d\n", ret);
2774		goto err;
2775	}
2776
2777	smb_direct_listener.cm_id = cm_id;
2778
2779	ret = rdma_listen(cm_id, 10);
2780	if (ret) {
2781		pr_err("Can't listen: %d\n", ret);
2782		goto err;
2783	}
2784	return 0;
2785err:
2786	smb_direct_listener.cm_id = NULL;
2787	rdma_destroy_id(cm_id);
2788	return ret;
2789}
2790
2791static int smb_direct_ib_client_add(struct ib_device *ib_dev)
2792{
2793	struct smb_direct_device *smb_dev;
2794
2795	/* Set 5445 port if device type is iWARP(No IB) */
2796	if (ib_dev->node_type != RDMA_NODE_IB_CA)
2797		smb_direct_port = SMB_DIRECT_PORT_IWARP;
2798
2799	if (!rdma_frwr_is_supported(&ib_dev->attrs))
2800		return 0;
2801
2802	smb_dev = kzalloc(sizeof(*smb_dev), KSMBD_DEFAULT_GFP);
2803	if (!smb_dev)
2804		return -ENOMEM;
2805	smb_dev->ib_dev = ib_dev;
2806
2807	write_lock(&smb_direct_device_lock);
2808	list_add(&smb_dev->list, &smb_direct_device_list);
2809	write_unlock(&smb_direct_device_lock);
2810
2811	ksmbd_debug(RDMA, "ib device added: name %s\n", ib_dev->name);
2812	return 0;
2813}
2814
2815static void smb_direct_ib_client_remove(struct ib_device *ib_dev,
2816					void *client_data)
2817{
2818	struct smb_direct_device *smb_dev, *tmp;
2819
2820	write_lock(&smb_direct_device_lock);
2821	list_for_each_entry_safe(smb_dev, tmp, &smb_direct_device_list, list) {
2822		if (smb_dev->ib_dev == ib_dev) {
2823			list_del(&smb_dev->list);
2824			kfree(smb_dev);
2825			break;
2826		}
2827	}
2828	write_unlock(&smb_direct_device_lock);
2829}
2830
2831static struct ib_client smb_direct_ib_client = {
2832	.name	= "ksmbd_smb_direct_ib",
2833	.add	= smb_direct_ib_client_add,
2834	.remove	= smb_direct_ib_client_remove,
2835};
2836
2837int ksmbd_rdma_init(void)
2838{
2839	int ret;
2840
2841	smb_direct_port = SMB_DIRECT_PORT_INFINIBAND;
2842	smb_direct_listener.cm_id = NULL;
2843
2844	ret = ib_register_client(&smb_direct_ib_client);
2845	if (ret) {
2846		pr_err("failed to ib_register_client\n");
2847		return ret;
2848	}
2849
2850	/* When a client is running out of send credits, the credits are
2851	 * granted by the server's sending a packet using this queue.
2852	 * This avoids the situation that a clients cannot send packets
2853	 * for lack of credits
2854	 */
2855	smb_direct_wq = alloc_workqueue("ksmbd-smb_direct-wq",
2856					WQ_HIGHPRI | WQ_MEM_RECLAIM | WQ_PERCPU,
2857					0);
2858	if (!smb_direct_wq)
2859		return -ENOMEM;
2860
2861	ret = smb_direct_listen(smb_direct_port);
2862	if (ret) {
2863		destroy_workqueue(smb_direct_wq);
2864		smb_direct_wq = NULL;
2865		pr_err("Can't listen: %d\n", ret);
2866		return ret;
2867	}
2868
2869	ksmbd_debug(RDMA, "init RDMA listener. cm_id=%p\n",
2870		    smb_direct_listener.cm_id);
2871	return 0;
2872}
2873
2874void ksmbd_rdma_stop_listening(void)
2875{
2876	if (!smb_direct_listener.cm_id)
2877		return;
2878
2879	ib_unregister_client(&smb_direct_ib_client);
2880	rdma_destroy_id(smb_direct_listener.cm_id);
2881
2882	smb_direct_listener.cm_id = NULL;
2883}
2884
2885void ksmbd_rdma_destroy(void)
2886{
2887	if (smb_direct_wq) {
2888		destroy_workqueue(smb_direct_wq);
2889		smb_direct_wq = NULL;
2890	}
2891}
2892
2893static bool ksmbd_find_rdma_capable_netdev(struct net_device *netdev)
2894{
2895	struct smb_direct_device *smb_dev;
2896	int i;
2897	bool rdma_capable = false;
2898
2899	read_lock(&smb_direct_device_lock);
2900	list_for_each_entry(smb_dev, &smb_direct_device_list, list) {
2901		for (i = 0; i < smb_dev->ib_dev->phys_port_cnt; i++) {
2902			struct net_device *ndev;
2903
2904			ndev = ib_device_get_netdev(smb_dev->ib_dev, i + 1);
2905			if (!ndev)
2906				continue;
2907
2908			if (ndev == netdev) {
2909				dev_put(ndev);
2910				rdma_capable = true;
2911				goto out;
2912			}
2913			dev_put(ndev);
2914		}
2915	}
2916out:
2917	read_unlock(&smb_direct_device_lock);
2918
2919	if (rdma_capable == false) {
2920		struct ib_device *ibdev;
2921
2922		ibdev = ib_device_get_by_netdev(netdev, RDMA_DRIVER_UNKNOWN);
2923		if (ibdev) {
2924			rdma_capable = rdma_frwr_is_supported(&ibdev->attrs);
2925			ib_device_put(ibdev);
2926		}
2927	}
2928
2929	ksmbd_debug(RDMA, "netdev(%s) rdma capable : %s\n",
2930		    netdev->name, str_true_false(rdma_capable));
2931
2932	return rdma_capable;
2933}
2934
2935bool ksmbd_rdma_capable_netdev(struct net_device *netdev)
2936{
2937	struct net_device *lower_dev;
2938	struct list_head *iter;
2939
2940	if (ksmbd_find_rdma_capable_netdev(netdev))
2941		return true;
2942
2943	/* check if netdev is bridge or VLAN */
2944	if (netif_is_bridge_master(netdev) ||
2945	    netdev->priv_flags & IFF_802_1Q_VLAN)
2946		netdev_for_each_lower_dev(netdev, lower_dev, iter)
2947			if (ksmbd_find_rdma_capable_netdev(lower_dev))
2948				return true;
2949
2950	/* check if netdev is IPoIB safely without layer violation */
2951	if (netdev->type == ARPHRD_INFINIBAND)
2952		return true;
2953
2954	return false;
2955}
2956
2957static const struct ksmbd_transport_ops ksmbd_smb_direct_transport_ops = {
2958	.prepare	= smb_direct_prepare,
2959	.disconnect	= smb_direct_disconnect,
2960	.shutdown	= smb_direct_shutdown,
2961	.writev		= smb_direct_writev,
2962	.read		= smb_direct_read,
2963	.rdma_read	= smb_direct_rdma_read,
2964	.rdma_write	= smb_direct_rdma_write,
2965	.free_transport = smb_direct_free_transport,
2966};