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