smbdirect.c

   1// SPDX-License-Identifier: GPL-2.0-or-later
   2/*
   3 *   Copyright (C) 2017, Microsoft Corporation.
   4 *
   5 *   Author(s): Long Li <longli@microsoft.com>
   6 */
   7#include <linux/module.h>
   8#include <linux/highmem.h>
   9#include <linux/folio_queue.h>
  10#define __SMBDIRECT_SOCKET_DISCONNECT(__sc) smbd_disconnect_rdma_connection(__sc)
  11#include "../common/smbdirect/smbdirect_pdu.h"
  12#include "smbdirect.h"
  13#include "cifs_debug.h"
  14#include "cifsproto.h"
  15#include "smb2proto.h"
  16
  17const struct smbdirect_socket_parameters *smbd_get_parameters(struct smbd_connection *conn)
  18{
  19	struct smbdirect_socket *sc = &conn->socket;
  20
  21	return &sc->parameters;
  22}
  23
  24static struct smbdirect_recv_io *get_receive_buffer(
  25		struct smbdirect_socket *sc);
  26static void put_receive_buffer(
  27		struct smbdirect_socket *sc,
  28		struct smbdirect_recv_io *response);
  29static int allocate_receive_buffers(struct smbdirect_socket *sc, int num_buf);
  30static void destroy_receive_buffers(struct smbdirect_socket *sc);
  31
  32static void enqueue_reassembly(
  33		struct smbdirect_socket *sc,
  34		struct smbdirect_recv_io *response, int data_length);
  35static struct smbdirect_recv_io *_get_first_reassembly(
  36		struct smbdirect_socket *sc);
  37
  38static int smbd_post_send(struct smbdirect_socket *sc,
  39			  struct smbdirect_send_batch *batch,
  40			  struct smbdirect_send_io *request);
  41
  42static int smbd_post_recv(
  43		struct smbdirect_socket *sc,
  44		struct smbdirect_recv_io *response);
  45
  46static int smbd_post_send_empty(struct smbdirect_socket *sc);
  47
  48static void destroy_mr_list(struct smbdirect_socket *sc);
  49static int allocate_mr_list(struct smbdirect_socket *sc);
  50
  51struct smb_extract_to_rdma {
  52	struct ib_sge		*sge;
  53	unsigned int		nr_sge;
  54	unsigned int		max_sge;
  55	struct ib_device	*device;
  56	u32			local_dma_lkey;
  57	enum dma_data_direction	direction;
  58};
  59static ssize_t smb_extract_iter_to_rdma(struct iov_iter *iter, size_t len,
  60					struct smb_extract_to_rdma *rdma);
  61
  62/* Port numbers for SMBD transport */
  63#define SMB_PORT	445
  64#define SMBD_PORT	5445
  65
  66/* Address lookup and resolve timeout in ms */
  67#define RDMA_RESOLVE_TIMEOUT	5000
  68
  69/* SMBD negotiation timeout in seconds */
  70#define SMBD_NEGOTIATE_TIMEOUT	120
  71
  72/* The timeout to wait for a keepalive message from peer in seconds */
  73#define KEEPALIVE_RECV_TIMEOUT 5
  74
  75/* SMBD minimum receive size and fragmented sized defined in [MS-SMBD] */
  76#define SMBD_MIN_RECEIVE_SIZE		128
  77#define SMBD_MIN_FRAGMENTED_SIZE	131072
  78
  79/*
  80 * Default maximum number of RDMA read/write outstanding on this connection
  81 * This value is possibly decreased during QP creation on hardware limit
  82 */
  83#define SMBD_CM_RESPONDER_RESOURCES	32
  84
  85/* Maximum number of retries on data transfer operations */
  86#define SMBD_CM_RETRY			6
  87/* No need to retry on Receiver Not Ready since SMBD manages credits */
  88#define SMBD_CM_RNR_RETRY		0
  89
  90/*
  91 * User configurable initial values per SMBD transport connection
  92 * as defined in [MS-SMBD] 3.1.1.1
  93 * Those may change after a SMBD negotiation
  94 */
  95/* The local peer's maximum number of credits to grant to the peer */
  96int smbd_receive_credit_max = 255;
  97
  98/* The remote peer's credit request of local peer */
  99int smbd_send_credit_target = 255;
 100
 101/* The maximum single message size can be sent to remote peer */
 102int smbd_max_send_size = 1364;
 103
 104/*
 105 * The maximum fragmented upper-layer payload receive size supported
 106 *
 107 * Assume max_payload_per_credit is
 108 * smbd_max_receive_size - 24 = 1340
 109 *
 110 * The maximum number would be
 111 * smbd_receive_credit_max * max_payload_per_credit
 112 *
 113 *                       1340 * 255 = 341700 (0x536C4)
 114 *
 115 * The minimum value from the spec is 131072 (0x20000)
 116 *
 117 * For now we use the logic we used in ksmbd before:
 118 *                 (1364 * 255) / 2 = 173910 (0x2A756)
 119 */
 120int smbd_max_fragmented_recv_size = (1364 * 255) / 2;
 121
 122/*  The maximum single-message size which can be received */
 123int smbd_max_receive_size = 1364;
 124
 125/* The timeout to initiate send of a keepalive message on idle */
 126int smbd_keep_alive_interval = 120;
 127
 128/*
 129 * User configurable initial values for RDMA transport
 130 * The actual values used may be lower and are limited to hardware capabilities
 131 */
 132/* Default maximum number of pages in a single RDMA write/read */
 133int smbd_max_frmr_depth = 2048;
 134
 135/* If payload is less than this byte, use RDMA send/recv not read/write */
 136int rdma_readwrite_threshold = 4096;
 137
 138/* Transport logging functions
 139 * Logging are defined as classes. They can be OR'ed to define the actual
 140 * logging level via module parameter smbd_logging_class
 141 * e.g. cifs.smbd_logging_class=0xa0 will log all log_rdma_recv() and
 142 * log_rdma_event()
 143 */
 144#define LOG_OUTGOING			0x1
 145#define LOG_INCOMING			0x2
 146#define LOG_READ			0x4
 147#define LOG_WRITE			0x8
 148#define LOG_RDMA_SEND			0x10
 149#define LOG_RDMA_RECV			0x20
 150#define LOG_KEEP_ALIVE			0x40
 151#define LOG_RDMA_EVENT			0x80
 152#define LOG_RDMA_MR			0x100
 153static unsigned int smbd_logging_class;
 154module_param(smbd_logging_class, uint, 0644);
 155MODULE_PARM_DESC(smbd_logging_class,
 156	"Logging class for SMBD transport 0x0 to 0x100");
 157
 158#define ERR		0x0
 159#define INFO		0x1
 160static unsigned int smbd_logging_level = ERR;
 161module_param(smbd_logging_level, uint, 0644);
 162MODULE_PARM_DESC(smbd_logging_level,
 163	"Logging level for SMBD transport, 0 (default): error, 1: info");
 164
 165#define log_rdma(level, class, fmt, args...)				\
 166do {									\
 167	if (level <= smbd_logging_level || class & smbd_logging_class)	\
 168		cifs_dbg(VFS, "%s:%d " fmt, __func__, __LINE__, ##args);\
 169} while (0)
 170
 171#define log_outgoing(level, fmt, args...) \
 172		log_rdma(level, LOG_OUTGOING, fmt, ##args)
 173#define log_incoming(level, fmt, args...) \
 174		log_rdma(level, LOG_INCOMING, fmt, ##args)
 175#define log_read(level, fmt, args...)	log_rdma(level, LOG_READ, fmt, ##args)
 176#define log_write(level, fmt, args...)	log_rdma(level, LOG_WRITE, fmt, ##args)
 177#define log_rdma_send(level, fmt, args...) \
 178		log_rdma(level, LOG_RDMA_SEND, fmt, ##args)
 179#define log_rdma_recv(level, fmt, args...) \
 180		log_rdma(level, LOG_RDMA_RECV, fmt, ##args)
 181#define log_keep_alive(level, fmt, args...) \
 182		log_rdma(level, LOG_KEEP_ALIVE, fmt, ##args)
 183#define log_rdma_event(level, fmt, args...) \
 184		log_rdma(level, LOG_RDMA_EVENT, fmt, ##args)
 185#define log_rdma_mr(level, fmt, args...) \
 186		log_rdma(level, LOG_RDMA_MR, fmt, ##args)
 187
 188static void smbd_disconnect_wake_up_all(struct smbdirect_socket *sc)
 189{
 190	/*
 191	 * Wake up all waiters in all wait queues
 192	 * in order to notice the broken connection.
 193	 */
 194	wake_up_all(&sc->status_wait);
 195	wake_up_all(&sc->send_io.lcredits.wait_queue);
 196	wake_up_all(&sc->send_io.credits.wait_queue);
 197	wake_up_all(&sc->send_io.pending.dec_wait_queue);
 198	wake_up_all(&sc->send_io.pending.zero_wait_queue);
 199	wake_up_all(&sc->recv_io.reassembly.wait_queue);
 200	wake_up_all(&sc->mr_io.ready.wait_queue);
 201	wake_up_all(&sc->mr_io.cleanup.wait_queue);
 202}
 203
 204static void smbd_disconnect_rdma_work(struct work_struct *work)
 205{
 206	struct smbdirect_socket *sc =
 207		container_of(work, struct smbdirect_socket, disconnect_work);
 208
 209	if (sc->first_error == 0)
 210		sc->first_error = -ECONNABORTED;
 211
 212	/*
 213	 * make sure this and other work is not queued again
 214	 * but here we don't block and avoid
 215	 * disable[_delayed]_work_sync()
 216	 */
 217	disable_work(&sc->disconnect_work);
 218	disable_work(&sc->recv_io.posted.refill_work);
 219	disable_work(&sc->mr_io.recovery_work);
 220	disable_work(&sc->idle.immediate_work);
 221	disable_delayed_work(&sc->idle.timer_work);
 222
 223	switch (sc->status) {
 224	case SMBDIRECT_SOCKET_NEGOTIATE_NEEDED:
 225	case SMBDIRECT_SOCKET_NEGOTIATE_RUNNING:
 226	case SMBDIRECT_SOCKET_NEGOTIATE_FAILED:
 227	case SMBDIRECT_SOCKET_CONNECTED:
 228	case SMBDIRECT_SOCKET_ERROR:
 229		sc->status = SMBDIRECT_SOCKET_DISCONNECTING;
 230		rdma_disconnect(sc->rdma.cm_id);
 231		break;
 232
 233	case SMBDIRECT_SOCKET_CREATED:
 234	case SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED:
 235	case SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING:
 236	case SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED:
 237	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED:
 238	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING:
 239	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED:
 240	case SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED:
 241	case SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING:
 242	case SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED:
 243		/*
 244		 * rdma_connect() never reached
 245		 * RDMA_CM_EVENT_ESTABLISHED
 246		 */
 247		sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
 248		break;
 249
 250	case SMBDIRECT_SOCKET_DISCONNECTING:
 251	case SMBDIRECT_SOCKET_DISCONNECTED:
 252	case SMBDIRECT_SOCKET_DESTROYED:
 253		break;
 254	}
 255
 256	/*
 257	 * Wake up all waiters in all wait queues
 258	 * in order to notice the broken connection.
 259	 */
 260	smbd_disconnect_wake_up_all(sc);
 261}
 262
 263static void smbd_disconnect_rdma_connection(struct smbdirect_socket *sc)
 264{
 265	if (sc->first_error == 0)
 266		sc->first_error = -ECONNABORTED;
 267
 268	/*
 269	 * make sure other work (than disconnect_work) is
 270	 * not queued again but here we don't block and avoid
 271	 * disable[_delayed]_work_sync()
 272	 */
 273	disable_work(&sc->recv_io.posted.refill_work);
 274	disable_work(&sc->mr_io.recovery_work);
 275	disable_work(&sc->idle.immediate_work);
 276	disable_delayed_work(&sc->idle.timer_work);
 277
 278	switch (sc->status) {
 279	case SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED:
 280	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED:
 281	case SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED:
 282	case SMBDIRECT_SOCKET_NEGOTIATE_FAILED:
 283	case SMBDIRECT_SOCKET_ERROR:
 284	case SMBDIRECT_SOCKET_DISCONNECTING:
 285	case SMBDIRECT_SOCKET_DISCONNECTED:
 286	case SMBDIRECT_SOCKET_DESTROYED:
 287		/*
 288		 * Keep the current error status
 289		 */
 290		break;
 291
 292	case SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED:
 293	case SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING:
 294		sc->status = SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED;
 295		break;
 296
 297	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED:
 298	case SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING:
 299		sc->status = SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED;
 300		break;
 301
 302	case SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED:
 303	case SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING:
 304		sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED;
 305		break;
 306
 307	case SMBDIRECT_SOCKET_NEGOTIATE_NEEDED:
 308	case SMBDIRECT_SOCKET_NEGOTIATE_RUNNING:
 309		sc->status = SMBDIRECT_SOCKET_NEGOTIATE_FAILED;
 310		break;
 311
 312	case SMBDIRECT_SOCKET_CREATED:
 313		sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
 314		break;
 315
 316	case SMBDIRECT_SOCKET_CONNECTED:
 317		sc->status = SMBDIRECT_SOCKET_ERROR;
 318		break;
 319	}
 320
 321	/*
 322	 * Wake up all waiters in all wait queues
 323	 * in order to notice the broken connection.
 324	 */
 325	smbd_disconnect_wake_up_all(sc);
 326
 327	queue_work(sc->workqueue, &sc->disconnect_work);
 328}
 329
 330/* Upcall from RDMA CM */
 331static int smbd_conn_upcall(
 332		struct rdma_cm_id *id, struct rdma_cm_event *event)
 333{
 334	struct smbdirect_socket *sc = id->context;
 335	struct smbdirect_socket_parameters *sp = &sc->parameters;
 336	const char *event_name = rdma_event_msg(event->event);
 337	u8 peer_initiator_depth;
 338	u8 peer_responder_resources;
 339
 340	log_rdma_event(INFO, "event=%s status=%d\n",
 341		event_name, event->status);
 342
 343	switch (event->event) {
 344	case RDMA_CM_EVENT_ADDR_RESOLVED:
 345		if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING))
 346			break;
 347		sc->status = SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED;
 348		wake_up(&sc->status_wait);
 349		break;
 350
 351	case RDMA_CM_EVENT_ROUTE_RESOLVED:
 352		if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING))
 353			break;
 354		sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED;
 355		wake_up(&sc->status_wait);
 356		break;
 357
 358	case RDMA_CM_EVENT_ADDR_ERROR:
 359		log_rdma_event(ERR, "connecting failed event=%s\n", event_name);
 360		sc->status = SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED;
 361		smbd_disconnect_rdma_work(&sc->disconnect_work);
 362		break;
 363
 364	case RDMA_CM_EVENT_ROUTE_ERROR:
 365		log_rdma_event(ERR, "connecting failed event=%s\n", event_name);
 366		sc->status = SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED;
 367		smbd_disconnect_rdma_work(&sc->disconnect_work);
 368		break;
 369
 370	case RDMA_CM_EVENT_ESTABLISHED:
 371		log_rdma_event(INFO, "connected event=%s\n", event_name);
 372
 373		/*
 374		 * Here we work around an inconsistency between
 375		 * iWarp and other devices (at least rxe and irdma using RoCEv2)
 376		 */
 377		if (rdma_protocol_iwarp(id->device, id->port_num)) {
 378			/*
 379			 * iWarp devices report the peer's values
 380			 * with the perspective of the peer here.
 381			 * Tested with siw and irdma (in iwarp mode)
 382			 * We need to change to our perspective here,
 383			 * so we need to switch the values.
 384			 */
 385			peer_initiator_depth = event->param.conn.responder_resources;
 386			peer_responder_resources = event->param.conn.initiator_depth;
 387		} else {
 388			/*
 389			 * Non iWarp devices report the peer's values
 390			 * already changed to our perspective here.
 391			 * Tested with rxe and irdma (in roce mode).
 392			 */
 393			peer_initiator_depth = event->param.conn.initiator_depth;
 394			peer_responder_resources = event->param.conn.responder_resources;
 395		}
 396		if (rdma_protocol_iwarp(id->device, id->port_num) &&
 397		    event->param.conn.private_data_len == 8) {
 398			/*
 399			 * Legacy clients with only iWarp MPA v1 support
 400			 * need a private blob in order to negotiate
 401			 * the IRD/ORD values.
 402			 */
 403			const __be32 *ird_ord_hdr = event->param.conn.private_data;
 404			u32 ird32 = be32_to_cpu(ird_ord_hdr[0]);
 405			u32 ord32 = be32_to_cpu(ird_ord_hdr[1]);
 406
 407			/*
 408			 * cifs.ko sends the legacy IRD/ORD negotiation
 409			 * event if iWarp MPA v2 was used.
 410			 *
 411			 * Here we check that the values match and only
 412			 * mark the client as legacy if they don't match.
 413			 */
 414			if ((u32)event->param.conn.initiator_depth != ird32 ||
 415			    (u32)event->param.conn.responder_resources != ord32) {
 416				/*
 417				 * There are broken clients (old cifs.ko)
 418				 * using little endian and also
 419				 * struct rdma_conn_param only uses u8
 420				 * for initiator_depth and responder_resources,
 421				 * so we truncate the value to U8_MAX.
 422				 *
 423				 * smb_direct_accept_client() will then
 424				 * do the real negotiation in order to
 425				 * select the minimum between client and
 426				 * server.
 427				 */
 428				ird32 = min_t(u32, ird32, U8_MAX);
 429				ord32 = min_t(u32, ord32, U8_MAX);
 430
 431				sc->rdma.legacy_iwarp = true;
 432				peer_initiator_depth = (u8)ird32;
 433				peer_responder_resources = (u8)ord32;
 434			}
 435		}
 436
 437		/*
 438		 * negotiate the value by using the minimum
 439		 * between client and server if the client provided
 440		 * non 0 values.
 441		 */
 442		if (peer_initiator_depth != 0)
 443			sp->initiator_depth =
 444					min_t(u8, sp->initiator_depth,
 445					      peer_initiator_depth);
 446		if (peer_responder_resources != 0)
 447			sp->responder_resources =
 448					min_t(u8, sp->responder_resources,
 449					      peer_responder_resources);
 450
 451		if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING))
 452			break;
 453		sc->status = SMBDIRECT_SOCKET_NEGOTIATE_NEEDED;
 454		wake_up(&sc->status_wait);
 455		break;
 456
 457	case RDMA_CM_EVENT_CONNECT_ERROR:
 458	case RDMA_CM_EVENT_UNREACHABLE:
 459	case RDMA_CM_EVENT_REJECTED:
 460		log_rdma_event(ERR, "connecting failed event=%s\n", event_name);
 461		sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED;
 462		smbd_disconnect_rdma_work(&sc->disconnect_work);
 463		break;
 464
 465	case RDMA_CM_EVENT_DEVICE_REMOVAL:
 466	case RDMA_CM_EVENT_DISCONNECTED:
 467		/* This happens when we fail the negotiation */
 468		if (sc->status == SMBDIRECT_SOCKET_NEGOTIATE_FAILED) {
 469			log_rdma_event(ERR, "event=%s during negotiation\n", event_name);
 470		}
 471
 472		sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
 473		smbd_disconnect_rdma_work(&sc->disconnect_work);
 474		break;
 475
 476	default:
 477		log_rdma_event(ERR, "unexpected event=%s status=%d\n",
 478			       event_name, event->status);
 479		break;
 480	}
 481
 482	return 0;
 483}
 484
 485/* Upcall from RDMA QP */
 486static void
 487smbd_qp_async_error_upcall(struct ib_event *event, void *context)
 488{
 489	struct smbdirect_socket *sc = context;
 490
 491	log_rdma_event(ERR, "%s on device %s socket %p\n",
 492		ib_event_msg(event->event), event->device->name, sc);
 493
 494	switch (event->event) {
 495	case IB_EVENT_CQ_ERR:
 496	case IB_EVENT_QP_FATAL:
 497		smbd_disconnect_rdma_connection(sc);
 498		break;
 499
 500	default:
 501		break;
 502	}
 503}
 504
 505static inline void *smbdirect_send_io_payload(struct smbdirect_send_io *request)
 506{
 507	return (void *)request->packet;
 508}
 509
 510static inline void *smbdirect_recv_io_payload(struct smbdirect_recv_io *response)
 511{
 512	return (void *)response->packet;
 513}
 514
 515static struct smbdirect_send_io *smbd_alloc_send_io(struct smbdirect_socket *sc)
 516{
 517	struct smbdirect_send_io *msg;
 518
 519	msg = mempool_alloc(sc->send_io.mem.pool, GFP_KERNEL);
 520	if (!msg)
 521		return ERR_PTR(-ENOMEM);
 522	msg->socket = sc;
 523	INIT_LIST_HEAD(&msg->sibling_list);
 524	msg->num_sge = 0;
 525
 526	return msg;
 527}
 528
 529static void smbd_free_send_io(struct smbdirect_send_io *msg)
 530{
 531	struct smbdirect_socket *sc = msg->socket;
 532	size_t i;
 533
 534	/*
 535	 * The list needs to be empty!
 536	 * The caller should take care of it.
 537	 */
 538	WARN_ON_ONCE(!list_empty(&msg->sibling_list));
 539
 540	/*
 541	 * Note we call ib_dma_unmap_page(), even if some sges are mapped using
 542	 * ib_dma_map_single().
 543	 *
 544	 * The difference between _single() and _page() only matters for the
 545	 * ib_dma_map_*() case.
 546	 *
 547	 * For the ib_dma_unmap_*() case it does not matter as both take the
 548	 * dma_addr_t and dma_unmap_single_attrs() is just an alias to
 549	 * dma_unmap_page_attrs().
 550	 */
 551	for (i = 0; i < msg->num_sge; i++)
 552		ib_dma_unmap_page(sc->ib.dev,
 553				  msg->sge[i].addr,
 554				  msg->sge[i].length,
 555				  DMA_TO_DEVICE);
 556
 557	mempool_free(msg, sc->send_io.mem.pool);
 558}
 559
 560/* Called when a RDMA send is done */
 561static void send_done(struct ib_cq *cq, struct ib_wc *wc)
 562{
 563	struct smbdirect_send_io *request =
 564		container_of(wc->wr_cqe, struct smbdirect_send_io, cqe);
 565	struct smbdirect_socket *sc = request->socket;
 566	struct smbdirect_send_io *sibling, *next;
 567	int lcredits = 0;
 568
 569	log_rdma_send(INFO, "smbdirect_send_io 0x%p completed wc->status=%s\n",
 570		request, ib_wc_status_msg(wc->status));
 571
 572	if (unlikely(!(request->wr.send_flags & IB_SEND_SIGNALED))) {
 573		/*
 574		 * This happens when smbdirect_send_io is a sibling
 575		 * before the final message, it is signaled on
 576		 * error anyway, so we need to skip
 577		 * smbdirect_connection_free_send_io here,
 578		 * otherwise is will destroy the memory
 579		 * of the siblings too, which will cause
 580		 * use after free problems for the others
 581		 * triggered from ib_drain_qp().
 582		 */
 583		if (wc->status != IB_WC_SUCCESS)
 584			goto skip_free;
 585
 586		/*
 587		 * This should not happen!
 588		 * But we better just close the
 589		 * connection...
 590		 */
 591		log_rdma_send(ERR,
 592			"unexpected send completion wc->status=%s (%d) wc->opcode=%d\n",
 593			ib_wc_status_msg(wc->status), wc->status, wc->opcode);
 594		smbd_disconnect_rdma_connection(sc);
 595		return;
 596	}
 597
 598	/*
 599	 * Free possible siblings and then the main send_io
 600	 */
 601	list_for_each_entry_safe(sibling, next, &request->sibling_list, sibling_list) {
 602		list_del_init(&sibling->sibling_list);
 603		smbd_free_send_io(sibling);
 604		lcredits += 1;
 605	}
 606	/* Note this frees wc->wr_cqe, but not wc */
 607	smbd_free_send_io(request);
 608	lcredits += 1;
 609
 610	if (wc->status != IB_WC_SUCCESS || wc->opcode != IB_WC_SEND) {
 611skip_free:
 612		if (wc->status != IB_WC_WR_FLUSH_ERR)
 613			log_rdma_send(ERR, "wc->status=%s wc->opcode=%d\n",
 614				ib_wc_status_msg(wc->status), wc->opcode);
 615		smbd_disconnect_rdma_connection(sc);
 616		return;
 617	}
 618
 619	atomic_add(lcredits, &sc->send_io.lcredits.count);
 620	wake_up(&sc->send_io.lcredits.wait_queue);
 621
 622	if (atomic_dec_and_test(&sc->send_io.pending.count))
 623		wake_up(&sc->send_io.pending.zero_wait_queue);
 624
 625	wake_up(&sc->send_io.pending.dec_wait_queue);
 626}
 627
 628static void dump_smbdirect_negotiate_resp(struct smbdirect_negotiate_resp *resp)
 629{
 630	log_rdma_event(INFO, "resp message min_version %u max_version %u negotiated_version %u credits_requested %u credits_granted %u status %u max_readwrite_size %u preferred_send_size %u max_receive_size %u max_fragmented_size %u\n",
 631		       resp->min_version, resp->max_version,
 632		       resp->negotiated_version, resp->credits_requested,
 633		       resp->credits_granted, resp->status,
 634		       resp->max_readwrite_size, resp->preferred_send_size,
 635		       resp->max_receive_size, resp->max_fragmented_size);
 636}
 637
 638/*
 639 * Process a negotiation response message, according to [MS-SMBD]3.1.5.7
 640 * response, packet_length: the negotiation response message
 641 * return value: true if negotiation is a success, false if failed
 642 */
 643static bool process_negotiation_response(
 644		struct smbdirect_recv_io *response, int packet_length)
 645{
 646	struct smbdirect_socket *sc = response->socket;
 647	struct smbdirect_socket_parameters *sp = &sc->parameters;
 648	struct smbdirect_negotiate_resp *packet = smbdirect_recv_io_payload(response);
 649
 650	if (packet_length < sizeof(struct smbdirect_negotiate_resp)) {
 651		log_rdma_event(ERR,
 652			"error: packet_length=%d\n", packet_length);
 653		return false;
 654	}
 655
 656	if (le16_to_cpu(packet->negotiated_version) != SMBDIRECT_V1) {
 657		log_rdma_event(ERR, "error: negotiated_version=%x\n",
 658			le16_to_cpu(packet->negotiated_version));
 659		return false;
 660	}
 661
 662	if (packet->credits_requested == 0) {
 663		log_rdma_event(ERR, "error: credits_requested==0\n");
 664		return false;
 665	}
 666	sc->recv_io.credits.target = le16_to_cpu(packet->credits_requested);
 667	sc->recv_io.credits.target = min_t(u16, sc->recv_io.credits.target, sp->recv_credit_max);
 668
 669	if (packet->credits_granted == 0) {
 670		log_rdma_event(ERR, "error: credits_granted==0\n");
 671		return false;
 672	}
 673	atomic_set(&sc->send_io.lcredits.count, sp->send_credit_target);
 674	atomic_set(&sc->send_io.credits.count, le16_to_cpu(packet->credits_granted));
 675
 676	if (le32_to_cpu(packet->preferred_send_size) > sp->max_recv_size) {
 677		log_rdma_event(ERR, "error: preferred_send_size=%d\n",
 678			le32_to_cpu(packet->preferred_send_size));
 679		return false;
 680	}
 681	sp->max_recv_size = le32_to_cpu(packet->preferred_send_size);
 682
 683	if (le32_to_cpu(packet->max_receive_size) < SMBD_MIN_RECEIVE_SIZE) {
 684		log_rdma_event(ERR, "error: max_receive_size=%d\n",
 685			le32_to_cpu(packet->max_receive_size));
 686		return false;
 687	}
 688	sp->max_send_size = min_t(u32, sp->max_send_size,
 689				  le32_to_cpu(packet->max_receive_size));
 690
 691	if (le32_to_cpu(packet->max_fragmented_size) <
 692			SMBD_MIN_FRAGMENTED_SIZE) {
 693		log_rdma_event(ERR, "error: max_fragmented_size=%d\n",
 694			le32_to_cpu(packet->max_fragmented_size));
 695		return false;
 696	}
 697	sp->max_fragmented_send_size =
 698		le32_to_cpu(packet->max_fragmented_size);
 699
 700
 701	sp->max_read_write_size = min_t(u32,
 702			le32_to_cpu(packet->max_readwrite_size),
 703			sp->max_frmr_depth * PAGE_SIZE);
 704	sp->max_frmr_depth = sp->max_read_write_size / PAGE_SIZE;
 705
 706	atomic_set(&sc->send_io.bcredits.count, 1);
 707	sc->recv_io.expected = SMBDIRECT_EXPECT_DATA_TRANSFER;
 708	return true;
 709}
 710
 711static void smbd_post_send_credits(struct work_struct *work)
 712{
 713	int rc;
 714	struct smbdirect_recv_io *response;
 715	struct smbdirect_socket *sc =
 716		container_of(work, struct smbdirect_socket, recv_io.posted.refill_work);
 717	int posted = 0;
 718
 719	if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
 720		return;
 721	}
 722
 723	if (sc->recv_io.credits.target >
 724		atomic_read(&sc->recv_io.credits.count)) {
 725		while (true) {
 726			response = get_receive_buffer(sc);
 727			if (!response)
 728				break;
 729
 730			response->first_segment = false;
 731			rc = smbd_post_recv(sc, response);
 732			if (rc) {
 733				log_rdma_recv(ERR,
 734					"post_recv failed rc=%d\n", rc);
 735				put_receive_buffer(sc, response);
 736				break;
 737			}
 738
 739			atomic_inc(&sc->recv_io.posted.count);
 740			posted += 1;
 741		}
 742	}
 743
 744	atomic_add(posted, &sc->recv_io.credits.available);
 745
 746	/*
 747	 * If the last send credit is waiting for credits
 748	 * it can grant we need to wake it up
 749	 */
 750	if (posted &&
 751	    atomic_read(&sc->send_io.bcredits.count) == 0 &&
 752	    atomic_read(&sc->send_io.credits.count) == 0)
 753		wake_up(&sc->send_io.credits.wait_queue);
 754
 755	/* Promptly send an immediate packet as defined in [MS-SMBD] 3.1.1.1 */
 756	if (atomic_read(&sc->recv_io.credits.count) <
 757		sc->recv_io.credits.target - 1) {
 758		log_keep_alive(INFO, "schedule send of an empty message\n");
 759		queue_work(sc->workqueue, &sc->idle.immediate_work);
 760	}
 761}
 762
 763/* Called from softirq, when recv is done */
 764static void recv_done(struct ib_cq *cq, struct ib_wc *wc)
 765{
 766	struct smbdirect_data_transfer *data_transfer;
 767	struct smbdirect_recv_io *response =
 768		container_of(wc->wr_cqe, struct smbdirect_recv_io, cqe);
 769	struct smbdirect_socket *sc = response->socket;
 770	struct smbdirect_socket_parameters *sp = &sc->parameters;
 771	int current_recv_credits;
 772	u16 old_recv_credit_target;
 773	u32 data_offset = 0;
 774	u32 data_length = 0;
 775	u32 remaining_data_length = 0;
 776	bool negotiate_done = false;
 777
 778	log_rdma_recv(INFO,
 779		      "response=0x%p type=%d wc status=%s wc opcode %d byte_len=%d pkey_index=%u\n",
 780		      response, sc->recv_io.expected,
 781		      ib_wc_status_msg(wc->status), wc->opcode,
 782		      wc->byte_len, wc->pkey_index);
 783
 784	if (wc->status != IB_WC_SUCCESS || wc->opcode != IB_WC_RECV) {
 785		if (wc->status != IB_WC_WR_FLUSH_ERR)
 786			log_rdma_recv(ERR, "wc->status=%s opcode=%d\n",
 787				ib_wc_status_msg(wc->status), wc->opcode);
 788		goto error;
 789	}
 790
 791	ib_dma_sync_single_for_cpu(
 792		wc->qp->device,
 793		response->sge.addr,
 794		response->sge.length,
 795		DMA_FROM_DEVICE);
 796
 797	/*
 798	 * Reset timer to the keepalive interval in
 799	 * order to trigger our next keepalive message.
 800	 */
 801	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_NONE;
 802	mod_delayed_work(sc->workqueue, &sc->idle.timer_work,
 803			 msecs_to_jiffies(sp->keepalive_interval_msec));
 804
 805	switch (sc->recv_io.expected) {
 806	/* SMBD negotiation response */
 807	case SMBDIRECT_EXPECT_NEGOTIATE_REP:
 808		dump_smbdirect_negotiate_resp(smbdirect_recv_io_payload(response));
 809		sc->recv_io.reassembly.full_packet_received = true;
 810		negotiate_done =
 811			process_negotiation_response(response, wc->byte_len);
 812		put_receive_buffer(sc, response);
 813		if (SMBDIRECT_CHECK_STATUS_WARN(sc, SMBDIRECT_SOCKET_NEGOTIATE_RUNNING))
 814			negotiate_done = false;
 815		if (!negotiate_done) {
 816			sc->status = SMBDIRECT_SOCKET_NEGOTIATE_FAILED;
 817			smbd_disconnect_rdma_connection(sc);
 818		} else {
 819			sc->status = SMBDIRECT_SOCKET_CONNECTED;
 820			wake_up(&sc->status_wait);
 821		}
 822
 823		return;
 824
 825	/* SMBD data transfer packet */
 826	case SMBDIRECT_EXPECT_DATA_TRANSFER:
 827		data_transfer = smbdirect_recv_io_payload(response);
 828
 829		if (wc->byte_len <
 830		    offsetof(struct smbdirect_data_transfer, padding))
 831			goto error;
 832
 833		remaining_data_length = le32_to_cpu(data_transfer->remaining_data_length);
 834		data_offset = le32_to_cpu(data_transfer->data_offset);
 835		data_length = le32_to_cpu(data_transfer->data_length);
 836		if (wc->byte_len < data_offset ||
 837		    (u64)wc->byte_len < (u64)data_offset + data_length)
 838			goto error;
 839
 840		if (remaining_data_length > sp->max_fragmented_recv_size ||
 841		    data_length > sp->max_fragmented_recv_size ||
 842		    (u64)remaining_data_length + (u64)data_length > (u64)sp->max_fragmented_recv_size)
 843			goto error;
 844
 845		if (data_length) {
 846			if (sc->recv_io.reassembly.full_packet_received)
 847				response->first_segment = true;
 848
 849			if (le32_to_cpu(data_transfer->remaining_data_length))
 850				sc->recv_io.reassembly.full_packet_received = false;
 851			else
 852				sc->recv_io.reassembly.full_packet_received = true;
 853		}
 854
 855		atomic_dec(&sc->recv_io.posted.count);
 856		current_recv_credits = atomic_dec_return(&sc->recv_io.credits.count);
 857
 858		old_recv_credit_target = sc->recv_io.credits.target;
 859		sc->recv_io.credits.target =
 860			le16_to_cpu(data_transfer->credits_requested);
 861		sc->recv_io.credits.target =
 862			min_t(u16, sc->recv_io.credits.target, sp->recv_credit_max);
 863		sc->recv_io.credits.target =
 864			max_t(u16, sc->recv_io.credits.target, 1);
 865		if (le16_to_cpu(data_transfer->credits_granted)) {
 866			atomic_add(le16_to_cpu(data_transfer->credits_granted),
 867				&sc->send_io.credits.count);
 868			/*
 869			 * We have new send credits granted from remote peer
 870			 * If any sender is waiting for credits, unblock it
 871			 */
 872			wake_up(&sc->send_io.credits.wait_queue);
 873		}
 874
 875		log_incoming(INFO, "data flags %d data_offset %d data_length %d remaining_data_length %d\n",
 876			     le16_to_cpu(data_transfer->flags),
 877			     le32_to_cpu(data_transfer->data_offset),
 878			     le32_to_cpu(data_transfer->data_length),
 879			     le32_to_cpu(data_transfer->remaining_data_length));
 880
 881		/* Send an immediate response right away if requested */
 882		if (le16_to_cpu(data_transfer->flags) &
 883				SMBDIRECT_FLAG_RESPONSE_REQUESTED) {
 884			log_keep_alive(INFO, "schedule send of immediate response\n");
 885			queue_work(sc->workqueue, &sc->idle.immediate_work);
 886		}
 887
 888		/*
 889		 * If this is a packet with data playload place the data in
 890		 * reassembly queue and wake up the reading thread
 891		 */
 892		if (data_length) {
 893			if (current_recv_credits <= (sc->recv_io.credits.target / 4) ||
 894			    sc->recv_io.credits.target > old_recv_credit_target)
 895				queue_work(sc->workqueue, &sc->recv_io.posted.refill_work);
 896
 897			enqueue_reassembly(sc, response, data_length);
 898			wake_up(&sc->recv_io.reassembly.wait_queue);
 899		} else
 900			put_receive_buffer(sc, response);
 901
 902		return;
 903
 904	case SMBDIRECT_EXPECT_NEGOTIATE_REQ:
 905		/* Only server... */
 906		break;
 907	}
 908
 909	/*
 910	 * This is an internal error!
 911	 */
 912	log_rdma_recv(ERR, "unexpected response type=%d\n", sc->recv_io.expected);
 913	WARN_ON_ONCE(sc->recv_io.expected != SMBDIRECT_EXPECT_DATA_TRANSFER);
 914error:
 915	put_receive_buffer(sc, response);
 916	smbd_disconnect_rdma_connection(sc);
 917}
 918
 919static struct rdma_cm_id *smbd_create_id(
 920		struct smbdirect_socket *sc,
 921		struct sockaddr *dstaddr, int port)
 922{
 923	struct smbdirect_socket_parameters *sp = &sc->parameters;
 924	struct rdma_cm_id *id;
 925	int rc;
 926	__be16 *sport;
 927
 928	id = rdma_create_id(&init_net, smbd_conn_upcall, sc,
 929		RDMA_PS_TCP, IB_QPT_RC);
 930	if (IS_ERR(id)) {
 931		rc = PTR_ERR(id);
 932		log_rdma_event(ERR, "rdma_create_id() failed %i\n", rc);
 933		return id;
 934	}
 935
 936	if (dstaddr->sa_family == AF_INET6)
 937		sport = &((struct sockaddr_in6 *)dstaddr)->sin6_port;
 938	else
 939		sport = &((struct sockaddr_in *)dstaddr)->sin_port;
 940
 941	*sport = htons(port);
 942
 943	WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED);
 944	sc->status = SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING;
 945	rc = rdma_resolve_addr(id, NULL, (struct sockaddr *)dstaddr,
 946		sp->resolve_addr_timeout_msec);
 947	if (rc) {
 948		log_rdma_event(ERR, "rdma_resolve_addr() failed %i\n", rc);
 949		goto out;
 950	}
 951	rc = wait_event_interruptible_timeout(
 952		sc->status_wait,
 953		sc->status != SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING,
 954		msecs_to_jiffies(sp->resolve_addr_timeout_msec));
 955	/* e.g. if interrupted returns -ERESTARTSYS */
 956	if (rc < 0) {
 957		log_rdma_event(ERR, "rdma_resolve_addr timeout rc: %i\n", rc);
 958		goto out;
 959	}
 960	if (sc->status == SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING) {
 961		rc = -ETIMEDOUT;
 962		log_rdma_event(ERR, "rdma_resolve_addr() completed %i\n", rc);
 963		goto out;
 964	}
 965	if (sc->status != SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED) {
 966		rc = -EHOSTUNREACH;
 967		log_rdma_event(ERR, "rdma_resolve_addr() completed %i\n", rc);
 968		goto out;
 969	}
 970
 971	WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED);
 972	sc->status = SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING;
 973	rc = rdma_resolve_route(id, sp->resolve_route_timeout_msec);
 974	if (rc) {
 975		log_rdma_event(ERR, "rdma_resolve_route() failed %i\n", rc);
 976		goto out;
 977	}
 978	rc = wait_event_interruptible_timeout(
 979		sc->status_wait,
 980		sc->status != SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING,
 981		msecs_to_jiffies(sp->resolve_route_timeout_msec));
 982	/* e.g. if interrupted returns -ERESTARTSYS */
 983	if (rc < 0)  {
 984		log_rdma_event(ERR, "rdma_resolve_addr timeout rc: %i\n", rc);
 985		goto out;
 986	}
 987	if (sc->status == SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING) {
 988		rc = -ETIMEDOUT;
 989		log_rdma_event(ERR, "rdma_resolve_route() completed %i\n", rc);
 990		goto out;
 991	}
 992	if (sc->status != SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED) {
 993		rc = -ENETUNREACH;
 994		log_rdma_event(ERR, "rdma_resolve_route() completed %i\n", rc);
 995		goto out;
 996	}
 997
 998	return id;
 999
1000out:
1001	rdma_destroy_id(id);
1002	return ERR_PTR(rc);
1003}
1004
1005/*
1006 * Test if FRWR (Fast Registration Work Requests) is supported on the device
1007 * This implementation requires FRWR on RDMA read/write
1008 * return value: true if it is supported
1009 */
1010static bool frwr_is_supported(struct ib_device_attr *attrs)
1011{
1012	if (!(attrs->device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS))
1013		return false;
1014	if (attrs->max_fast_reg_page_list_len == 0)
1015		return false;
1016	return true;
1017}
1018
1019static int smbd_ia_open(
1020		struct smbdirect_socket *sc,
1021		struct sockaddr *dstaddr, int port)
1022{
1023	struct smbdirect_socket_parameters *sp = &sc->parameters;
1024	int rc;
1025
1026	WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_CREATED);
1027	sc->status = SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED;
1028
1029	sc->rdma.cm_id = smbd_create_id(sc, dstaddr, port);
1030	if (IS_ERR(sc->rdma.cm_id)) {
1031		rc = PTR_ERR(sc->rdma.cm_id);
1032		goto out1;
1033	}
1034	sc->ib.dev = sc->rdma.cm_id->device;
1035
1036	if (!frwr_is_supported(&sc->ib.dev->attrs)) {
1037		log_rdma_event(ERR, "Fast Registration Work Requests (FRWR) is not supported\n");
1038		log_rdma_event(ERR, "Device capability flags = %llx max_fast_reg_page_list_len = %u\n",
1039			       sc->ib.dev->attrs.device_cap_flags,
1040			       sc->ib.dev->attrs.max_fast_reg_page_list_len);
1041		rc = -EPROTONOSUPPORT;
1042		goto out2;
1043	}
1044	sp->max_frmr_depth = min_t(u32,
1045		sp->max_frmr_depth,
1046		sc->ib.dev->attrs.max_fast_reg_page_list_len);
1047	sc->mr_io.type = IB_MR_TYPE_MEM_REG;
1048	if (sc->ib.dev->attrs.kernel_cap_flags & IBK_SG_GAPS_REG)
1049		sc->mr_io.type = IB_MR_TYPE_SG_GAPS;
1050
1051	return 0;
1052
1053out2:
1054	rdma_destroy_id(sc->rdma.cm_id);
1055	sc->rdma.cm_id = NULL;
1056
1057out1:
1058	return rc;
1059}
1060
1061/*
1062 * Send a negotiation request message to the peer
1063 * The negotiation procedure is in [MS-SMBD] 3.1.5.2 and 3.1.5.3
1064 * After negotiation, the transport is connected and ready for
1065 * carrying upper layer SMB payload
1066 */
1067static int smbd_post_send_negotiate_req(struct smbdirect_socket *sc)
1068{
1069	struct smbdirect_socket_parameters *sp = &sc->parameters;
1070	int rc;
1071	struct smbdirect_send_io *request;
1072	struct smbdirect_negotiate_req *packet;
1073
1074	request = smbd_alloc_send_io(sc);
1075	if (IS_ERR(request))
1076		return PTR_ERR(request);
1077
1078	packet = smbdirect_send_io_payload(request);
1079	packet->min_version = cpu_to_le16(SMBDIRECT_V1);
1080	packet->max_version = cpu_to_le16(SMBDIRECT_V1);
1081	packet->reserved = 0;
1082	packet->credits_requested = cpu_to_le16(sp->send_credit_target);
1083	packet->preferred_send_size = cpu_to_le32(sp->max_send_size);
1084	packet->max_receive_size = cpu_to_le32(sp->max_recv_size);
1085	packet->max_fragmented_size =
1086		cpu_to_le32(sp->max_fragmented_recv_size);
1087
1088	request->sge[0].addr = ib_dma_map_single(
1089				sc->ib.dev, (void *)packet,
1090				sizeof(*packet), DMA_TO_DEVICE);
1091	if (ib_dma_mapping_error(sc->ib.dev, request->sge[0].addr)) {
1092		rc = -EIO;
1093		goto dma_mapping_failed;
1094	}
1095	request->num_sge = 1;
1096
1097	request->sge[0].length = sizeof(*packet);
1098	request->sge[0].lkey = sc->ib.pd->local_dma_lkey;
1099
1100	rc = smbd_post_send(sc, NULL, request);
1101	if (!rc)
1102		return 0;
1103
1104	if (rc == -EAGAIN)
1105		rc = -EIO;
1106
1107dma_mapping_failed:
1108	smbd_free_send_io(request);
1109	return rc;
1110}
1111
1112/*
1113 * Extend the credits to remote peer
1114 * This implements [MS-SMBD] 3.1.5.9
1115 * The idea is that we should extend credits to remote peer as quickly as
1116 * it's allowed, to maintain data flow. We allocate as much receive
1117 * buffer as possible, and extend the receive credits to remote peer
1118 * return value: the new credtis being granted.
1119 */
1120static int manage_credits_prior_sending(struct smbdirect_socket *sc)
1121{
1122	int missing;
1123	int available;
1124	int new_credits;
1125
1126	if (atomic_read(&sc->recv_io.credits.count) >= sc->recv_io.credits.target)
1127		return 0;
1128
1129	missing = (int)sc->recv_io.credits.target - atomic_read(&sc->recv_io.credits.count);
1130	available = atomic_xchg(&sc->recv_io.credits.available, 0);
1131	new_credits = (u16)min3(U16_MAX, missing, available);
1132	if (new_credits <= 0) {
1133		/*
1134		 * If credits are available, but not granted
1135		 * we need to re-add them again.
1136		 */
1137		if (available)
1138			atomic_add(available, &sc->recv_io.credits.available);
1139		return 0;
1140	}
1141
1142	if (new_credits < available) {
1143		/*
1144		 * Readd the remaining available again.
1145		 */
1146		available -= new_credits;
1147		atomic_add(available, &sc->recv_io.credits.available);
1148	}
1149
1150	/*
1151	 * Remember we granted the credits
1152	 */
1153	atomic_add(new_credits, &sc->recv_io.credits.count);
1154	return new_credits;
1155}
1156
1157/*
1158 * Check if we need to send a KEEP_ALIVE message
1159 * The idle connection timer triggers a KEEP_ALIVE message when expires
1160 * SMBDIRECT_FLAG_RESPONSE_REQUESTED is set in the message flag to have peer send
1161 * back a response.
1162 * return value:
1163 * 1 if SMBDIRECT_FLAG_RESPONSE_REQUESTED needs to be set
1164 * 0: otherwise
1165 */
1166static int manage_keep_alive_before_sending(struct smbdirect_socket *sc)
1167{
1168	struct smbdirect_socket_parameters *sp = &sc->parameters;
1169
1170	if (sc->idle.keepalive == SMBDIRECT_KEEPALIVE_PENDING) {
1171		sc->idle.keepalive = SMBDIRECT_KEEPALIVE_SENT;
1172		/*
1173		 * Now use the keepalive timeout (instead of keepalive interval)
1174		 * in order to wait for a response
1175		 */
1176		mod_delayed_work(sc->workqueue, &sc->idle.timer_work,
1177				 msecs_to_jiffies(sp->keepalive_timeout_msec));
1178		return 1;
1179	}
1180	return 0;
1181}
1182
1183static int smbd_ib_post_send(struct smbdirect_socket *sc,
1184			     struct ib_send_wr *wr)
1185{
1186	int ret;
1187
1188	atomic_inc(&sc->send_io.pending.count);
1189	ret = ib_post_send(sc->ib.qp, wr, NULL);
1190	if (ret) {
1191		pr_err("failed to post send: %d\n", ret);
1192		smbd_disconnect_rdma_connection(sc);
1193		ret = -EAGAIN;
1194	}
1195	return ret;
1196}
1197
1198/* Post the send request */
1199static int smbd_post_send(struct smbdirect_socket *sc,
1200			  struct smbdirect_send_batch *batch,
1201			  struct smbdirect_send_io *request)
1202{
1203	int i;
1204
1205	for (i = 0; i < request->num_sge; i++) {
1206		log_rdma_send(INFO,
1207			"rdma_request sge[%d] addr=0x%llx length=%u\n",
1208			i, request->sge[i].addr, request->sge[i].length);
1209		ib_dma_sync_single_for_device(
1210			sc->ib.dev,
1211			request->sge[i].addr,
1212			request->sge[i].length,
1213			DMA_TO_DEVICE);
1214	}
1215
1216	request->cqe.done = send_done;
1217	request->wr.next = NULL;
1218	request->wr.sg_list = request->sge;
1219	request->wr.num_sge = request->num_sge;
1220	request->wr.opcode = IB_WR_SEND;
1221
1222	if (batch) {
1223		request->wr.wr_cqe = NULL;
1224		request->wr.send_flags = 0;
1225		if (!list_empty(&batch->msg_list)) {
1226			struct smbdirect_send_io *last;
1227
1228			last = list_last_entry(&batch->msg_list,
1229					       struct smbdirect_send_io,
1230					       sibling_list);
1231			last->wr.next = &request->wr;
1232		}
1233		list_add_tail(&request->sibling_list, &batch->msg_list);
1234		batch->wr_cnt++;
1235		return 0;
1236	}
1237
1238	request->wr.wr_cqe = &request->cqe;
1239	request->wr.send_flags = IB_SEND_SIGNALED;
1240	return smbd_ib_post_send(sc, &request->wr);
1241}
1242
1243static void smbd_send_batch_init(struct smbdirect_send_batch *batch,
1244				 bool need_invalidate_rkey,
1245				 unsigned int remote_key)
1246{
1247	INIT_LIST_HEAD(&batch->msg_list);
1248	batch->wr_cnt = 0;
1249	batch->need_invalidate_rkey = need_invalidate_rkey;
1250	batch->remote_key = remote_key;
1251	batch->credit = 0;
1252}
1253
1254static int smbd_send_batch_flush(struct smbdirect_socket *sc,
1255				 struct smbdirect_send_batch *batch,
1256				 bool is_last)
1257{
1258	struct smbdirect_send_io *first, *last;
1259	int ret = 0;
1260
1261	if (list_empty(&batch->msg_list))
1262		goto release_credit;
1263
1264	first = list_first_entry(&batch->msg_list,
1265				 struct smbdirect_send_io,
1266				 sibling_list);
1267	last = list_last_entry(&batch->msg_list,
1268			       struct smbdirect_send_io,
1269			       sibling_list);
1270
1271	if (batch->need_invalidate_rkey) {
1272		first->wr.opcode = IB_WR_SEND_WITH_INV;
1273		first->wr.ex.invalidate_rkey = batch->remote_key;
1274		batch->need_invalidate_rkey = false;
1275		batch->remote_key = 0;
1276	}
1277
1278	last->wr.send_flags = IB_SEND_SIGNALED;
1279	last->wr.wr_cqe = &last->cqe;
1280
1281	/*
1282	 * Remove last from batch->msg_list
1283	 * and splice the rest of batch->msg_list
1284	 * to last->sibling_list.
1285	 *
1286	 * batch->msg_list is a valid empty list
1287	 * at the end.
1288	 */
1289	list_del_init(&last->sibling_list);
1290	list_splice_tail_init(&batch->msg_list, &last->sibling_list);
1291	batch->wr_cnt = 0;
1292
1293	ret = smbd_ib_post_send(sc, &first->wr);
1294	if (ret) {
1295		struct smbdirect_send_io *sibling, *next;
1296
1297		list_for_each_entry_safe(sibling, next, &last->sibling_list, sibling_list) {
1298			list_del_init(&sibling->sibling_list);
1299			smbd_free_send_io(sibling);
1300		}
1301		smbd_free_send_io(last);
1302	}
1303
1304release_credit:
1305	if (is_last && !ret && batch->credit) {
1306		atomic_add(batch->credit, &sc->send_io.bcredits.count);
1307		batch->credit = 0;
1308		wake_up(&sc->send_io.bcredits.wait_queue);
1309	}
1310
1311	return ret;
1312}
1313
1314static int wait_for_credits(struct smbdirect_socket *sc,
1315			    wait_queue_head_t *waitq, atomic_t *total_credits,
1316			    int needed)
1317{
1318	int ret;
1319
1320	do {
1321		if (atomic_sub_return(needed, total_credits) >= 0)
1322			return 0;
1323
1324		atomic_add(needed, total_credits);
1325		ret = wait_event_interruptible(*waitq,
1326					       atomic_read(total_credits) >= needed ||
1327					       sc->status != SMBDIRECT_SOCKET_CONNECTED);
1328
1329		if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
1330			return -ENOTCONN;
1331		else if (ret < 0)
1332			return ret;
1333	} while (true);
1334}
1335
1336static int wait_for_send_bcredit(struct smbdirect_socket *sc,
1337				 struct smbdirect_send_batch *batch)
1338{
1339	int ret;
1340
1341	if (batch->credit)
1342		return 0;
1343
1344	ret = wait_for_credits(sc,
1345			       &sc->send_io.bcredits.wait_queue,
1346			       &sc->send_io.bcredits.count,
1347			       1);
1348	if (ret)
1349		return ret;
1350
1351	batch->credit = 1;
1352	return 0;
1353}
1354
1355static int wait_for_send_lcredit(struct smbdirect_socket *sc,
1356				 struct smbdirect_send_batch *batch)
1357{
1358	if (batch && (atomic_read(&sc->send_io.lcredits.count) <= 1)) {
1359		int ret;
1360
1361		ret = smbd_send_batch_flush(sc, batch, false);
1362		if (ret)
1363			return ret;
1364	}
1365
1366	return wait_for_credits(sc,
1367				&sc->send_io.lcredits.wait_queue,
1368				&sc->send_io.lcredits.count,
1369				1);
1370}
1371
1372static int wait_for_send_credits(struct smbdirect_socket *sc,
1373				 struct smbdirect_send_batch *batch)
1374{
1375	if (batch &&
1376	    (batch->wr_cnt >= 16 || atomic_read(&sc->send_io.credits.count) <= 1)) {
1377		int ret;
1378
1379		ret = smbd_send_batch_flush(sc, batch, false);
1380		if (ret)
1381			return ret;
1382	}
1383
1384	return wait_for_credits(sc,
1385				&sc->send_io.credits.wait_queue,
1386				&sc->send_io.credits.count,
1387				1);
1388}
1389
1390static int smbd_post_send_iter(struct smbdirect_socket *sc,
1391			       struct smbdirect_send_batch *batch,
1392			       struct iov_iter *iter,
1393			       int *_remaining_data_length)
1394{
1395	struct smbdirect_socket_parameters *sp = &sc->parameters;
1396	int rc;
1397	int header_length;
1398	int data_length;
1399	struct smbdirect_send_io *request;
1400	struct smbdirect_data_transfer *packet;
1401	int new_credits = 0;
1402	struct smbdirect_send_batch _batch;
1403
1404	if (!batch) {
1405		smbd_send_batch_init(&_batch, false, 0);
1406		batch = &_batch;
1407	}
1408
1409	rc = wait_for_send_bcredit(sc, batch);
1410	if (rc) {
1411		log_outgoing(ERR, "disconnected not sending on wait_bcredit\n");
1412		rc = -EAGAIN;
1413		goto err_wait_bcredit;
1414	}
1415
1416	rc = wait_for_send_lcredit(sc, batch);
1417	if (rc) {
1418		log_outgoing(ERR, "disconnected not sending on wait_lcredit\n");
1419		rc = -EAGAIN;
1420		goto err_wait_lcredit;
1421	}
1422
1423	rc = wait_for_send_credits(sc, batch);
1424	if (rc) {
1425		log_outgoing(ERR, "disconnected not sending on wait_credit\n");
1426		rc = -EAGAIN;
1427		goto err_wait_credit;
1428	}
1429
1430	new_credits = manage_credits_prior_sending(sc);
1431	if (new_credits == 0 &&
1432	    atomic_read(&sc->send_io.credits.count) == 0 &&
1433	    atomic_read(&sc->recv_io.credits.count) == 0) {
1434		queue_work(sc->workqueue, &sc->recv_io.posted.refill_work);
1435		rc = wait_event_interruptible(sc->send_io.credits.wait_queue,
1436					      atomic_read(&sc->send_io.credits.count) >= 1 ||
1437					      atomic_read(&sc->recv_io.credits.available) >= 1 ||
1438					      sc->status != SMBDIRECT_SOCKET_CONNECTED);
1439		if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
1440			rc = -ENOTCONN;
1441		if (rc < 0) {
1442			log_outgoing(ERR, "disconnected not sending on last credit\n");
1443			rc = -EAGAIN;
1444			goto err_wait_credit;
1445		}
1446
1447		new_credits = manage_credits_prior_sending(sc);
1448	}
1449
1450	request = smbd_alloc_send_io(sc);
1451	if (IS_ERR(request)) {
1452		rc = PTR_ERR(request);
1453		goto err_alloc;
1454	}
1455
1456	memset(request->sge, 0, sizeof(request->sge));
1457
1458	/* Map the packet to DMA */
1459	header_length = sizeof(struct smbdirect_data_transfer);
1460	/* If this is a packet without payload, don't send padding */
1461	if (!iter)
1462		header_length = offsetof(struct smbdirect_data_transfer, padding);
1463
1464	packet = smbdirect_send_io_payload(request);
1465	request->sge[0].addr = ib_dma_map_single(sc->ib.dev,
1466						 (void *)packet,
1467						 header_length,
1468						 DMA_TO_DEVICE);
1469	if (ib_dma_mapping_error(sc->ib.dev, request->sge[0].addr)) {
1470		rc = -EIO;
1471		goto err_dma;
1472	}
1473
1474	request->sge[0].length = header_length;
1475	request->sge[0].lkey = sc->ib.pd->local_dma_lkey;
1476	request->num_sge = 1;
1477
1478	/* Fill in the data payload to find out how much data we can add */
1479	if (iter) {
1480		struct smb_extract_to_rdma extract = {
1481			.nr_sge		= request->num_sge,
1482			.max_sge	= SMBDIRECT_SEND_IO_MAX_SGE,
1483			.sge		= request->sge,
1484			.device		= sc->ib.dev,
1485			.local_dma_lkey	= sc->ib.pd->local_dma_lkey,
1486			.direction	= DMA_TO_DEVICE,
1487		};
1488		size_t payload_len = umin(*_remaining_data_length,
1489					  sp->max_send_size - sizeof(*packet));
1490
1491		rc = smb_extract_iter_to_rdma(iter, payload_len,
1492					      &extract);
1493		if (rc < 0)
1494			goto err_dma;
1495		data_length = rc;
1496		request->num_sge = extract.nr_sge;
1497		*_remaining_data_length -= data_length;
1498	} else {
1499		data_length = 0;
1500	}
1501
1502	/* Fill in the packet header */
1503	packet->credits_requested = cpu_to_le16(sp->send_credit_target);
1504	packet->credits_granted = cpu_to_le16(new_credits);
1505
1506	packet->flags = 0;
1507	if (manage_keep_alive_before_sending(sc))
1508		packet->flags |= cpu_to_le16(SMBDIRECT_FLAG_RESPONSE_REQUESTED);
1509
1510	packet->reserved = 0;
1511	if (!data_length)
1512		packet->data_offset = 0;
1513	else
1514		packet->data_offset = cpu_to_le32(24);
1515	packet->data_length = cpu_to_le32(data_length);
1516	packet->remaining_data_length = cpu_to_le32(*_remaining_data_length);
1517	packet->padding = 0;
1518
1519	log_outgoing(INFO, "credits_requested=%d credits_granted=%d data_offset=%d data_length=%d remaining_data_length=%d\n",
1520		     le16_to_cpu(packet->credits_requested),
1521		     le16_to_cpu(packet->credits_granted),
1522		     le32_to_cpu(packet->data_offset),
1523		     le32_to_cpu(packet->data_length),
1524		     le32_to_cpu(packet->remaining_data_length));
1525
1526	rc = smbd_post_send(sc, batch, request);
1527	if (!rc) {
1528		/*
1529		 * From here request is moved to batch
1530		 * and we should not free it explicitly.
1531		 */
1532
1533		if (batch != &_batch)
1534			return 0;
1535
1536		rc = smbd_send_batch_flush(sc, batch, true);
1537		if (!rc)
1538			return 0;
1539
1540		goto err_flush;
1541	}
1542
1543err_dma:
1544	smbd_free_send_io(request);
1545
1546err_flush:
1547err_alloc:
1548	atomic_inc(&sc->send_io.credits.count);
1549	wake_up(&sc->send_io.credits.wait_queue);
1550
1551err_wait_credit:
1552	atomic_inc(&sc->send_io.lcredits.count);
1553	wake_up(&sc->send_io.lcredits.wait_queue);
1554
1555err_wait_lcredit:
1556	atomic_add(batch->credit, &sc->send_io.bcredits.count);
1557	batch->credit = 0;
1558	wake_up(&sc->send_io.bcredits.wait_queue);
1559
1560err_wait_bcredit:
1561	return rc;
1562}
1563
1564/*
1565 * Send an empty message
1566 * Empty message is used to extend credits to peer to for keep live
1567 * while there is no upper layer payload to send at the time
1568 */
1569static int smbd_post_send_empty(struct smbdirect_socket *sc)
1570{
1571	int remaining_data_length = 0;
1572
1573	sc->statistics.send_empty++;
1574	return smbd_post_send_iter(sc, NULL, NULL, &remaining_data_length);
1575}
1576
1577static int smbd_post_send_full_iter(struct smbdirect_socket *sc,
1578				    struct smbdirect_send_batch *batch,
1579				    struct iov_iter *iter,
1580				    int *_remaining_data_length)
1581{
1582	int rc = 0;
1583
1584	/*
1585	 * smbd_post_send_iter() respects the
1586	 * negotiated max_send_size, so we need to
1587	 * loop until the full iter is posted
1588	 */
1589
1590	while (iov_iter_count(iter) > 0) {
1591		rc = smbd_post_send_iter(sc, batch, iter, _remaining_data_length);
1592		if (rc < 0)
1593			break;
1594	}
1595
1596	return rc;
1597}
1598
1599/*
1600 * Post a receive request to the transport
1601 * The remote peer can only send data when a receive request is posted
1602 * The interaction is controlled by send/receive credit system
1603 */
1604static int smbd_post_recv(
1605		struct smbdirect_socket *sc, struct smbdirect_recv_io *response)
1606{
1607	struct smbdirect_socket_parameters *sp = &sc->parameters;
1608	struct ib_recv_wr recv_wr;
1609	int rc = -EIO;
1610
1611	response->sge.addr = ib_dma_map_single(
1612				sc->ib.dev, response->packet,
1613				sp->max_recv_size, DMA_FROM_DEVICE);
1614	if (ib_dma_mapping_error(sc->ib.dev, response->sge.addr))
1615		return rc;
1616
1617	response->sge.length = sp->max_recv_size;
1618	response->sge.lkey = sc->ib.pd->local_dma_lkey;
1619
1620	response->cqe.done = recv_done;
1621
1622	recv_wr.wr_cqe = &response->cqe;
1623	recv_wr.next = NULL;
1624	recv_wr.sg_list = &response->sge;
1625	recv_wr.num_sge = 1;
1626
1627	rc = ib_post_recv(sc->ib.qp, &recv_wr, NULL);
1628	if (rc) {
1629		ib_dma_unmap_single(sc->ib.dev, response->sge.addr,
1630				    response->sge.length, DMA_FROM_DEVICE);
1631		response->sge.length = 0;
1632		smbd_disconnect_rdma_connection(sc);
1633		log_rdma_recv(ERR, "ib_post_recv failed rc=%d\n", rc);
1634	}
1635
1636	return rc;
1637}
1638
1639/* Perform SMBD negotiate according to [MS-SMBD] 3.1.5.2 */
1640static int smbd_negotiate(struct smbdirect_socket *sc)
1641{
1642	struct smbdirect_socket_parameters *sp = &sc->parameters;
1643	int rc;
1644	struct smbdirect_recv_io *response = get_receive_buffer(sc);
1645
1646	WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_NEGOTIATE_NEEDED);
1647	sc->status = SMBDIRECT_SOCKET_NEGOTIATE_RUNNING;
1648
1649	sc->recv_io.expected = SMBDIRECT_EXPECT_NEGOTIATE_REP;
1650	rc = smbd_post_recv(sc, response);
1651	log_rdma_event(INFO, "smbd_post_recv rc=%d iov.addr=0x%llx iov.length=%u iov.lkey=0x%x\n",
1652		       rc, response->sge.addr,
1653		       response->sge.length, response->sge.lkey);
1654	if (rc) {
1655		put_receive_buffer(sc, response);
1656		return rc;
1657	}
1658
1659	rc = smbd_post_send_negotiate_req(sc);
1660	if (rc)
1661		return rc;
1662
1663	rc = wait_event_interruptible_timeout(
1664		sc->status_wait,
1665		sc->status != SMBDIRECT_SOCKET_NEGOTIATE_RUNNING,
1666		msecs_to_jiffies(sp->negotiate_timeout_msec));
1667	log_rdma_event(INFO, "wait_event_interruptible_timeout rc=%d\n", rc);
1668
1669	if (sc->status == SMBDIRECT_SOCKET_CONNECTED)
1670		return 0;
1671
1672	if (rc == 0)
1673		rc = -ETIMEDOUT;
1674	else if (rc == -ERESTARTSYS)
1675		rc = -EINTR;
1676	else
1677		rc = -ENOTCONN;
1678
1679	return rc;
1680}
1681
1682/*
1683 * Implement Connection.FragmentReassemblyBuffer defined in [MS-SMBD] 3.1.1.1
1684 * This is a queue for reassembling upper layer payload and present to upper
1685 * layer. All the inncoming payload go to the reassembly queue, regardless of
1686 * if reassembly is required. The uuper layer code reads from the queue for all
1687 * incoming payloads.
1688 * Put a received packet to the reassembly queue
1689 * response: the packet received
1690 * data_length: the size of payload in this packet
1691 */
1692static void enqueue_reassembly(
1693	struct smbdirect_socket *sc,
1694	struct smbdirect_recv_io *response,
1695	int data_length)
1696{
1697	unsigned long flags;
1698
1699	spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
1700	list_add_tail(&response->list, &sc->recv_io.reassembly.list);
1701	sc->recv_io.reassembly.queue_length++;
1702	/*
1703	 * Make sure reassembly_data_length is updated after list and
1704	 * reassembly_queue_length are updated. On the dequeue side
1705	 * reassembly_data_length is checked without a lock to determine
1706	 * if reassembly_queue_length and list is up to date
1707	 */
1708	virt_wmb();
1709	sc->recv_io.reassembly.data_length += data_length;
1710	spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
1711	sc->statistics.enqueue_reassembly_queue++;
1712}
1713
1714/*
1715 * Get the first entry at the front of reassembly queue
1716 * Caller is responsible for locking
1717 * return value: the first entry if any, NULL if queue is empty
1718 */
1719static struct smbdirect_recv_io *_get_first_reassembly(struct smbdirect_socket *sc)
1720{
1721	struct smbdirect_recv_io *ret = NULL;
1722
1723	if (!list_empty(&sc->recv_io.reassembly.list)) {
1724		ret = list_first_entry(
1725			&sc->recv_io.reassembly.list,
1726			struct smbdirect_recv_io, list);
1727	}
1728	return ret;
1729}
1730
1731/*
1732 * Get a receive buffer
1733 * For each remote send, we need to post a receive. The receive buffers are
1734 * pre-allocated in advance.
1735 * return value: the receive buffer, NULL if none is available
1736 */
1737static struct smbdirect_recv_io *get_receive_buffer(struct smbdirect_socket *sc)
1738{
1739	struct smbdirect_recv_io *ret = NULL;
1740	unsigned long flags;
1741
1742	spin_lock_irqsave(&sc->recv_io.free.lock, flags);
1743	if (!list_empty(&sc->recv_io.free.list)) {
1744		ret = list_first_entry(
1745			&sc->recv_io.free.list,
1746			struct smbdirect_recv_io, list);
1747		list_del(&ret->list);
1748		sc->statistics.get_receive_buffer++;
1749	}
1750	spin_unlock_irqrestore(&sc->recv_io.free.lock, flags);
1751
1752	return ret;
1753}
1754
1755/*
1756 * Return a receive buffer
1757 * Upon returning of a receive buffer, we can post new receive and extend
1758 * more receive credits to remote peer. This is done immediately after a
1759 * receive buffer is returned.
1760 */
1761static void put_receive_buffer(
1762	struct smbdirect_socket *sc, struct smbdirect_recv_io *response)
1763{
1764	unsigned long flags;
1765
1766	if (likely(response->sge.length != 0)) {
1767		ib_dma_unmap_single(sc->ib.dev,
1768				    response->sge.addr,
1769				    response->sge.length,
1770				    DMA_FROM_DEVICE);
1771		response->sge.length = 0;
1772	}
1773
1774	spin_lock_irqsave(&sc->recv_io.free.lock, flags);
1775	list_add_tail(&response->list, &sc->recv_io.free.list);
1776	sc->statistics.put_receive_buffer++;
1777	spin_unlock_irqrestore(&sc->recv_io.free.lock, flags);
1778
1779	queue_work(sc->workqueue, &sc->recv_io.posted.refill_work);
1780}
1781
1782/* Preallocate all receive buffer on transport establishment */
1783static int allocate_receive_buffers(struct smbdirect_socket *sc, int num_buf)
1784{
1785	struct smbdirect_recv_io *response;
1786	int i;
1787
1788	for (i = 0; i < num_buf; i++) {
1789		response = mempool_alloc(sc->recv_io.mem.pool, GFP_KERNEL);
1790		if (!response)
1791			goto allocate_failed;
1792
1793		response->socket = sc;
1794		response->sge.length = 0;
1795		list_add_tail(&response->list, &sc->recv_io.free.list);
1796	}
1797
1798	return 0;
1799
1800allocate_failed:
1801	while (!list_empty(&sc->recv_io.free.list)) {
1802		response = list_first_entry(
1803				&sc->recv_io.free.list,
1804				struct smbdirect_recv_io, list);
1805		list_del(&response->list);
1806
1807		mempool_free(response, sc->recv_io.mem.pool);
1808	}
1809	return -ENOMEM;
1810}
1811
1812static void destroy_receive_buffers(struct smbdirect_socket *sc)
1813{
1814	struct smbdirect_recv_io *response;
1815
1816	while ((response = get_receive_buffer(sc)))
1817		mempool_free(response, sc->recv_io.mem.pool);
1818}
1819
1820static void send_immediate_empty_message(struct work_struct *work)
1821{
1822	struct smbdirect_socket *sc =
1823		container_of(work, struct smbdirect_socket, idle.immediate_work);
1824
1825	if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
1826		return;
1827
1828	log_keep_alive(INFO, "send an empty message\n");
1829	smbd_post_send_empty(sc);
1830}
1831
1832/* Implement idle connection timer [MS-SMBD] 3.1.6.2 */
1833static void idle_connection_timer(struct work_struct *work)
1834{
1835	struct smbdirect_socket *sc =
1836		container_of(work, struct smbdirect_socket, idle.timer_work.work);
1837	struct smbdirect_socket_parameters *sp = &sc->parameters;
1838
1839	if (sc->idle.keepalive != SMBDIRECT_KEEPALIVE_NONE) {
1840		log_keep_alive(ERR,
1841			"error status sc->idle.keepalive=%d\n",
1842			sc->idle.keepalive);
1843		smbd_disconnect_rdma_connection(sc);
1844		return;
1845	}
1846
1847	if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
1848		return;
1849
1850	/*
1851	 * Now use the keepalive timeout (instead of keepalive interval)
1852	 * in order to wait for a response
1853	 */
1854	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_PENDING;
1855	mod_delayed_work(sc->workqueue, &sc->idle.timer_work,
1856			 msecs_to_jiffies(sp->keepalive_timeout_msec));
1857	log_keep_alive(INFO, "schedule send of empty idle message\n");
1858	queue_work(sc->workqueue, &sc->idle.immediate_work);
1859}
1860
1861/*
1862 * Destroy the transport and related RDMA and memory resources
1863 * Need to go through all the pending counters and make sure on one is using
1864 * the transport while it is destroyed
1865 */
1866void smbd_destroy(struct TCP_Server_Info *server)
1867{
1868	struct smbd_connection *info = server->smbd_conn;
1869	struct smbdirect_socket *sc;
1870	struct smbdirect_recv_io *response;
1871	unsigned long flags;
1872
1873	if (!info) {
1874		log_rdma_event(INFO, "rdma session already destroyed\n");
1875		return;
1876	}
1877	sc = &info->socket;
1878
1879	log_rdma_event(INFO, "cancelling and disable disconnect_work\n");
1880	disable_work_sync(&sc->disconnect_work);
1881
1882	log_rdma_event(INFO, "destroying rdma session\n");
1883	if (sc->status < SMBDIRECT_SOCKET_DISCONNECTING)
1884		smbd_disconnect_rdma_work(&sc->disconnect_work);
1885	if (sc->status < SMBDIRECT_SOCKET_DISCONNECTED) {
1886		log_rdma_event(INFO, "wait for transport being disconnected\n");
1887		wait_event(sc->status_wait, sc->status == SMBDIRECT_SOCKET_DISCONNECTED);
1888		log_rdma_event(INFO, "waited for transport being disconnected\n");
1889	}
1890
1891	/*
1892	 * Wake up all waiters in all wait queues
1893	 * in order to notice the broken connection.
1894	 *
1895	 * Most likely this was already called via
1896	 * smbd_disconnect_rdma_work(), but call it again...
1897	 */
1898	smbd_disconnect_wake_up_all(sc);
1899
1900	log_rdma_event(INFO, "cancelling recv_io.posted.refill_work\n");
1901	disable_work_sync(&sc->recv_io.posted.refill_work);
1902
1903	log_rdma_event(INFO, "destroying qp\n");
1904	ib_drain_qp(sc->ib.qp);
1905	rdma_destroy_qp(sc->rdma.cm_id);
1906	sc->ib.qp = NULL;
1907
1908	log_rdma_event(INFO, "cancelling idle timer\n");
1909	disable_delayed_work_sync(&sc->idle.timer_work);
1910	log_rdma_event(INFO, "cancelling send immediate work\n");
1911	disable_work_sync(&sc->idle.immediate_work);
1912
1913	/* It's not possible for upper layer to get to reassembly */
1914	log_rdma_event(INFO, "drain the reassembly queue\n");
1915	do {
1916		spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
1917		response = _get_first_reassembly(sc);
1918		if (response) {
1919			list_del(&response->list);
1920			spin_unlock_irqrestore(
1921				&sc->recv_io.reassembly.lock, flags);
1922			put_receive_buffer(sc, response);
1923		} else
1924			spin_unlock_irqrestore(
1925				&sc->recv_io.reassembly.lock, flags);
1926	} while (response);
1927	sc->recv_io.reassembly.data_length = 0;
1928
1929	log_rdma_event(INFO, "free receive buffers\n");
1930	destroy_receive_buffers(sc);
1931
1932	log_rdma_event(INFO, "freeing mr list\n");
1933	destroy_mr_list(sc);
1934
1935	ib_free_cq(sc->ib.send_cq);
1936	ib_free_cq(sc->ib.recv_cq);
1937	ib_dealloc_pd(sc->ib.pd);
1938	rdma_destroy_id(sc->rdma.cm_id);
1939
1940	/* free mempools */
1941	mempool_destroy(sc->send_io.mem.pool);
1942	kmem_cache_destroy(sc->send_io.mem.cache);
1943
1944	mempool_destroy(sc->recv_io.mem.pool);
1945	kmem_cache_destroy(sc->recv_io.mem.cache);
1946
1947	sc->status = SMBDIRECT_SOCKET_DESTROYED;
1948
1949	destroy_workqueue(sc->workqueue);
1950	log_rdma_event(INFO,  "rdma session destroyed\n");
1951	kfree(info);
1952	server->smbd_conn = NULL;
1953}
1954
1955/*
1956 * Reconnect this SMBD connection, called from upper layer
1957 * return value: 0 on success, or actual error code
1958 */
1959int smbd_reconnect(struct TCP_Server_Info *server)
1960{
1961	log_rdma_event(INFO, "reconnecting rdma session\n");
1962
1963	if (!server->smbd_conn) {
1964		log_rdma_event(INFO, "rdma session already destroyed\n");
1965		goto create_conn;
1966	}
1967
1968	/*
1969	 * This is possible if transport is disconnected and we haven't received
1970	 * notification from RDMA, but upper layer has detected timeout
1971	 */
1972	if (server->smbd_conn->socket.status == SMBDIRECT_SOCKET_CONNECTED) {
1973		log_rdma_event(INFO, "disconnecting transport\n");
1974		smbd_destroy(server);
1975	}
1976
1977create_conn:
1978	log_rdma_event(INFO, "creating rdma session\n");
1979	server->smbd_conn = smbd_get_connection(
1980		server, (struct sockaddr *) &server->dstaddr);
1981
1982	if (server->smbd_conn) {
1983		cifs_dbg(VFS, "RDMA transport re-established\n");
1984		trace_smb3_smbd_connect_done(server->hostname, server->conn_id, &server->dstaddr);
1985		return 0;
1986	}
1987	trace_smb3_smbd_connect_err(server->hostname, server->conn_id, &server->dstaddr);
1988	return -ENOENT;
1989}
1990
1991static void destroy_caches(struct smbdirect_socket *sc)
1992{
1993	destroy_receive_buffers(sc);
1994	mempool_destroy(sc->recv_io.mem.pool);
1995	kmem_cache_destroy(sc->recv_io.mem.cache);
1996	mempool_destroy(sc->send_io.mem.pool);
1997	kmem_cache_destroy(sc->send_io.mem.cache);
1998}
1999
2000#define MAX_NAME_LEN	80
2001static int allocate_caches(struct smbdirect_socket *sc)
2002{
2003	struct smbdirect_socket_parameters *sp = &sc->parameters;
2004	char name[MAX_NAME_LEN];
2005	int rc;
2006
2007	if (WARN_ON_ONCE(sp->max_recv_size < sizeof(struct smbdirect_data_transfer)))
2008		return -ENOMEM;
2009
2010	scnprintf(name, MAX_NAME_LEN, "smbdirect_send_io_%p", sc);
2011	sc->send_io.mem.cache =
2012		kmem_cache_create(
2013			name,
2014			sizeof(struct smbdirect_send_io) +
2015				sizeof(struct smbdirect_data_transfer),
2016			0, SLAB_HWCACHE_ALIGN, NULL);
2017	if (!sc->send_io.mem.cache)
2018		return -ENOMEM;
2019
2020	sc->send_io.mem.pool =
2021		mempool_create(sp->send_credit_target, mempool_alloc_slab,
2022			mempool_free_slab, sc->send_io.mem.cache);
2023	if (!sc->send_io.mem.pool)
2024		goto out1;
2025
2026	scnprintf(name, MAX_NAME_LEN, "smbdirect_recv_io_%p", sc);
2027
2028	struct kmem_cache_args response_args = {
2029		.align		= __alignof__(struct smbdirect_recv_io),
2030		.useroffset	= (offsetof(struct smbdirect_recv_io, packet) +
2031				   sizeof(struct smbdirect_data_transfer)),
2032		.usersize	= sp->max_recv_size - sizeof(struct smbdirect_data_transfer),
2033	};
2034	sc->recv_io.mem.cache =
2035		kmem_cache_create(name,
2036				  sizeof(struct smbdirect_recv_io) + sp->max_recv_size,
2037				  &response_args, SLAB_HWCACHE_ALIGN);
2038	if (!sc->recv_io.mem.cache)
2039		goto out2;
2040
2041	sc->recv_io.mem.pool =
2042		mempool_create(sp->recv_credit_max, mempool_alloc_slab,
2043		       mempool_free_slab, sc->recv_io.mem.cache);
2044	if (!sc->recv_io.mem.pool)
2045		goto out3;
2046
2047	rc = allocate_receive_buffers(sc, sp->recv_credit_max);
2048	if (rc) {
2049		log_rdma_event(ERR, "failed to allocate receive buffers\n");
2050		goto out4;
2051	}
2052
2053	return 0;
2054
2055out4:
2056	mempool_destroy(sc->recv_io.mem.pool);
2057out3:
2058	kmem_cache_destroy(sc->recv_io.mem.cache);
2059out2:
2060	mempool_destroy(sc->send_io.mem.pool);
2061out1:
2062	kmem_cache_destroy(sc->send_io.mem.cache);
2063	return -ENOMEM;
2064}
2065
2066/* Create a SMBD connection, called by upper layer */
2067static struct smbd_connection *_smbd_get_connection(
2068	struct TCP_Server_Info *server, struct sockaddr *dstaddr, int port)
2069{
2070	int rc;
2071	struct smbd_connection *info;
2072	struct smbdirect_socket *sc;
2073	struct smbdirect_socket_parameters *sp;
2074	struct rdma_conn_param conn_param;
2075	struct ib_qp_cap qp_cap;
2076	struct ib_qp_init_attr qp_attr;
2077	struct sockaddr_in *addr_in = (struct sockaddr_in *) dstaddr;
2078	struct ib_port_immutable port_immutable;
2079	__be32 ird_ord_hdr[2];
2080	char wq_name[80];
2081	struct workqueue_struct *workqueue;
2082
2083	info = kzalloc(sizeof(struct smbd_connection), GFP_KERNEL);
2084	if (!info)
2085		return NULL;
2086	sc = &info->socket;
2087	scnprintf(wq_name, ARRAY_SIZE(wq_name), "smbd_%p", sc);
2088	workqueue = create_workqueue(wq_name);
2089	if (!workqueue)
2090		goto create_wq_failed;
2091	smbdirect_socket_init(sc);
2092	sc->workqueue = workqueue;
2093	sp = &sc->parameters;
2094
2095	INIT_WORK(&sc->disconnect_work, smbd_disconnect_rdma_work);
2096
2097	sp->resolve_addr_timeout_msec = RDMA_RESOLVE_TIMEOUT;
2098	sp->resolve_route_timeout_msec = RDMA_RESOLVE_TIMEOUT;
2099	sp->rdma_connect_timeout_msec = RDMA_RESOLVE_TIMEOUT;
2100	sp->negotiate_timeout_msec = SMBD_NEGOTIATE_TIMEOUT * 1000;
2101	sp->initiator_depth = 1;
2102	sp->responder_resources = SMBD_CM_RESPONDER_RESOURCES;
2103	sp->recv_credit_max = smbd_receive_credit_max;
2104	sp->send_credit_target = smbd_send_credit_target;
2105	sp->max_send_size = smbd_max_send_size;
2106	sp->max_fragmented_recv_size = smbd_max_fragmented_recv_size;
2107	sp->max_recv_size = smbd_max_receive_size;
2108	sp->max_frmr_depth = smbd_max_frmr_depth;
2109	sp->keepalive_interval_msec = smbd_keep_alive_interval * 1000;
2110	sp->keepalive_timeout_msec = KEEPALIVE_RECV_TIMEOUT * 1000;
2111
2112	rc = smbd_ia_open(sc, dstaddr, port);
2113	if (rc) {
2114		log_rdma_event(INFO, "smbd_ia_open rc=%d\n", rc);
2115		goto create_id_failed;
2116	}
2117
2118	if (sp->send_credit_target > sc->ib.dev->attrs.max_cqe ||
2119	    sp->send_credit_target > sc->ib.dev->attrs.max_qp_wr) {
2120		log_rdma_event(ERR, "consider lowering send_credit_target = %d. Possible CQE overrun, device reporting max_cqe %d max_qp_wr %d\n",
2121			       sp->send_credit_target,
2122			       sc->ib.dev->attrs.max_cqe,
2123			       sc->ib.dev->attrs.max_qp_wr);
2124		goto config_failed;
2125	}
2126
2127	if (sp->recv_credit_max > sc->ib.dev->attrs.max_cqe ||
2128	    sp->recv_credit_max > sc->ib.dev->attrs.max_qp_wr) {
2129		log_rdma_event(ERR, "consider lowering receive_credit_max = %d. Possible CQE overrun, device reporting max_cqe %d max_qp_wr %d\n",
2130			       sp->recv_credit_max,
2131			       sc->ib.dev->attrs.max_cqe,
2132			       sc->ib.dev->attrs.max_qp_wr);
2133		goto config_failed;
2134	}
2135
2136	if (sc->ib.dev->attrs.max_send_sge < SMBDIRECT_SEND_IO_MAX_SGE ||
2137	    sc->ib.dev->attrs.max_recv_sge < SMBDIRECT_RECV_IO_MAX_SGE) {
2138		log_rdma_event(ERR,
2139			"device %.*s max_send_sge/max_recv_sge = %d/%d too small\n",
2140			IB_DEVICE_NAME_MAX,
2141			sc->ib.dev->name,
2142			sc->ib.dev->attrs.max_send_sge,
2143			sc->ib.dev->attrs.max_recv_sge);
2144		goto config_failed;
2145	}
2146
2147	sp->responder_resources =
2148		min_t(u8, sp->responder_resources,
2149		      sc->ib.dev->attrs.max_qp_rd_atom);
2150	log_rdma_mr(INFO, "responder_resources=%d\n",
2151		sp->responder_resources);
2152
2153	/*
2154	 * We use allocate sp->responder_resources * 2 MRs
2155	 * and each MR needs WRs for REG and INV, so
2156	 * we use '* 4'.
2157	 *
2158	 * +1 for ib_drain_qp()
2159	 */
2160	memset(&qp_cap, 0, sizeof(qp_cap));
2161	qp_cap.max_send_wr = sp->send_credit_target + sp->responder_resources * 4 + 1;
2162	qp_cap.max_recv_wr = sp->recv_credit_max + 1;
2163	qp_cap.max_send_sge = SMBDIRECT_SEND_IO_MAX_SGE;
2164	qp_cap.max_recv_sge = SMBDIRECT_RECV_IO_MAX_SGE;
2165
2166	sc->ib.pd = ib_alloc_pd(sc->ib.dev, 0);
2167	if (IS_ERR(sc->ib.pd)) {
2168		rc = PTR_ERR(sc->ib.pd);
2169		sc->ib.pd = NULL;
2170		log_rdma_event(ERR, "ib_alloc_pd() returned %d\n", rc);
2171		goto alloc_pd_failed;
2172	}
2173
2174	sc->ib.send_cq =
2175		ib_alloc_cq_any(sc->ib.dev, sc,
2176				qp_cap.max_send_wr, IB_POLL_SOFTIRQ);
2177	if (IS_ERR(sc->ib.send_cq)) {
2178		sc->ib.send_cq = NULL;
2179		goto alloc_cq_failed;
2180	}
2181
2182	sc->ib.recv_cq =
2183		ib_alloc_cq_any(sc->ib.dev, sc,
2184				qp_cap.max_recv_wr, IB_POLL_SOFTIRQ);
2185	if (IS_ERR(sc->ib.recv_cq)) {
2186		sc->ib.recv_cq = NULL;
2187		goto alloc_cq_failed;
2188	}
2189
2190	memset(&qp_attr, 0, sizeof(qp_attr));
2191	qp_attr.event_handler = smbd_qp_async_error_upcall;
2192	qp_attr.qp_context = sc;
2193	qp_attr.cap = qp_cap;
2194	qp_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
2195	qp_attr.qp_type = IB_QPT_RC;
2196	qp_attr.send_cq = sc->ib.send_cq;
2197	qp_attr.recv_cq = sc->ib.recv_cq;
2198	qp_attr.port_num = ~0;
2199
2200	rc = rdma_create_qp(sc->rdma.cm_id, sc->ib.pd, &qp_attr);
2201	if (rc) {
2202		log_rdma_event(ERR, "rdma_create_qp failed %i\n", rc);
2203		goto create_qp_failed;
2204	}
2205	sc->ib.qp = sc->rdma.cm_id->qp;
2206
2207	memset(&conn_param, 0, sizeof(conn_param));
2208	conn_param.initiator_depth = sp->initiator_depth;
2209	conn_param.responder_resources = sp->responder_resources;
2210
2211	/* Need to send IRD/ORD in private data for iWARP */
2212	sc->ib.dev->ops.get_port_immutable(
2213		sc->ib.dev, sc->rdma.cm_id->port_num, &port_immutable);
2214	if (port_immutable.core_cap_flags & RDMA_CORE_PORT_IWARP) {
2215		ird_ord_hdr[0] = cpu_to_be32(conn_param.responder_resources);
2216		ird_ord_hdr[1] = cpu_to_be32(conn_param.initiator_depth);
2217		conn_param.private_data = ird_ord_hdr;
2218		conn_param.private_data_len = sizeof(ird_ord_hdr);
2219	} else {
2220		conn_param.private_data = NULL;
2221		conn_param.private_data_len = 0;
2222	}
2223
2224	conn_param.retry_count = SMBD_CM_RETRY;
2225	conn_param.rnr_retry_count = SMBD_CM_RNR_RETRY;
2226	conn_param.flow_control = 0;
2227
2228	log_rdma_event(INFO, "connecting to IP %pI4 port %d\n",
2229		&addr_in->sin_addr, port);
2230
2231	WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED);
2232	sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING;
2233	rc = rdma_connect(sc->rdma.cm_id, &conn_param);
2234	if (rc) {
2235		log_rdma_event(ERR, "rdma_connect() failed with %i\n", rc);
2236		goto rdma_connect_failed;
2237	}
2238
2239	wait_event_interruptible_timeout(
2240		sc->status_wait,
2241		sc->status != SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING,
2242		msecs_to_jiffies(sp->rdma_connect_timeout_msec));
2243
2244	if (sc->status != SMBDIRECT_SOCKET_NEGOTIATE_NEEDED) {
2245		log_rdma_event(ERR, "rdma_connect failed port=%d\n", port);
2246		goto rdma_connect_failed;
2247	}
2248
2249	log_rdma_event(INFO, "rdma_connect connected\n");
2250
2251	rc = allocate_caches(sc);
2252	if (rc) {
2253		log_rdma_event(ERR, "cache allocation failed\n");
2254		goto allocate_cache_failed;
2255	}
2256
2257	INIT_WORK(&sc->idle.immediate_work, send_immediate_empty_message);
2258	INIT_DELAYED_WORK(&sc->idle.timer_work, idle_connection_timer);
2259	/*
2260	 * start with the negotiate timeout and SMBDIRECT_KEEPALIVE_PENDING
2261	 * so that the timer will cause a disconnect.
2262	 */
2263	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_PENDING;
2264	mod_delayed_work(sc->workqueue, &sc->idle.timer_work,
2265			 msecs_to_jiffies(sp->negotiate_timeout_msec));
2266
2267	INIT_WORK(&sc->recv_io.posted.refill_work, smbd_post_send_credits);
2268
2269	rc = smbd_negotiate(sc);
2270	if (rc) {
2271		log_rdma_event(ERR, "smbd_negotiate rc=%d\n", rc);
2272		goto negotiation_failed;
2273	}
2274
2275	rc = allocate_mr_list(sc);
2276	if (rc) {
2277		log_rdma_mr(ERR, "memory registration allocation failed\n");
2278		goto allocate_mr_failed;
2279	}
2280
2281	return info;
2282
2283allocate_mr_failed:
2284	/* At this point, need to a full transport shutdown */
2285	server->smbd_conn = info;
2286	smbd_destroy(server);
2287	return NULL;
2288
2289negotiation_failed:
2290	disable_delayed_work_sync(&sc->idle.timer_work);
2291	destroy_caches(sc);
2292	sc->status = SMBDIRECT_SOCKET_NEGOTIATE_FAILED;
2293	rdma_disconnect(sc->rdma.cm_id);
2294	wait_event(sc->status_wait,
2295		sc->status == SMBDIRECT_SOCKET_DISCONNECTED);
2296
2297allocate_cache_failed:
2298rdma_connect_failed:
2299	rdma_destroy_qp(sc->rdma.cm_id);
2300
2301create_qp_failed:
2302alloc_cq_failed:
2303	if (sc->ib.send_cq)
2304		ib_free_cq(sc->ib.send_cq);
2305	if (sc->ib.recv_cq)
2306		ib_free_cq(sc->ib.recv_cq);
2307
2308	ib_dealloc_pd(sc->ib.pd);
2309
2310alloc_pd_failed:
2311config_failed:
2312	rdma_destroy_id(sc->rdma.cm_id);
2313
2314create_id_failed:
2315	destroy_workqueue(sc->workqueue);
2316create_wq_failed:
2317	kfree(info);
2318	return NULL;
2319}
2320
2321struct smbd_connection *smbd_get_connection(
2322	struct TCP_Server_Info *server, struct sockaddr *dstaddr)
2323{
2324	struct smbd_connection *ret;
2325	const struct smbdirect_socket_parameters *sp;
2326	int port = SMBD_PORT;
2327
2328try_again:
2329	ret = _smbd_get_connection(server, dstaddr, port);
2330
2331	/* Try SMB_PORT if SMBD_PORT doesn't work */
2332	if (!ret && port == SMBD_PORT) {
2333		port = SMB_PORT;
2334		goto try_again;
2335	}
2336	if (!ret)
2337		return NULL;
2338
2339	sp = &ret->socket.parameters;
2340
2341	server->rdma_readwrite_threshold =
2342		rdma_readwrite_threshold > sp->max_fragmented_send_size ?
2343		sp->max_fragmented_send_size :
2344		rdma_readwrite_threshold;
2345
2346	return ret;
2347}
2348
2349/*
2350 * Receive data from the transport's receive reassembly queue
2351 * All the incoming data packets are placed in reassembly queue
2352 * iter: the buffer to read data into
2353 * size: the length of data to read
2354 * return value: actual data read
2355 *
2356 * Note: this implementation copies the data from reassembly queue to receive
2357 * buffers used by upper layer. This is not the optimal code path. A better way
2358 * to do it is to not have upper layer allocate its receive buffers but rather
2359 * borrow the buffer from reassembly queue, and return it after data is
2360 * consumed. But this will require more changes to upper layer code, and also
2361 * need to consider packet boundaries while they still being reassembled.
2362 */
2363int smbd_recv(struct smbd_connection *info, struct msghdr *msg)
2364{
2365	struct smbdirect_socket *sc = &info->socket;
2366	struct smbdirect_recv_io *response;
2367	struct smbdirect_data_transfer *data_transfer;
2368	size_t size = iov_iter_count(&msg->msg_iter);
2369	int to_copy, to_read, data_read, offset;
2370	u32 data_length, remaining_data_length, data_offset;
2371	int rc;
2372
2373	if (WARN_ON_ONCE(iov_iter_rw(&msg->msg_iter) == WRITE))
2374		return -EINVAL; /* It's a bug in upper layer to get there */
2375
2376again:
2377	/*
2378	 * No need to hold the reassembly queue lock all the time as we are
2379	 * the only one reading from the front of the queue. The transport
2380	 * may add more entries to the back of the queue at the same time
2381	 */
2382	log_read(INFO, "size=%zd sc->recv_io.reassembly.data_length=%d\n", size,
2383		sc->recv_io.reassembly.data_length);
2384	if (sc->recv_io.reassembly.data_length >= size) {
2385		int queue_length;
2386		int queue_removed = 0;
2387		unsigned long flags;
2388
2389		/*
2390		 * Need to make sure reassembly_data_length is read before
2391		 * reading reassembly_queue_length and calling
2392		 * _get_first_reassembly. This call is lock free
2393		 * as we never read at the end of the queue which are being
2394		 * updated in SOFTIRQ as more data is received
2395		 */
2396		virt_rmb();
2397		queue_length = sc->recv_io.reassembly.queue_length;
2398		data_read = 0;
2399		to_read = size;
2400		offset = sc->recv_io.reassembly.first_entry_offset;
2401		while (data_read < size) {
2402			response = _get_first_reassembly(sc);
2403			data_transfer = smbdirect_recv_io_payload(response);
2404			data_length = le32_to_cpu(data_transfer->data_length);
2405			remaining_data_length =
2406				le32_to_cpu(
2407					data_transfer->remaining_data_length);
2408			data_offset = le32_to_cpu(data_transfer->data_offset);
2409
2410			/*
2411			 * The upper layer expects RFC1002 length at the
2412			 * beginning of the payload. Return it to indicate
2413			 * the total length of the packet. This minimize the
2414			 * change to upper layer packet processing logic. This
2415			 * will be eventually remove when an intermediate
2416			 * transport layer is added
2417			 */
2418			if (response->first_segment && size == 4) {
2419				unsigned int rfc1002_len =
2420					data_length + remaining_data_length;
2421				__be32 rfc1002_hdr = cpu_to_be32(rfc1002_len);
2422				if (copy_to_iter(&rfc1002_hdr, sizeof(rfc1002_hdr),
2423						 &msg->msg_iter) != sizeof(rfc1002_hdr))
2424					return -EFAULT;
2425				data_read = 4;
2426				response->first_segment = false;
2427				log_read(INFO, "returning rfc1002 length %d\n",
2428					rfc1002_len);
2429				goto read_rfc1002_done;
2430			}
2431
2432			to_copy = min_t(int, data_length - offset, to_read);
2433			if (copy_to_iter((char *)data_transfer + data_offset + offset,
2434					 to_copy, &msg->msg_iter) != to_copy)
2435				return -EFAULT;
2436
2437			/* move on to the next buffer? */
2438			if (to_copy == data_length - offset) {
2439				queue_length--;
2440				/*
2441				 * No need to lock if we are not at the
2442				 * end of the queue
2443				 */
2444				if (queue_length)
2445					list_del(&response->list);
2446				else {
2447					spin_lock_irqsave(
2448						&sc->recv_io.reassembly.lock, flags);
2449					list_del(&response->list);
2450					spin_unlock_irqrestore(
2451						&sc->recv_io.reassembly.lock, flags);
2452				}
2453				queue_removed++;
2454				sc->statistics.dequeue_reassembly_queue++;
2455				put_receive_buffer(sc, response);
2456				offset = 0;
2457				log_read(INFO, "put_receive_buffer offset=0\n");
2458			} else
2459				offset += to_copy;
2460
2461			to_read -= to_copy;
2462			data_read += to_copy;
2463
2464			log_read(INFO, "_get_first_reassembly memcpy %d bytes data_transfer_length-offset=%d after that to_read=%d data_read=%d offset=%d\n",
2465				 to_copy, data_length - offset,
2466				 to_read, data_read, offset);
2467		}
2468
2469		spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
2470		sc->recv_io.reassembly.data_length -= data_read;
2471		sc->recv_io.reassembly.queue_length -= queue_removed;
2472		spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
2473
2474		sc->recv_io.reassembly.first_entry_offset = offset;
2475		log_read(INFO, "returning to thread data_read=%d reassembly_data_length=%d first_entry_offset=%d\n",
2476			 data_read, sc->recv_io.reassembly.data_length,
2477			 sc->recv_io.reassembly.first_entry_offset);
2478read_rfc1002_done:
2479		return data_read;
2480	}
2481
2482	log_read(INFO, "wait_event on more data\n");
2483	rc = wait_event_interruptible(
2484		sc->recv_io.reassembly.wait_queue,
2485		sc->recv_io.reassembly.data_length >= size ||
2486			sc->status != SMBDIRECT_SOCKET_CONNECTED);
2487	/* Don't return any data if interrupted */
2488	if (rc)
2489		return rc;
2490
2491	if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
2492		log_read(ERR, "disconnected\n");
2493		return -ECONNABORTED;
2494	}
2495
2496	goto again;
2497}
2498
2499/*
2500 * Send data to transport
2501 * Each rqst is transported as a SMBDirect payload
2502 * rqst: the data to write
2503 * return value: 0 if successfully write, otherwise error code
2504 */
2505int smbd_send(struct TCP_Server_Info *server,
2506	int num_rqst, struct smb_rqst *rqst_array)
2507{
2508	struct smbd_connection *info = server->smbd_conn;
2509	struct smbdirect_socket *sc = &info->socket;
2510	struct smbdirect_socket_parameters *sp = &sc->parameters;
2511	struct smb_rqst *rqst;
2512	struct iov_iter iter;
2513	struct smbdirect_send_batch batch;
2514	unsigned int remaining_data_length, klen;
2515	int rc, i, rqst_idx;
2516	int error = 0;
2517
2518	if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
2519		return -EAGAIN;
2520
2521	/*
2522	 * Add in the page array if there is one. The caller needs to set
2523	 * rq_tailsz to PAGE_SIZE when the buffer has multiple pages and
2524	 * ends at page boundary
2525	 */
2526	remaining_data_length = 0;
2527	for (i = 0; i < num_rqst; i++)
2528		remaining_data_length += smb_rqst_len(server, &rqst_array[i]);
2529
2530	if (unlikely(remaining_data_length > sp->max_fragmented_send_size)) {
2531		/* assertion: payload never exceeds negotiated maximum */
2532		log_write(ERR, "payload size %d > max size %d\n",
2533			remaining_data_length, sp->max_fragmented_send_size);
2534		return -EINVAL;
2535	}
2536
2537	log_write(INFO, "num_rqst=%d total length=%u\n",
2538			num_rqst, remaining_data_length);
2539
2540	rqst_idx = 0;
2541	smbd_send_batch_init(&batch, false, 0);
2542	do {
2543		rqst = &rqst_array[rqst_idx];
2544
2545		cifs_dbg(FYI, "Sending smb (RDMA): idx=%d smb_len=%lu\n",
2546			 rqst_idx, smb_rqst_len(server, rqst));
2547		for (i = 0; i < rqst->rq_nvec; i++)
2548			dump_smb(rqst->rq_iov[i].iov_base, rqst->rq_iov[i].iov_len);
2549
2550		log_write(INFO, "RDMA-WR[%u] nvec=%d len=%u iter=%zu rqlen=%lu\n",
2551			  rqst_idx, rqst->rq_nvec, remaining_data_length,
2552			  iov_iter_count(&rqst->rq_iter), smb_rqst_len(server, rqst));
2553
2554		/* Send the metadata pages. */
2555		klen = 0;
2556		for (i = 0; i < rqst->rq_nvec; i++)
2557			klen += rqst->rq_iov[i].iov_len;
2558		iov_iter_kvec(&iter, ITER_SOURCE, rqst->rq_iov, rqst->rq_nvec, klen);
2559
2560		rc = smbd_post_send_full_iter(sc, &batch, &iter, &remaining_data_length);
2561		if (rc < 0) {
2562			error = rc;
2563			break;
2564		}
2565
2566		if (iov_iter_count(&rqst->rq_iter) > 0) {
2567			/* And then the data pages if there are any */
2568			rc = smbd_post_send_full_iter(sc, &batch, &rqst->rq_iter,
2569						      &remaining_data_length);
2570			if (rc < 0) {
2571				error = rc;
2572				break;
2573			}
2574		}
2575
2576	} while (++rqst_idx < num_rqst);
2577
2578	rc = smbd_send_batch_flush(sc, &batch, true);
2579	if (unlikely(!rc && error))
2580		rc = error;
2581
2582	/*
2583	 * As an optimization, we don't wait for individual I/O to finish
2584	 * before sending the next one.
2585	 * Send them all and wait for pending send count to get to 0
2586	 * that means all the I/Os have been out and we are good to return
2587	 */
2588
2589	wait_event(sc->send_io.pending.zero_wait_queue,
2590		atomic_read(&sc->send_io.pending.count) == 0 ||
2591		sc->status != SMBDIRECT_SOCKET_CONNECTED);
2592
2593	if (sc->status != SMBDIRECT_SOCKET_CONNECTED && rc == 0)
2594		rc = -EAGAIN;
2595
2596	return rc;
2597}
2598
2599static void register_mr_done(struct ib_cq *cq, struct ib_wc *wc)
2600{
2601	struct smbdirect_mr_io *mr =
2602		container_of(wc->wr_cqe, struct smbdirect_mr_io, cqe);
2603	struct smbdirect_socket *sc = mr->socket;
2604
2605	if (wc->status) {
2606		log_rdma_mr(ERR, "status=%d\n", wc->status);
2607		smbd_disconnect_rdma_connection(sc);
2608	}
2609}
2610
2611/*
2612 * The work queue function that recovers MRs
2613 * We need to call ib_dereg_mr() and ib_alloc_mr() before this MR can be used
2614 * again. Both calls are slow, so finish them in a workqueue. This will not
2615 * block I/O path.
2616 * There is one workqueue that recovers MRs, there is no need to lock as the
2617 * I/O requests calling smbd_register_mr will never update the links in the
2618 * mr_list.
2619 */
2620static void smbd_mr_recovery_work(struct work_struct *work)
2621{
2622	struct smbdirect_socket *sc =
2623		container_of(work, struct smbdirect_socket, mr_io.recovery_work);
2624	struct smbdirect_socket_parameters *sp = &sc->parameters;
2625	struct smbdirect_mr_io *smbdirect_mr;
2626	int rc;
2627
2628	list_for_each_entry(smbdirect_mr, &sc->mr_io.all.list, list) {
2629		if (smbdirect_mr->state == SMBDIRECT_MR_ERROR) {
2630
2631			/* recover this MR entry */
2632			rc = ib_dereg_mr(smbdirect_mr->mr);
2633			if (rc) {
2634				log_rdma_mr(ERR,
2635					"ib_dereg_mr failed rc=%x\n",
2636					rc);
2637				smbd_disconnect_rdma_connection(sc);
2638				continue;
2639			}
2640
2641			smbdirect_mr->mr = ib_alloc_mr(
2642				sc->ib.pd, sc->mr_io.type,
2643				sp->max_frmr_depth);
2644			if (IS_ERR(smbdirect_mr->mr)) {
2645				log_rdma_mr(ERR, "ib_alloc_mr failed mr_type=%x max_frmr_depth=%x\n",
2646					    sc->mr_io.type,
2647					    sp->max_frmr_depth);
2648				smbd_disconnect_rdma_connection(sc);
2649				continue;
2650			}
2651		} else
2652			/* This MR is being used, don't recover it */
2653			continue;
2654
2655		smbdirect_mr->state = SMBDIRECT_MR_READY;
2656
2657		/* smbdirect_mr->state is updated by this function
2658		 * and is read and updated by I/O issuing CPUs trying
2659		 * to get a MR, the call to atomic_inc_return
2660		 * implicates a memory barrier and guarantees this
2661		 * value is updated before waking up any calls to
2662		 * get_mr() from the I/O issuing CPUs
2663		 */
2664		if (atomic_inc_return(&sc->mr_io.ready.count) == 1)
2665			wake_up(&sc->mr_io.ready.wait_queue);
2666	}
2667}
2668
2669static void smbd_mr_disable_locked(struct smbdirect_mr_io *mr)
2670{
2671	struct smbdirect_socket *sc = mr->socket;
2672
2673	lockdep_assert_held(&mr->mutex);
2674
2675	if (mr->state == SMBDIRECT_MR_DISABLED)
2676		return;
2677
2678	if (mr->mr)
2679		ib_dereg_mr(mr->mr);
2680	if (mr->sgt.nents)
2681		ib_dma_unmap_sg(sc->ib.dev, mr->sgt.sgl, mr->sgt.nents, mr->dir);
2682	kfree(mr->sgt.sgl);
2683
2684	mr->mr = NULL;
2685	mr->sgt.sgl = NULL;
2686	mr->sgt.nents = 0;
2687
2688	mr->state = SMBDIRECT_MR_DISABLED;
2689}
2690
2691static void smbd_mr_free_locked(struct kref *kref)
2692{
2693	struct smbdirect_mr_io *mr =
2694		container_of(kref, struct smbdirect_mr_io, kref);
2695
2696	lockdep_assert_held(&mr->mutex);
2697
2698	/*
2699	 * smbd_mr_disable_locked() should already be called!
2700	 */
2701	if (WARN_ON_ONCE(mr->state != SMBDIRECT_MR_DISABLED))
2702		smbd_mr_disable_locked(mr);
2703
2704	mutex_unlock(&mr->mutex);
2705	mutex_destroy(&mr->mutex);
2706	kfree(mr);
2707}
2708
2709static void destroy_mr_list(struct smbdirect_socket *sc)
2710{
2711	struct smbdirect_mr_io *mr, *tmp;
2712	LIST_HEAD(all_list);
2713	unsigned long flags;
2714
2715	disable_work_sync(&sc->mr_io.recovery_work);
2716
2717	spin_lock_irqsave(&sc->mr_io.all.lock, flags);
2718	list_splice_tail_init(&sc->mr_io.all.list, &all_list);
2719	spin_unlock_irqrestore(&sc->mr_io.all.lock, flags);
2720
2721	list_for_each_entry_safe(mr, tmp, &all_list, list) {
2722		mutex_lock(&mr->mutex);
2723
2724		smbd_mr_disable_locked(mr);
2725		list_del(&mr->list);
2726		mr->socket = NULL;
2727
2728		/*
2729		 * No kref_put_mutex() as it's already locked.
2730		 *
2731		 * If smbd_mr_free_locked() is called
2732		 * and the mutex is unlocked and mr is gone,
2733		 * in that case kref_put() returned 1.
2734		 *
2735		 * If kref_put() returned 0 we know that
2736		 * smbd_mr_free_locked() didn't
2737		 * run. Not by us nor by anyone else, as we
2738		 * still hold the mutex, so we need to unlock.
2739		 *
2740		 * If the mr is still registered it will
2741		 * be dangling (detached from the connection
2742		 * waiting for smbd_deregister_mr() to be
2743		 * called in order to free the memory.
2744		 */
2745		if (!kref_put(&mr->kref, smbd_mr_free_locked))
2746			mutex_unlock(&mr->mutex);
2747	}
2748}
2749
2750/*
2751 * Allocate MRs used for RDMA read/write
2752 * The number of MRs will not exceed hardware capability in responder_resources
2753 * All MRs are kept in mr_list. The MR can be recovered after it's used
2754 * Recovery is done in smbd_mr_recovery_work. The content of list entry changes
2755 * as MRs are used and recovered for I/O, but the list links will not change
2756 */
2757static int allocate_mr_list(struct smbdirect_socket *sc)
2758{
2759	struct smbdirect_socket_parameters *sp = &sc->parameters;
2760	struct smbdirect_mr_io *mr;
2761	int ret;
2762	u32 i;
2763
2764	if (sp->responder_resources == 0) {
2765		log_rdma_mr(ERR, "responder_resources negotiated as 0\n");
2766		return -EINVAL;
2767	}
2768
2769	/* Allocate more MRs (2x) than hardware responder_resources */
2770	for (i = 0; i < sp->responder_resources * 2; i++) {
2771		mr = kzalloc(sizeof(*mr), GFP_KERNEL);
2772		if (!mr) {
2773			ret = -ENOMEM;
2774			goto kzalloc_mr_failed;
2775		}
2776
2777		kref_init(&mr->kref);
2778		mutex_init(&mr->mutex);
2779
2780		mr->mr = ib_alloc_mr(sc->ib.pd,
2781				     sc->mr_io.type,
2782				     sp->max_frmr_depth);
2783		if (IS_ERR(mr->mr)) {
2784			ret = PTR_ERR(mr->mr);
2785			log_rdma_mr(ERR, "ib_alloc_mr failed mr_type=%x max_frmr_depth=%x\n",
2786				    sc->mr_io.type, sp->max_frmr_depth);
2787			goto ib_alloc_mr_failed;
2788		}
2789
2790		mr->sgt.sgl = kcalloc(sp->max_frmr_depth,
2791				      sizeof(struct scatterlist),
2792				      GFP_KERNEL);
2793		if (!mr->sgt.sgl) {
2794			ret = -ENOMEM;
2795			log_rdma_mr(ERR, "failed to allocate sgl\n");
2796			goto kcalloc_sgl_failed;
2797		}
2798		mr->state = SMBDIRECT_MR_READY;
2799		mr->socket = sc;
2800
2801		list_add_tail(&mr->list, &sc->mr_io.all.list);
2802		atomic_inc(&sc->mr_io.ready.count);
2803	}
2804
2805	INIT_WORK(&sc->mr_io.recovery_work, smbd_mr_recovery_work);
2806
2807	return 0;
2808
2809kcalloc_sgl_failed:
2810	ib_dereg_mr(mr->mr);
2811ib_alloc_mr_failed:
2812	mutex_destroy(&mr->mutex);
2813	kfree(mr);
2814kzalloc_mr_failed:
2815	destroy_mr_list(sc);
2816	return ret;
2817}
2818
2819/*
2820 * Get a MR from mr_list. This function waits until there is at least one
2821 * MR available in the list. It may access the list while the
2822 * smbd_mr_recovery_work is recovering the MR list. This doesn't need a lock
2823 * as they never modify the same places. However, there may be several CPUs
2824 * issuing I/O trying to get MR at the same time, mr_list_lock is used to
2825 * protect this situation.
2826 */
2827static struct smbdirect_mr_io *get_mr(struct smbdirect_socket *sc)
2828{
2829	struct smbdirect_mr_io *ret;
2830	unsigned long flags;
2831	int rc;
2832again:
2833	rc = wait_event_interruptible(sc->mr_io.ready.wait_queue,
2834		atomic_read(&sc->mr_io.ready.count) ||
2835		sc->status != SMBDIRECT_SOCKET_CONNECTED);
2836	if (rc) {
2837		log_rdma_mr(ERR, "wait_event_interruptible rc=%x\n", rc);
2838		return NULL;
2839	}
2840
2841	if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
2842		log_rdma_mr(ERR, "sc->status=%x\n", sc->status);
2843		return NULL;
2844	}
2845
2846	spin_lock_irqsave(&sc->mr_io.all.lock, flags);
2847	list_for_each_entry(ret, &sc->mr_io.all.list, list) {
2848		if (ret->state == SMBDIRECT_MR_READY) {
2849			ret->state = SMBDIRECT_MR_REGISTERED;
2850			kref_get(&ret->kref);
2851			spin_unlock_irqrestore(&sc->mr_io.all.lock, flags);
2852			atomic_dec(&sc->mr_io.ready.count);
2853			atomic_inc(&sc->mr_io.used.count);
2854			return ret;
2855		}
2856	}
2857
2858	spin_unlock_irqrestore(&sc->mr_io.all.lock, flags);
2859	/*
2860	 * It is possible that we could fail to get MR because other processes may
2861	 * try to acquire a MR at the same time. If this is the case, retry it.
2862	 */
2863	goto again;
2864}
2865
2866/*
2867 * Transcribe the pages from an iterator into an MR scatterlist.
2868 */
2869static int smbd_iter_to_mr(struct iov_iter *iter,
2870			   struct sg_table *sgt,
2871			   unsigned int max_sg)
2872{
2873	int ret;
2874
2875	memset(sgt->sgl, 0, max_sg * sizeof(struct scatterlist));
2876
2877	ret = extract_iter_to_sg(iter, iov_iter_count(iter), sgt, max_sg, 0);
2878	WARN_ON(ret < 0);
2879	if (sgt->nents > 0)
2880		sg_mark_end(&sgt->sgl[sgt->nents - 1]);
2881	return ret;
2882}
2883
2884/*
2885 * Register memory for RDMA read/write
2886 * iter: the buffer to register memory with
2887 * writing: true if this is a RDMA write (SMB read), false for RDMA read
2888 * need_invalidate: true if this MR needs to be locally invalidated after I/O
2889 * return value: the MR registered, NULL if failed.
2890 */
2891struct smbdirect_mr_io *smbd_register_mr(struct smbd_connection *info,
2892				 struct iov_iter *iter,
2893				 bool writing, bool need_invalidate)
2894{
2895	struct smbdirect_socket *sc = &info->socket;
2896	struct smbdirect_socket_parameters *sp = &sc->parameters;
2897	struct smbdirect_mr_io *mr;
2898	int rc, num_pages;
2899	struct ib_reg_wr *reg_wr;
2900
2901	num_pages = iov_iter_npages(iter, sp->max_frmr_depth + 1);
2902	if (num_pages > sp->max_frmr_depth) {
2903		log_rdma_mr(ERR, "num_pages=%d max_frmr_depth=%d\n",
2904			num_pages, sp->max_frmr_depth);
2905		WARN_ON_ONCE(1);
2906		return NULL;
2907	}
2908
2909	mr = get_mr(sc);
2910	if (!mr) {
2911		log_rdma_mr(ERR, "get_mr returning NULL\n");
2912		return NULL;
2913	}
2914
2915	mutex_lock(&mr->mutex);
2916
2917	mr->dir = writing ? DMA_FROM_DEVICE : DMA_TO_DEVICE;
2918	mr->need_invalidate = need_invalidate;
2919	mr->sgt.nents = 0;
2920	mr->sgt.orig_nents = 0;
2921
2922	log_rdma_mr(INFO, "num_pages=0x%x count=0x%zx depth=%u\n",
2923		    num_pages, iov_iter_count(iter), sp->max_frmr_depth);
2924	smbd_iter_to_mr(iter, &mr->sgt, sp->max_frmr_depth);
2925
2926	rc = ib_dma_map_sg(sc->ib.dev, mr->sgt.sgl, mr->sgt.nents, mr->dir);
2927	if (!rc) {
2928		log_rdma_mr(ERR, "ib_dma_map_sg num_pages=%x dir=%x rc=%x\n",
2929			    num_pages, mr->dir, rc);
2930		goto dma_map_error;
2931	}
2932
2933	rc = ib_map_mr_sg(mr->mr, mr->sgt.sgl, mr->sgt.nents, NULL, PAGE_SIZE);
2934	if (rc != mr->sgt.nents) {
2935		log_rdma_mr(ERR,
2936			    "ib_map_mr_sg failed rc = %d nents = %x\n",
2937			    rc, mr->sgt.nents);
2938		goto map_mr_error;
2939	}
2940
2941	ib_update_fast_reg_key(mr->mr, ib_inc_rkey(mr->mr->rkey));
2942	reg_wr = &mr->wr;
2943	reg_wr->wr.opcode = IB_WR_REG_MR;
2944	mr->cqe.done = register_mr_done;
2945	reg_wr->wr.wr_cqe = &mr->cqe;
2946	reg_wr->wr.num_sge = 0;
2947	reg_wr->wr.send_flags = IB_SEND_SIGNALED;
2948	reg_wr->mr = mr->mr;
2949	reg_wr->key = mr->mr->rkey;
2950	reg_wr->access = writing ?
2951			IB_ACCESS_REMOTE_WRITE | IB_ACCESS_LOCAL_WRITE :
2952			IB_ACCESS_REMOTE_READ;
2953
2954	/*
2955	 * There is no need for waiting for complemtion on ib_post_send
2956	 * on IB_WR_REG_MR. Hardware enforces a barrier and order of execution
2957	 * on the next ib_post_send when we actually send I/O to remote peer
2958	 */
2959	rc = ib_post_send(sc->ib.qp, &reg_wr->wr, NULL);
2960	if (!rc) {
2961		/*
2962		 * get_mr() gave us a reference
2963		 * via kref_get(&mr->kref), we keep that and let
2964		 * the caller use smbd_deregister_mr()
2965		 * to remove it again.
2966		 */
2967		mutex_unlock(&mr->mutex);
2968		return mr;
2969	}
2970
2971	log_rdma_mr(ERR, "ib_post_send failed rc=%x reg_wr->key=%x\n",
2972		rc, reg_wr->key);
2973
2974	/* If all failed, attempt to recover this MR by setting it SMBDIRECT_MR_ERROR*/
2975map_mr_error:
2976	ib_dma_unmap_sg(sc->ib.dev, mr->sgt.sgl, mr->sgt.nents, mr->dir);
2977
2978dma_map_error:
2979	mr->sgt.nents = 0;
2980	mr->state = SMBDIRECT_MR_ERROR;
2981	if (atomic_dec_and_test(&sc->mr_io.used.count))
2982		wake_up(&sc->mr_io.cleanup.wait_queue);
2983
2984	smbd_disconnect_rdma_connection(sc);
2985
2986	/*
2987	 * get_mr() gave us a reference
2988	 * via kref_get(&mr->kref), we need to remove it again
2989	 * on error.
2990	 *
2991	 * No kref_put_mutex() as it's already locked.
2992	 *
2993	 * If smbd_mr_free_locked() is called
2994	 * and the mutex is unlocked and mr is gone,
2995	 * in that case kref_put() returned 1.
2996	 *
2997	 * If kref_put() returned 0 we know that
2998	 * smbd_mr_free_locked() didn't
2999	 * run. Not by us nor by anyone else, as we
3000	 * still hold the mutex, so we need to unlock.
3001	 */
3002	if (!kref_put(&mr->kref, smbd_mr_free_locked))
3003		mutex_unlock(&mr->mutex);
3004
3005	return NULL;
3006}
3007
3008static void local_inv_done(struct ib_cq *cq, struct ib_wc *wc)
3009{
3010	struct smbdirect_mr_io *smbdirect_mr;
3011	struct ib_cqe *cqe;
3012
3013	cqe = wc->wr_cqe;
3014	smbdirect_mr = container_of(cqe, struct smbdirect_mr_io, cqe);
3015	smbdirect_mr->state = SMBDIRECT_MR_INVALIDATED;
3016	if (wc->status != IB_WC_SUCCESS) {
3017		log_rdma_mr(ERR, "invalidate failed status=%x\n", wc->status);
3018		smbdirect_mr->state = SMBDIRECT_MR_ERROR;
3019	}
3020	complete(&smbdirect_mr->invalidate_done);
3021}
3022
3023/*
3024 * Deregister a MR after I/O is done
3025 * This function may wait if remote invalidation is not used
3026 * and we have to locally invalidate the buffer to prevent data is being
3027 * modified by remote peer after upper layer consumes it
3028 */
3029void smbd_deregister_mr(struct smbdirect_mr_io *mr)
3030{
3031	struct smbdirect_socket *sc = mr->socket;
3032
3033	mutex_lock(&mr->mutex);
3034	if (mr->state == SMBDIRECT_MR_DISABLED)
3035		goto put_kref;
3036
3037	if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
3038		smbd_mr_disable_locked(mr);
3039		goto put_kref;
3040	}
3041
3042	if (mr->need_invalidate) {
3043		struct ib_send_wr *wr = &mr->inv_wr;
3044		int rc;
3045
3046		/* Need to finish local invalidation before returning */
3047		wr->opcode = IB_WR_LOCAL_INV;
3048		mr->cqe.done = local_inv_done;
3049		wr->wr_cqe = &mr->cqe;
3050		wr->num_sge = 0;
3051		wr->ex.invalidate_rkey = mr->mr->rkey;
3052		wr->send_flags = IB_SEND_SIGNALED;
3053
3054		init_completion(&mr->invalidate_done);
3055		rc = ib_post_send(sc->ib.qp, wr, NULL);
3056		if (rc) {
3057			log_rdma_mr(ERR, "ib_post_send failed rc=%x\n", rc);
3058			smbd_mr_disable_locked(mr);
3059			smbd_disconnect_rdma_connection(sc);
3060			goto done;
3061		}
3062		wait_for_completion(&mr->invalidate_done);
3063		mr->need_invalidate = false;
3064	} else
3065		/*
3066		 * For remote invalidation, just set it to SMBDIRECT_MR_INVALIDATED
3067		 * and defer to mr_recovery_work to recover the MR for next use
3068		 */
3069		mr->state = SMBDIRECT_MR_INVALIDATED;
3070
3071	if (mr->sgt.nents) {
3072		ib_dma_unmap_sg(sc->ib.dev, mr->sgt.sgl, mr->sgt.nents, mr->dir);
3073		mr->sgt.nents = 0;
3074	}
3075
3076	if (mr->state == SMBDIRECT_MR_INVALIDATED) {
3077		mr->state = SMBDIRECT_MR_READY;
3078		if (atomic_inc_return(&sc->mr_io.ready.count) == 1)
3079			wake_up(&sc->mr_io.ready.wait_queue);
3080	} else
3081		/*
3082		 * Schedule the work to do MR recovery for future I/Os MR
3083		 * recovery is slow and don't want it to block current I/O
3084		 */
3085		queue_work(sc->workqueue, &sc->mr_io.recovery_work);
3086
3087done:
3088	if (atomic_dec_and_test(&sc->mr_io.used.count))
3089		wake_up(&sc->mr_io.cleanup.wait_queue);
3090
3091put_kref:
3092	/*
3093	 * No kref_put_mutex() as it's already locked.
3094	 *
3095	 * If smbd_mr_free_locked() is called
3096	 * and the mutex is unlocked and mr is gone,
3097	 * in that case kref_put() returned 1.
3098	 *
3099	 * If kref_put() returned 0 we know that
3100	 * smbd_mr_free_locked() didn't
3101	 * run. Not by us nor by anyone else, as we
3102	 * still hold the mutex, so we need to unlock
3103	 * and keep the mr in SMBDIRECT_MR_READY or
3104	 * SMBDIRECT_MR_ERROR state.
3105	 */
3106	if (!kref_put(&mr->kref, smbd_mr_free_locked))
3107		mutex_unlock(&mr->mutex);
3108}
3109
3110static bool smb_set_sge(struct smb_extract_to_rdma *rdma,
3111			struct page *lowest_page, size_t off, size_t len)
3112{
3113	struct ib_sge *sge = &rdma->sge[rdma->nr_sge];
3114	u64 addr;
3115
3116	addr = ib_dma_map_page(rdma->device, lowest_page,
3117			       off, len, rdma->direction);
3118	if (ib_dma_mapping_error(rdma->device, addr))
3119		return false;
3120
3121	sge->addr   = addr;
3122	sge->length = len;
3123	sge->lkey   = rdma->local_dma_lkey;
3124	rdma->nr_sge++;
3125	return true;
3126}
3127
3128/*
3129 * Extract page fragments from a BVEC-class iterator and add them to an RDMA
3130 * element list.  The pages are not pinned.
3131 */
3132static ssize_t smb_extract_bvec_to_rdma(struct iov_iter *iter,
3133					struct smb_extract_to_rdma *rdma,
3134					ssize_t maxsize)
3135{
3136	const struct bio_vec *bv = iter->bvec;
3137	unsigned long start = iter->iov_offset;
3138	unsigned int i;
3139	ssize_t ret = 0;
3140
3141	for (i = 0; i < iter->nr_segs; i++) {
3142		size_t off, len;
3143
3144		len = bv[i].bv_len;
3145		if (start >= len) {
3146			start -= len;
3147			continue;
3148		}
3149
3150		len = min_t(size_t, maxsize, len - start);
3151		off = bv[i].bv_offset + start;
3152
3153		if (!smb_set_sge(rdma, bv[i].bv_page, off, len))
3154			return -EIO;
3155
3156		ret += len;
3157		maxsize -= len;
3158		if (rdma->nr_sge >= rdma->max_sge || maxsize <= 0)
3159			break;
3160		start = 0;
3161	}
3162
3163	if (ret > 0)
3164		iov_iter_advance(iter, ret);
3165	return ret;
3166}
3167
3168/*
3169 * Extract fragments from a KVEC-class iterator and add them to an RDMA list.
3170 * This can deal with vmalloc'd buffers as well as kmalloc'd or static buffers.
3171 * The pages are not pinned.
3172 */
3173static ssize_t smb_extract_kvec_to_rdma(struct iov_iter *iter,
3174					struct smb_extract_to_rdma *rdma,
3175					ssize_t maxsize)
3176{
3177	const struct kvec *kv = iter->kvec;
3178	unsigned long start = iter->iov_offset;
3179	unsigned int i;
3180	ssize_t ret = 0;
3181
3182	for (i = 0; i < iter->nr_segs; i++) {
3183		struct page *page;
3184		unsigned long kaddr;
3185		size_t off, len, seg;
3186
3187		len = kv[i].iov_len;
3188		if (start >= len) {
3189			start -= len;
3190			continue;
3191		}
3192
3193		kaddr = (unsigned long)kv[i].iov_base + start;
3194		off = kaddr & ~PAGE_MASK;
3195		len = min_t(size_t, maxsize, len - start);
3196		kaddr &= PAGE_MASK;
3197
3198		maxsize -= len;
3199		do {
3200			seg = min_t(size_t, len, PAGE_SIZE - off);
3201
3202			if (is_vmalloc_or_module_addr((void *)kaddr))
3203				page = vmalloc_to_page((void *)kaddr);
3204			else
3205				page = virt_to_page((void *)kaddr);
3206
3207			if (!smb_set_sge(rdma, page, off, seg))
3208				return -EIO;
3209
3210			ret += seg;
3211			len -= seg;
3212			kaddr += PAGE_SIZE;
3213			off = 0;
3214		} while (len > 0 && rdma->nr_sge < rdma->max_sge);
3215
3216		if (rdma->nr_sge >= rdma->max_sge || maxsize <= 0)
3217			break;
3218		start = 0;
3219	}
3220
3221	if (ret > 0)
3222		iov_iter_advance(iter, ret);
3223	return ret;
3224}
3225
3226/*
3227 * Extract folio fragments from a FOLIOQ-class iterator and add them to an RDMA
3228 * list.  The folios are not pinned.
3229 */
3230static ssize_t smb_extract_folioq_to_rdma(struct iov_iter *iter,
3231					  struct smb_extract_to_rdma *rdma,
3232					  ssize_t maxsize)
3233{
3234	const struct folio_queue *folioq = iter->folioq;
3235	unsigned int slot = iter->folioq_slot;
3236	ssize_t ret = 0;
3237	size_t offset = iter->iov_offset;
3238
3239	BUG_ON(!folioq);
3240
3241	if (slot >= folioq_nr_slots(folioq)) {
3242		folioq = folioq->next;
3243		if (WARN_ON_ONCE(!folioq))
3244			return -EIO;
3245		slot = 0;
3246	}
3247
3248	do {
3249		struct folio *folio = folioq_folio(folioq, slot);
3250		size_t fsize = folioq_folio_size(folioq, slot);
3251
3252		if (offset < fsize) {
3253			size_t part = umin(maxsize, fsize - offset);
3254
3255			if (!smb_set_sge(rdma, folio_page(folio, 0), offset, part))
3256				return -EIO;
3257
3258			offset += part;
3259			ret += part;
3260			maxsize -= part;
3261		}
3262
3263		if (offset >= fsize) {
3264			offset = 0;
3265			slot++;
3266			if (slot >= folioq_nr_slots(folioq)) {
3267				if (!folioq->next) {
3268					WARN_ON_ONCE(ret < iter->count);
3269					break;
3270				}
3271				folioq = folioq->next;
3272				slot = 0;
3273			}
3274		}
3275	} while (rdma->nr_sge < rdma->max_sge && maxsize > 0);
3276
3277	iter->folioq = folioq;
3278	iter->folioq_slot = slot;
3279	iter->iov_offset = offset;
3280	iter->count -= ret;
3281	return ret;
3282}
3283
3284/*
3285 * Extract page fragments from up to the given amount of the source iterator
3286 * and build up an RDMA list that refers to all of those bits.  The RDMA list
3287 * is appended to, up to the maximum number of elements set in the parameter
3288 * block.
3289 *
3290 * The extracted page fragments are not pinned or ref'd in any way; if an
3291 * IOVEC/UBUF-type iterator is to be used, it should be converted to a
3292 * BVEC-type iterator and the pages pinned, ref'd or otherwise held in some
3293 * way.
3294 */
3295static ssize_t smb_extract_iter_to_rdma(struct iov_iter *iter, size_t len,
3296					struct smb_extract_to_rdma *rdma)
3297{
3298	ssize_t ret;
3299	int before = rdma->nr_sge;
3300
3301	switch (iov_iter_type(iter)) {
3302	case ITER_BVEC:
3303		ret = smb_extract_bvec_to_rdma(iter, rdma, len);
3304		break;
3305	case ITER_KVEC:
3306		ret = smb_extract_kvec_to_rdma(iter, rdma, len);
3307		break;
3308	case ITER_FOLIOQ:
3309		ret = smb_extract_folioq_to_rdma(iter, rdma, len);
3310		break;
3311	default:
3312		WARN_ON_ONCE(1);
3313		return -EIO;
3314	}
3315
3316	if (ret < 0) {
3317		while (rdma->nr_sge > before) {
3318			struct ib_sge *sge = &rdma->sge[rdma->nr_sge--];
3319
3320			ib_dma_unmap_single(rdma->device, sge->addr, sge->length,
3321					    rdma->direction);
3322			sge->addr = 0;
3323		}
3324	}
3325
3326	return ret;
3327}