df95c7568445664a25483fcad97651147ceb3409
[sfrench/cifs-2.6.git] / fs / cifs / smbdirect.c
1 /*
2  *   Copyright (C) 2017, Microsoft Corporation.
3  *
4  *   Author(s): Long Li <longli@microsoft.com>
5  *
6  *   This program is free software;  you can redistribute it and/or modify
7  *   it under the terms of the GNU General Public License as published by
8  *   the Free Software Foundation; either version 2 of the License, or
9  *   (at your option) any later version.
10  *
11  *   This program is distributed in the hope that it will be useful,
12  *   but WITHOUT ANY WARRANTY;  without even the implied warranty of
13  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See
14  *   the GNU General Public License for more details.
15  */
16 #include <linux/module.h>
17 #include <linux/highmem.h>
18 #include "smbdirect.h"
19 #include "cifs_debug.h"
20 #include "cifsproto.h"
21 #include "smb2proto.h"
22
23 static struct smbd_response *get_empty_queue_buffer(
24                 struct smbd_connection *info);
25 static struct smbd_response *get_receive_buffer(
26                 struct smbd_connection *info);
27 static void put_receive_buffer(
28                 struct smbd_connection *info,
29                 struct smbd_response *response);
30 static int allocate_receive_buffers(struct smbd_connection *info, int num_buf);
31 static void destroy_receive_buffers(struct smbd_connection *info);
32
33 static void put_empty_packet(
34                 struct smbd_connection *info, struct smbd_response *response);
35 static void enqueue_reassembly(
36                 struct smbd_connection *info,
37                 struct smbd_response *response, int data_length);
38 static struct smbd_response *_get_first_reassembly(
39                 struct smbd_connection *info);
40
41 static int smbd_post_recv(
42                 struct smbd_connection *info,
43                 struct smbd_response *response);
44
45 static int smbd_post_send_empty(struct smbd_connection *info);
46 static int smbd_post_send_data(
47                 struct smbd_connection *info,
48                 struct kvec *iov, int n_vec, int remaining_data_length);
49 static int smbd_post_send_page(struct smbd_connection *info,
50                 struct page *page, unsigned long offset,
51                 size_t size, int remaining_data_length);
52
53 static void destroy_mr_list(struct smbd_connection *info);
54 static int allocate_mr_list(struct smbd_connection *info);
55
56 /* SMBD version number */
57 #define SMBD_V1 0x0100
58
59 /* Port numbers for SMBD transport */
60 #define SMB_PORT        445
61 #define SMBD_PORT       5445
62
63 /* Address lookup and resolve timeout in ms */
64 #define RDMA_RESOLVE_TIMEOUT    5000
65
66 /* SMBD negotiation timeout in seconds */
67 #define SMBD_NEGOTIATE_TIMEOUT  120
68
69 /* SMBD minimum receive size and fragmented sized defined in [MS-SMBD] */
70 #define SMBD_MIN_RECEIVE_SIZE           128
71 #define SMBD_MIN_FRAGMENTED_SIZE        131072
72
73 /*
74  * Default maximum number of RDMA read/write outstanding on this connection
75  * This value is possibly decreased during QP creation on hardware limit
76  */
77 #define SMBD_CM_RESPONDER_RESOURCES     32
78
79 /* Maximum number of retries on data transfer operations */
80 #define SMBD_CM_RETRY                   6
81 /* No need to retry on Receiver Not Ready since SMBD manages credits */
82 #define SMBD_CM_RNR_RETRY               0
83
84 /*
85  * User configurable initial values per SMBD transport connection
86  * as defined in [MS-SMBD] 3.1.1.1
87  * Those may change after a SMBD negotiation
88  */
89 /* The local peer's maximum number of credits to grant to the peer */
90 int smbd_receive_credit_max = 255;
91
92 /* The remote peer's credit request of local peer */
93 int smbd_send_credit_target = 255;
94
95 /* The maximum single message size can be sent to remote peer */
96 int smbd_max_send_size = 1364;
97
98 /*  The maximum fragmented upper-layer payload receive size supported */
99 int smbd_max_fragmented_recv_size = 1024 * 1024;
100
101 /*  The maximum single-message size which can be received */
102 int smbd_max_receive_size = 8192;
103
104 /* The timeout to initiate send of a keepalive message on idle */
105 int smbd_keep_alive_interval = 120;
106
107 /*
108  * User configurable initial values for RDMA transport
109  * The actual values used may be lower and are limited to hardware capabilities
110  */
111 /* Default maximum number of SGEs in a RDMA write/read */
112 int smbd_max_frmr_depth = 2048;
113
114 /* If payload is less than this byte, use RDMA send/recv not read/write */
115 int rdma_readwrite_threshold = 4096;
116
117 /* Transport logging functions
118  * Logging are defined as classes. They can be OR'ed to define the actual
119  * logging level via module parameter smbd_logging_class
120  * e.g. cifs.smbd_logging_class=0xa0 will log all log_rdma_recv() and
121  * log_rdma_event()
122  */
123 #define LOG_OUTGOING                    0x1
124 #define LOG_INCOMING                    0x2
125 #define LOG_READ                        0x4
126 #define LOG_WRITE                       0x8
127 #define LOG_RDMA_SEND                   0x10
128 #define LOG_RDMA_RECV                   0x20
129 #define LOG_KEEP_ALIVE                  0x40
130 #define LOG_RDMA_EVENT                  0x80
131 #define LOG_RDMA_MR                     0x100
132 static unsigned int smbd_logging_class;
133 module_param(smbd_logging_class, uint, 0644);
134 MODULE_PARM_DESC(smbd_logging_class,
135         "Logging class for SMBD transport 0x0 to 0x100");
136
137 #define ERR             0x0
138 #define INFO            0x1
139 static unsigned int smbd_logging_level = ERR;
140 module_param(smbd_logging_level, uint, 0644);
141 MODULE_PARM_DESC(smbd_logging_level,
142         "Logging level for SMBD transport, 0 (default): error, 1: info");
143
144 #define log_rdma(level, class, fmt, args...)                            \
145 do {                                                                    \
146         if (level <= smbd_logging_level || class & smbd_logging_class)  \
147                 cifs_dbg(VFS, "%s:%d " fmt, __func__, __LINE__, ##args);\
148 } while (0)
149
150 #define log_outgoing(level, fmt, args...) \
151                 log_rdma(level, LOG_OUTGOING, fmt, ##args)
152 #define log_incoming(level, fmt, args...) \
153                 log_rdma(level, LOG_INCOMING, fmt, ##args)
154 #define log_read(level, fmt, args...)   log_rdma(level, LOG_READ, fmt, ##args)
155 #define log_write(level, fmt, args...)  log_rdma(level, LOG_WRITE, fmt, ##args)
156 #define log_rdma_send(level, fmt, args...) \
157                 log_rdma(level, LOG_RDMA_SEND, fmt, ##args)
158 #define log_rdma_recv(level, fmt, args...) \
159                 log_rdma(level, LOG_RDMA_RECV, fmt, ##args)
160 #define log_keep_alive(level, fmt, args...) \
161                 log_rdma(level, LOG_KEEP_ALIVE, fmt, ##args)
162 #define log_rdma_event(level, fmt, args...) \
163                 log_rdma(level, LOG_RDMA_EVENT, fmt, ##args)
164 #define log_rdma_mr(level, fmt, args...) \
165                 log_rdma(level, LOG_RDMA_MR, fmt, ##args)
166
167 static void smbd_disconnect_rdma_work(struct work_struct *work)
168 {
169         struct smbd_connection *info =
170                 container_of(work, struct smbd_connection, disconnect_work);
171
172         if (info->transport_status == SMBD_CONNECTED) {
173                 info->transport_status = SMBD_DISCONNECTING;
174                 rdma_disconnect(info->id);
175         }
176 }
177
178 static void smbd_disconnect_rdma_connection(struct smbd_connection *info)
179 {
180         queue_work(info->workqueue, &info->disconnect_work);
181 }
182
183 /* Upcall from RDMA CM */
184 static int smbd_conn_upcall(
185                 struct rdma_cm_id *id, struct rdma_cm_event *event)
186 {
187         struct smbd_connection *info = id->context;
188
189         log_rdma_event(INFO, "event=%d status=%d\n",
190                 event->event, event->status);
191
192         switch (event->event) {
193         case RDMA_CM_EVENT_ADDR_RESOLVED:
194         case RDMA_CM_EVENT_ROUTE_RESOLVED:
195                 info->ri_rc = 0;
196                 complete(&info->ri_done);
197                 break;
198
199         case RDMA_CM_EVENT_ADDR_ERROR:
200                 info->ri_rc = -EHOSTUNREACH;
201                 complete(&info->ri_done);
202                 break;
203
204         case RDMA_CM_EVENT_ROUTE_ERROR:
205                 info->ri_rc = -ENETUNREACH;
206                 complete(&info->ri_done);
207                 break;
208
209         case RDMA_CM_EVENT_ESTABLISHED:
210                 log_rdma_event(INFO, "connected event=%d\n", event->event);
211                 info->transport_status = SMBD_CONNECTED;
212                 wake_up_interruptible(&info->conn_wait);
213                 break;
214
215         case RDMA_CM_EVENT_CONNECT_ERROR:
216         case RDMA_CM_EVENT_UNREACHABLE:
217         case RDMA_CM_EVENT_REJECTED:
218                 log_rdma_event(INFO, "connecting failed event=%d\n", event->event);
219                 info->transport_status = SMBD_DISCONNECTED;
220                 wake_up_interruptible(&info->conn_wait);
221                 break;
222
223         case RDMA_CM_EVENT_DEVICE_REMOVAL:
224         case RDMA_CM_EVENT_DISCONNECTED:
225                 /* This happenes when we fail the negotiation */
226                 if (info->transport_status == SMBD_NEGOTIATE_FAILED) {
227                         info->transport_status = SMBD_DISCONNECTED;
228                         wake_up(&info->conn_wait);
229                         break;
230                 }
231
232                 info->transport_status = SMBD_DISCONNECTED;
233                 wake_up_interruptible(&info->disconn_wait);
234                 wake_up_interruptible(&info->wait_reassembly_queue);
235                 wake_up_interruptible_all(&info->wait_send_queue);
236                 break;
237
238         default:
239                 break;
240         }
241
242         return 0;
243 }
244
245 /* Upcall from RDMA QP */
246 static void
247 smbd_qp_async_error_upcall(struct ib_event *event, void *context)
248 {
249         struct smbd_connection *info = context;
250
251         log_rdma_event(ERR, "%s on device %s info %p\n",
252                 ib_event_msg(event->event), event->device->name, info);
253
254         switch (event->event) {
255         case IB_EVENT_CQ_ERR:
256         case IB_EVENT_QP_FATAL:
257                 smbd_disconnect_rdma_connection(info);
258
259         default:
260                 break;
261         }
262 }
263
264 static inline void *smbd_request_payload(struct smbd_request *request)
265 {
266         return (void *)request->packet;
267 }
268
269 static inline void *smbd_response_payload(struct smbd_response *response)
270 {
271         return (void *)response->packet;
272 }
273
274 /* Called when a RDMA send is done */
275 static void send_done(struct ib_cq *cq, struct ib_wc *wc)
276 {
277         int i;
278         struct smbd_request *request =
279                 container_of(wc->wr_cqe, struct smbd_request, cqe);
280
281         log_rdma_send(INFO, "smbd_request %p completed wc->status=%d\n",
282                 request, wc->status);
283
284         if (wc->status != IB_WC_SUCCESS || wc->opcode != IB_WC_SEND) {
285                 log_rdma_send(ERR, "wc->status=%d wc->opcode=%d\n",
286                         wc->status, wc->opcode);
287                 smbd_disconnect_rdma_connection(request->info);
288         }
289
290         for (i = 0; i < request->num_sge; i++)
291                 ib_dma_unmap_single(request->info->id->device,
292                         request->sge[i].addr,
293                         request->sge[i].length,
294                         DMA_TO_DEVICE);
295
296         if (request->has_payload) {
297                 if (atomic_dec_and_test(&request->info->send_payload_pending))
298                         wake_up(&request->info->wait_send_payload_pending);
299         } else {
300                 if (atomic_dec_and_test(&request->info->send_pending))
301                         wake_up(&request->info->wait_send_pending);
302         }
303
304         mempool_free(request, request->info->request_mempool);
305 }
306
307 static void dump_smbd_negotiate_resp(struct smbd_negotiate_resp *resp)
308 {
309         log_rdma_event(INFO, "resp message min_version %u max_version %u "
310                 "negotiated_version %u credits_requested %u "
311                 "credits_granted %u status %u max_readwrite_size %u "
312                 "preferred_send_size %u max_receive_size %u "
313                 "max_fragmented_size %u\n",
314                 resp->min_version, resp->max_version, resp->negotiated_version,
315                 resp->credits_requested, resp->credits_granted, resp->status,
316                 resp->max_readwrite_size, resp->preferred_send_size,
317                 resp->max_receive_size, resp->max_fragmented_size);
318 }
319
320 /*
321  * Process a negotiation response message, according to [MS-SMBD]3.1.5.7
322  * response, packet_length: the negotiation response message
323  * return value: true if negotiation is a success, false if failed
324  */
325 static bool process_negotiation_response(
326                 struct smbd_response *response, int packet_length)
327 {
328         struct smbd_connection *info = response->info;
329         struct smbd_negotiate_resp *packet = smbd_response_payload(response);
330
331         if (packet_length < sizeof(struct smbd_negotiate_resp)) {
332                 log_rdma_event(ERR,
333                         "error: packet_length=%d\n", packet_length);
334                 return false;
335         }
336
337         if (le16_to_cpu(packet->negotiated_version) != SMBD_V1) {
338                 log_rdma_event(ERR, "error: negotiated_version=%x\n",
339                         le16_to_cpu(packet->negotiated_version));
340                 return false;
341         }
342         info->protocol = le16_to_cpu(packet->negotiated_version);
343
344         if (packet->credits_requested == 0) {
345                 log_rdma_event(ERR, "error: credits_requested==0\n");
346                 return false;
347         }
348         info->receive_credit_target = le16_to_cpu(packet->credits_requested);
349
350         if (packet->credits_granted == 0) {
351                 log_rdma_event(ERR, "error: credits_granted==0\n");
352                 return false;
353         }
354         atomic_set(&info->send_credits, le16_to_cpu(packet->credits_granted));
355
356         atomic_set(&info->receive_credits, 0);
357
358         if (le32_to_cpu(packet->preferred_send_size) > info->max_receive_size) {
359                 log_rdma_event(ERR, "error: preferred_send_size=%d\n",
360                         le32_to_cpu(packet->preferred_send_size));
361                 return false;
362         }
363         info->max_receive_size = le32_to_cpu(packet->preferred_send_size);
364
365         if (le32_to_cpu(packet->max_receive_size) < SMBD_MIN_RECEIVE_SIZE) {
366                 log_rdma_event(ERR, "error: max_receive_size=%d\n",
367                         le32_to_cpu(packet->max_receive_size));
368                 return false;
369         }
370         info->max_send_size = min_t(int, info->max_send_size,
371                                         le32_to_cpu(packet->max_receive_size));
372
373         if (le32_to_cpu(packet->max_fragmented_size) <
374                         SMBD_MIN_FRAGMENTED_SIZE) {
375                 log_rdma_event(ERR, "error: max_fragmented_size=%d\n",
376                         le32_to_cpu(packet->max_fragmented_size));
377                 return false;
378         }
379         info->max_fragmented_send_size =
380                 le32_to_cpu(packet->max_fragmented_size);
381         info->rdma_readwrite_threshold =
382                 rdma_readwrite_threshold > info->max_fragmented_send_size ?
383                 info->max_fragmented_send_size :
384                 rdma_readwrite_threshold;
385
386
387         info->max_readwrite_size = min_t(u32,
388                         le32_to_cpu(packet->max_readwrite_size),
389                         info->max_frmr_depth * PAGE_SIZE);
390         info->max_frmr_depth = info->max_readwrite_size / PAGE_SIZE;
391
392         return true;
393 }
394
395 /*
396  * Check and schedule to send an immediate packet
397  * This is used to extend credtis to remote peer to keep the transport busy
398  */
399 static void check_and_send_immediate(struct smbd_connection *info)
400 {
401         if (info->transport_status != SMBD_CONNECTED)
402                 return;
403
404         info->send_immediate = true;
405
406         /*
407          * Promptly send a packet if our peer is running low on receive
408          * credits
409          */
410         if (atomic_read(&info->receive_credits) <
411                 info->receive_credit_target - 1)
412                 queue_delayed_work(
413                         info->workqueue, &info->send_immediate_work, 0);
414 }
415
416 static void smbd_post_send_credits(struct work_struct *work)
417 {
418         int ret = 0;
419         int use_receive_queue = 1;
420         int rc;
421         struct smbd_response *response;
422         struct smbd_connection *info =
423                 container_of(work, struct smbd_connection,
424                         post_send_credits_work);
425
426         if (info->transport_status != SMBD_CONNECTED) {
427                 wake_up(&info->wait_receive_queues);
428                 return;
429         }
430
431         if (info->receive_credit_target >
432                 atomic_read(&info->receive_credits)) {
433                 while (true) {
434                         if (use_receive_queue)
435                                 response = get_receive_buffer(info);
436                         else
437                                 response = get_empty_queue_buffer(info);
438                         if (!response) {
439                                 /* now switch to emtpy packet queue */
440                                 if (use_receive_queue) {
441                                         use_receive_queue = 0;
442                                         continue;
443                                 } else
444                                         break;
445                         }
446
447                         response->type = SMBD_TRANSFER_DATA;
448                         response->first_segment = false;
449                         rc = smbd_post_recv(info, response);
450                         if (rc) {
451                                 log_rdma_recv(ERR,
452                                         "post_recv failed rc=%d\n", rc);
453                                 put_receive_buffer(info, response);
454                                 break;
455                         }
456
457                         ret++;
458                 }
459         }
460
461         spin_lock(&info->lock_new_credits_offered);
462         info->new_credits_offered += ret;
463         spin_unlock(&info->lock_new_credits_offered);
464
465         atomic_add(ret, &info->receive_credits);
466
467         /* Check if we can post new receive and grant credits to peer */
468         check_and_send_immediate(info);
469 }
470
471 static void smbd_recv_done_work(struct work_struct *work)
472 {
473         struct smbd_connection *info =
474                 container_of(work, struct smbd_connection, recv_done_work);
475
476         /*
477          * We may have new send credits granted from remote peer
478          * If any sender is blcoked on lack of credets, unblock it
479          */
480         if (atomic_read(&info->send_credits))
481                 wake_up_interruptible(&info->wait_send_queue);
482
483         /*
484          * Check if we need to send something to remote peer to
485          * grant more credits or respond to KEEP_ALIVE packet
486          */
487         check_and_send_immediate(info);
488 }
489
490 /* Called from softirq, when recv is done */
491 static void recv_done(struct ib_cq *cq, struct ib_wc *wc)
492 {
493         struct smbd_data_transfer *data_transfer;
494         struct smbd_response *response =
495                 container_of(wc->wr_cqe, struct smbd_response, cqe);
496         struct smbd_connection *info = response->info;
497         int data_length = 0;
498
499         log_rdma_recv(INFO, "response=%p type=%d wc status=%d wc opcode %d "
500                       "byte_len=%d pkey_index=%x\n",
501                 response, response->type, wc->status, wc->opcode,
502                 wc->byte_len, wc->pkey_index);
503
504         if (wc->status != IB_WC_SUCCESS || wc->opcode != IB_WC_RECV) {
505                 log_rdma_recv(INFO, "wc->status=%d opcode=%d\n",
506                         wc->status, wc->opcode);
507                 smbd_disconnect_rdma_connection(info);
508                 goto error;
509         }
510
511         ib_dma_sync_single_for_cpu(
512                 wc->qp->device,
513                 response->sge.addr,
514                 response->sge.length,
515                 DMA_FROM_DEVICE);
516
517         switch (response->type) {
518         /* SMBD negotiation response */
519         case SMBD_NEGOTIATE_RESP:
520                 dump_smbd_negotiate_resp(smbd_response_payload(response));
521                 info->full_packet_received = true;
522                 info->negotiate_done =
523                         process_negotiation_response(response, wc->byte_len);
524                 complete(&info->negotiate_completion);
525                 break;
526
527         /* SMBD data transfer packet */
528         case SMBD_TRANSFER_DATA:
529                 data_transfer = smbd_response_payload(response);
530                 data_length = le32_to_cpu(data_transfer->data_length);
531
532                 /*
533                  * If this is a packet with data playload place the data in
534                  * reassembly queue and wake up the reading thread
535                  */
536                 if (data_length) {
537                         if (info->full_packet_received)
538                                 response->first_segment = true;
539
540                         if (le32_to_cpu(data_transfer->remaining_data_length))
541                                 info->full_packet_received = false;
542                         else
543                                 info->full_packet_received = true;
544
545                         enqueue_reassembly(
546                                 info,
547                                 response,
548                                 data_length);
549                 } else
550                         put_empty_packet(info, response);
551
552                 if (data_length)
553                         wake_up_interruptible(&info->wait_reassembly_queue);
554
555                 atomic_dec(&info->receive_credits);
556                 info->receive_credit_target =
557                         le16_to_cpu(data_transfer->credits_requested);
558                 atomic_add(le16_to_cpu(data_transfer->credits_granted),
559                         &info->send_credits);
560
561                 log_incoming(INFO, "data flags %d data_offset %d "
562                         "data_length %d remaining_data_length %d\n",
563                         le16_to_cpu(data_transfer->flags),
564                         le32_to_cpu(data_transfer->data_offset),
565                         le32_to_cpu(data_transfer->data_length),
566                         le32_to_cpu(data_transfer->remaining_data_length));
567
568                 /* Send a KEEP_ALIVE response right away if requested */
569                 info->keep_alive_requested = KEEP_ALIVE_NONE;
570                 if (le16_to_cpu(data_transfer->flags) &
571                                 SMB_DIRECT_RESPONSE_REQUESTED) {
572                         info->keep_alive_requested = KEEP_ALIVE_PENDING;
573                 }
574
575                 queue_work(info->workqueue, &info->recv_done_work);
576                 return;
577
578         default:
579                 log_rdma_recv(ERR,
580                         "unexpected response type=%d\n", response->type);
581         }
582
583 error:
584         put_receive_buffer(info, response);
585 }
586
587 static struct rdma_cm_id *smbd_create_id(
588                 struct smbd_connection *info,
589                 struct sockaddr *dstaddr, int port)
590 {
591         struct rdma_cm_id *id;
592         int rc;
593         __be16 *sport;
594
595         id = rdma_create_id(&init_net, smbd_conn_upcall, info,
596                 RDMA_PS_TCP, IB_QPT_RC);
597         if (IS_ERR(id)) {
598                 rc = PTR_ERR(id);
599                 log_rdma_event(ERR, "rdma_create_id() failed %i\n", rc);
600                 return id;
601         }
602
603         if (dstaddr->sa_family == AF_INET6)
604                 sport = &((struct sockaddr_in6 *)dstaddr)->sin6_port;
605         else
606                 sport = &((struct sockaddr_in *)dstaddr)->sin_port;
607
608         *sport = htons(port);
609
610         init_completion(&info->ri_done);
611         info->ri_rc = -ETIMEDOUT;
612
613         rc = rdma_resolve_addr(id, NULL, (struct sockaddr *)dstaddr,
614                 RDMA_RESOLVE_TIMEOUT);
615         if (rc) {
616                 log_rdma_event(ERR, "rdma_resolve_addr() failed %i\n", rc);
617                 goto out;
618         }
619         wait_for_completion_interruptible_timeout(
620                 &info->ri_done, msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT));
621         rc = info->ri_rc;
622         if (rc) {
623                 log_rdma_event(ERR, "rdma_resolve_addr() completed %i\n", rc);
624                 goto out;
625         }
626
627         info->ri_rc = -ETIMEDOUT;
628         rc = rdma_resolve_route(id, RDMA_RESOLVE_TIMEOUT);
629         if (rc) {
630                 log_rdma_event(ERR, "rdma_resolve_route() failed %i\n", rc);
631                 goto out;
632         }
633         wait_for_completion_interruptible_timeout(
634                 &info->ri_done, msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT));
635         rc = info->ri_rc;
636         if (rc) {
637                 log_rdma_event(ERR, "rdma_resolve_route() completed %i\n", rc);
638                 goto out;
639         }
640
641         return id;
642
643 out:
644         rdma_destroy_id(id);
645         return ERR_PTR(rc);
646 }
647
648 /*
649  * Test if FRWR (Fast Registration Work Requests) is supported on the device
650  * This implementation requries FRWR on RDMA read/write
651  * return value: true if it is supported
652  */
653 static bool frwr_is_supported(struct ib_device_attr *attrs)
654 {
655         if (!(attrs->device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS))
656                 return false;
657         if (attrs->max_fast_reg_page_list_len == 0)
658                 return false;
659         return true;
660 }
661
662 static int smbd_ia_open(
663                 struct smbd_connection *info,
664                 struct sockaddr *dstaddr, int port)
665 {
666         int rc;
667
668         info->id = smbd_create_id(info, dstaddr, port);
669         if (IS_ERR(info->id)) {
670                 rc = PTR_ERR(info->id);
671                 goto out1;
672         }
673
674         if (!frwr_is_supported(&info->id->device->attrs)) {
675                 log_rdma_event(ERR,
676                         "Fast Registration Work Requests "
677                         "(FRWR) is not supported\n");
678                 log_rdma_event(ERR,
679                         "Device capability flags = %llx "
680                         "max_fast_reg_page_list_len = %u\n",
681                         info->id->device->attrs.device_cap_flags,
682                         info->id->device->attrs.max_fast_reg_page_list_len);
683                 rc = -EPROTONOSUPPORT;
684                 goto out2;
685         }
686         info->max_frmr_depth = min_t(int,
687                 smbd_max_frmr_depth,
688                 info->id->device->attrs.max_fast_reg_page_list_len);
689         info->mr_type = IB_MR_TYPE_MEM_REG;
690         if (info->id->device->attrs.device_cap_flags & IB_DEVICE_SG_GAPS_REG)
691                 info->mr_type = IB_MR_TYPE_SG_GAPS;
692
693         info->pd = ib_alloc_pd(info->id->device, 0);
694         if (IS_ERR(info->pd)) {
695                 rc = PTR_ERR(info->pd);
696                 log_rdma_event(ERR, "ib_alloc_pd() returned %d\n", rc);
697                 goto out2;
698         }
699
700         return 0;
701
702 out2:
703         rdma_destroy_id(info->id);
704         info->id = NULL;
705
706 out1:
707         return rc;
708 }
709
710 /*
711  * Send a negotiation request message to the peer
712  * The negotiation procedure is in [MS-SMBD] 3.1.5.2 and 3.1.5.3
713  * After negotiation, the transport is connected and ready for
714  * carrying upper layer SMB payload
715  */
716 static int smbd_post_send_negotiate_req(struct smbd_connection *info)
717 {
718         struct ib_send_wr send_wr;
719         int rc = -ENOMEM;
720         struct smbd_request *request;
721         struct smbd_negotiate_req *packet;
722
723         request = mempool_alloc(info->request_mempool, GFP_KERNEL);
724         if (!request)
725                 return rc;
726
727         request->info = info;
728
729         packet = smbd_request_payload(request);
730         packet->min_version = cpu_to_le16(SMBD_V1);
731         packet->max_version = cpu_to_le16(SMBD_V1);
732         packet->reserved = 0;
733         packet->credits_requested = cpu_to_le16(info->send_credit_target);
734         packet->preferred_send_size = cpu_to_le32(info->max_send_size);
735         packet->max_receive_size = cpu_to_le32(info->max_receive_size);
736         packet->max_fragmented_size =
737                 cpu_to_le32(info->max_fragmented_recv_size);
738
739         request->num_sge = 1;
740         request->sge[0].addr = ib_dma_map_single(
741                                 info->id->device, (void *)packet,
742                                 sizeof(*packet), DMA_TO_DEVICE);
743         if (ib_dma_mapping_error(info->id->device, request->sge[0].addr)) {
744                 rc = -EIO;
745                 goto dma_mapping_failed;
746         }
747
748         request->sge[0].length = sizeof(*packet);
749         request->sge[0].lkey = info->pd->local_dma_lkey;
750
751         ib_dma_sync_single_for_device(
752                 info->id->device, request->sge[0].addr,
753                 request->sge[0].length, DMA_TO_DEVICE);
754
755         request->cqe.done = send_done;
756
757         send_wr.next = NULL;
758         send_wr.wr_cqe = &request->cqe;
759         send_wr.sg_list = request->sge;
760         send_wr.num_sge = request->num_sge;
761         send_wr.opcode = IB_WR_SEND;
762         send_wr.send_flags = IB_SEND_SIGNALED;
763
764         log_rdma_send(INFO, "sge addr=%llx length=%x lkey=%x\n",
765                 request->sge[0].addr,
766                 request->sge[0].length, request->sge[0].lkey);
767
768         request->has_payload = false;
769         atomic_inc(&info->send_pending);
770         rc = ib_post_send(info->id->qp, &send_wr, NULL);
771         if (!rc)
772                 return 0;
773
774         /* if we reach here, post send failed */
775         log_rdma_send(ERR, "ib_post_send failed rc=%d\n", rc);
776         atomic_dec(&info->send_pending);
777         ib_dma_unmap_single(info->id->device, request->sge[0].addr,
778                 request->sge[0].length, DMA_TO_DEVICE);
779
780         smbd_disconnect_rdma_connection(info);
781
782 dma_mapping_failed:
783         mempool_free(request, info->request_mempool);
784         return rc;
785 }
786
787 /*
788  * Extend the credits to remote peer
789  * This implements [MS-SMBD] 3.1.5.9
790  * The idea is that we should extend credits to remote peer as quickly as
791  * it's allowed, to maintain data flow. We allocate as much receive
792  * buffer as possible, and extend the receive credits to remote peer
793  * return value: the new credtis being granted.
794  */
795 static int manage_credits_prior_sending(struct smbd_connection *info)
796 {
797         int new_credits;
798
799         spin_lock(&info->lock_new_credits_offered);
800         new_credits = info->new_credits_offered;
801         info->new_credits_offered = 0;
802         spin_unlock(&info->lock_new_credits_offered);
803
804         return new_credits;
805 }
806
807 /*
808  * Check if we need to send a KEEP_ALIVE message
809  * The idle connection timer triggers a KEEP_ALIVE message when expires
810  * SMB_DIRECT_RESPONSE_REQUESTED is set in the message flag to have peer send
811  * back a response.
812  * return value:
813  * 1 if SMB_DIRECT_RESPONSE_REQUESTED needs to be set
814  * 0: otherwise
815  */
816 static int manage_keep_alive_before_sending(struct smbd_connection *info)
817 {
818         if (info->keep_alive_requested == KEEP_ALIVE_PENDING) {
819                 info->keep_alive_requested = KEEP_ALIVE_SENT;
820                 return 1;
821         }
822         return 0;
823 }
824
825 /*
826  * Build and prepare the SMBD packet header
827  * This function waits for avaialbe send credits and build a SMBD packet
828  * header. The caller then optional append payload to the packet after
829  * the header
830  * intput values
831  * size: the size of the payload
832  * remaining_data_length: remaining data to send if this is part of a
833  * fragmented packet
834  * output values
835  * request_out: the request allocated from this function
836  * return values: 0 on success, otherwise actual error code returned
837  */
838 static int smbd_create_header(struct smbd_connection *info,
839                 int size, int remaining_data_length,
840                 struct smbd_request **request_out)
841 {
842         struct smbd_request *request;
843         struct smbd_data_transfer *packet;
844         int header_length;
845         int rc;
846
847         /* Wait for send credits. A SMBD packet needs one credit */
848         rc = wait_event_interruptible(info->wait_send_queue,
849                 atomic_read(&info->send_credits) > 0 ||
850                 info->transport_status != SMBD_CONNECTED);
851         if (rc)
852                 return rc;
853
854         if (info->transport_status != SMBD_CONNECTED) {
855                 log_outgoing(ERR, "disconnected not sending\n");
856                 return -ENOENT;
857         }
858         atomic_dec(&info->send_credits);
859
860         request = mempool_alloc(info->request_mempool, GFP_KERNEL);
861         if (!request) {
862                 rc = -ENOMEM;
863                 goto err;
864         }
865
866         request->info = info;
867
868         /* Fill in the packet header */
869         packet = smbd_request_payload(request);
870         packet->credits_requested = cpu_to_le16(info->send_credit_target);
871         packet->credits_granted =
872                 cpu_to_le16(manage_credits_prior_sending(info));
873         info->send_immediate = false;
874
875         packet->flags = 0;
876         if (manage_keep_alive_before_sending(info))
877                 packet->flags |= cpu_to_le16(SMB_DIRECT_RESPONSE_REQUESTED);
878
879         packet->reserved = 0;
880         if (!size)
881                 packet->data_offset = 0;
882         else
883                 packet->data_offset = cpu_to_le32(24);
884         packet->data_length = cpu_to_le32(size);
885         packet->remaining_data_length = cpu_to_le32(remaining_data_length);
886         packet->padding = 0;
887
888         log_outgoing(INFO, "credits_requested=%d credits_granted=%d "
889                 "data_offset=%d data_length=%d remaining_data_length=%d\n",
890                 le16_to_cpu(packet->credits_requested),
891                 le16_to_cpu(packet->credits_granted),
892                 le32_to_cpu(packet->data_offset),
893                 le32_to_cpu(packet->data_length),
894                 le32_to_cpu(packet->remaining_data_length));
895
896         /* Map the packet to DMA */
897         header_length = sizeof(struct smbd_data_transfer);
898         /* If this is a packet without payload, don't send padding */
899         if (!size)
900                 header_length = offsetof(struct smbd_data_transfer, padding);
901
902         request->num_sge = 1;
903         request->sge[0].addr = ib_dma_map_single(info->id->device,
904                                                  (void *)packet,
905                                                  header_length,
906                                                  DMA_BIDIRECTIONAL);
907         if (ib_dma_mapping_error(info->id->device, request->sge[0].addr)) {
908                 mempool_free(request, info->request_mempool);
909                 rc = -EIO;
910                 goto err;
911         }
912
913         request->sge[0].length = header_length;
914         request->sge[0].lkey = info->pd->local_dma_lkey;
915
916         *request_out = request;
917         return 0;
918
919 err:
920         atomic_inc(&info->send_credits);
921         return rc;
922 }
923
924 static void smbd_destroy_header(struct smbd_connection *info,
925                 struct smbd_request *request)
926 {
927
928         ib_dma_unmap_single(info->id->device,
929                             request->sge[0].addr,
930                             request->sge[0].length,
931                             DMA_TO_DEVICE);
932         mempool_free(request, info->request_mempool);
933         atomic_inc(&info->send_credits);
934 }
935
936 /* Post the send request */
937 static int smbd_post_send(struct smbd_connection *info,
938                 struct smbd_request *request, bool has_payload)
939 {
940         struct ib_send_wr send_wr;
941         int rc, i;
942
943         for (i = 0; i < request->num_sge; i++) {
944                 log_rdma_send(INFO,
945                         "rdma_request sge[%d] addr=%llu length=%u\n",
946                         i, request->sge[i].addr, request->sge[i].length);
947                 ib_dma_sync_single_for_device(
948                         info->id->device,
949                         request->sge[i].addr,
950                         request->sge[i].length,
951                         DMA_TO_DEVICE);
952         }
953
954         request->cqe.done = send_done;
955
956         send_wr.next = NULL;
957         send_wr.wr_cqe = &request->cqe;
958         send_wr.sg_list = request->sge;
959         send_wr.num_sge = request->num_sge;
960         send_wr.opcode = IB_WR_SEND;
961         send_wr.send_flags = IB_SEND_SIGNALED;
962
963         if (has_payload) {
964                 request->has_payload = true;
965                 atomic_inc(&info->send_payload_pending);
966         } else {
967                 request->has_payload = false;
968                 atomic_inc(&info->send_pending);
969         }
970
971         rc = ib_post_send(info->id->qp, &send_wr, NULL);
972         if (rc) {
973                 log_rdma_send(ERR, "ib_post_send failed rc=%d\n", rc);
974                 if (has_payload) {
975                         if (atomic_dec_and_test(&info->send_payload_pending))
976                                 wake_up(&info->wait_send_payload_pending);
977                 } else {
978                         if (atomic_dec_and_test(&info->send_pending))
979                                 wake_up(&info->wait_send_pending);
980                 }
981                 smbd_disconnect_rdma_connection(info);
982         } else
983                 /* Reset timer for idle connection after packet is sent */
984                 mod_delayed_work(info->workqueue, &info->idle_timer_work,
985                         info->keep_alive_interval*HZ);
986
987         return rc;
988 }
989
990 static int smbd_post_send_sgl(struct smbd_connection *info,
991         struct scatterlist *sgl, int data_length, int remaining_data_length)
992 {
993         int num_sgs;
994         int i, rc;
995         struct smbd_request *request;
996         struct scatterlist *sg;
997
998         rc = smbd_create_header(
999                 info, data_length, remaining_data_length, &request);
1000         if (rc)
1001                 return rc;
1002
1003         num_sgs = sgl ? sg_nents(sgl) : 0;
1004         for_each_sg(sgl, sg, num_sgs, i) {
1005                 request->sge[i+1].addr =
1006                         ib_dma_map_page(info->id->device, sg_page(sg),
1007                                sg->offset, sg->length, DMA_BIDIRECTIONAL);
1008                 if (ib_dma_mapping_error(
1009                                 info->id->device, request->sge[i+1].addr)) {
1010                         rc = -EIO;
1011                         request->sge[i+1].addr = 0;
1012                         goto dma_mapping_failure;
1013                 }
1014                 request->sge[i+1].length = sg->length;
1015                 request->sge[i+1].lkey = info->pd->local_dma_lkey;
1016                 request->num_sge++;
1017         }
1018
1019         rc = smbd_post_send(info, request, data_length);
1020         if (!rc)
1021                 return 0;
1022
1023 dma_mapping_failure:
1024         for (i = 1; i < request->num_sge; i++)
1025                 if (request->sge[i].addr)
1026                         ib_dma_unmap_single(info->id->device,
1027                                             request->sge[i].addr,
1028                                             request->sge[i].length,
1029                                             DMA_TO_DEVICE);
1030         smbd_destroy_header(info, request);
1031         return rc;
1032 }
1033
1034 /*
1035  * Send a page
1036  * page: the page to send
1037  * offset: offset in the page to send
1038  * size: length in the page to send
1039  * remaining_data_length: remaining data to send in this payload
1040  */
1041 static int smbd_post_send_page(struct smbd_connection *info, struct page *page,
1042                 unsigned long offset, size_t size, int remaining_data_length)
1043 {
1044         struct scatterlist sgl;
1045
1046         sg_init_table(&sgl, 1);
1047         sg_set_page(&sgl, page, size, offset);
1048
1049         return smbd_post_send_sgl(info, &sgl, size, remaining_data_length);
1050 }
1051
1052 /*
1053  * Send an empty message
1054  * Empty message is used to extend credits to peer to for keep live
1055  * while there is no upper layer payload to send at the time
1056  */
1057 static int smbd_post_send_empty(struct smbd_connection *info)
1058 {
1059         info->count_send_empty++;
1060         return smbd_post_send_sgl(info, NULL, 0, 0);
1061 }
1062
1063 /*
1064  * Send a data buffer
1065  * iov: the iov array describing the data buffers
1066  * n_vec: number of iov array
1067  * remaining_data_length: remaining data to send following this packet
1068  * in segmented SMBD packet
1069  */
1070 static int smbd_post_send_data(
1071         struct smbd_connection *info, struct kvec *iov, int n_vec,
1072         int remaining_data_length)
1073 {
1074         int i;
1075         u32 data_length = 0;
1076         struct scatterlist sgl[SMBDIRECT_MAX_SGE];
1077
1078         if (n_vec > SMBDIRECT_MAX_SGE) {
1079                 cifs_dbg(VFS, "Can't fit data to SGL, n_vec=%d\n", n_vec);
1080                 return -ENOMEM;
1081         }
1082
1083         sg_init_table(sgl, n_vec);
1084         for (i = 0; i < n_vec; i++) {
1085                 data_length += iov[i].iov_len;
1086                 sg_set_buf(&sgl[i], iov[i].iov_base, iov[i].iov_len);
1087         }
1088
1089         return smbd_post_send_sgl(info, sgl, data_length, remaining_data_length);
1090 }
1091
1092 /*
1093  * Post a receive request to the transport
1094  * The remote peer can only send data when a receive request is posted
1095  * The interaction is controlled by send/receive credit system
1096  */
1097 static int smbd_post_recv(
1098                 struct smbd_connection *info, struct smbd_response *response)
1099 {
1100         struct ib_recv_wr recv_wr;
1101         int rc = -EIO;
1102
1103         response->sge.addr = ib_dma_map_single(
1104                                 info->id->device, response->packet,
1105                                 info->max_receive_size, DMA_FROM_DEVICE);
1106         if (ib_dma_mapping_error(info->id->device, response->sge.addr))
1107                 return rc;
1108
1109         response->sge.length = info->max_receive_size;
1110         response->sge.lkey = info->pd->local_dma_lkey;
1111
1112         response->cqe.done = recv_done;
1113
1114         recv_wr.wr_cqe = &response->cqe;
1115         recv_wr.next = NULL;
1116         recv_wr.sg_list = &response->sge;
1117         recv_wr.num_sge = 1;
1118
1119         rc = ib_post_recv(info->id->qp, &recv_wr, NULL);
1120         if (rc) {
1121                 ib_dma_unmap_single(info->id->device, response->sge.addr,
1122                                     response->sge.length, DMA_FROM_DEVICE);
1123                 smbd_disconnect_rdma_connection(info);
1124                 log_rdma_recv(ERR, "ib_post_recv failed rc=%d\n", rc);
1125         }
1126
1127         return rc;
1128 }
1129
1130 /* Perform SMBD negotiate according to [MS-SMBD] 3.1.5.2 */
1131 static int smbd_negotiate(struct smbd_connection *info)
1132 {
1133         int rc;
1134         struct smbd_response *response = get_receive_buffer(info);
1135
1136         response->type = SMBD_NEGOTIATE_RESP;
1137         rc = smbd_post_recv(info, response);
1138         log_rdma_event(INFO,
1139                 "smbd_post_recv rc=%d iov.addr=%llx iov.length=%x "
1140                 "iov.lkey=%x\n",
1141                 rc, response->sge.addr,
1142                 response->sge.length, response->sge.lkey);
1143         if (rc)
1144                 return rc;
1145
1146         init_completion(&info->negotiate_completion);
1147         info->negotiate_done = false;
1148         rc = smbd_post_send_negotiate_req(info);
1149         if (rc)
1150                 return rc;
1151
1152         rc = wait_for_completion_interruptible_timeout(
1153                 &info->negotiate_completion, SMBD_NEGOTIATE_TIMEOUT * HZ);
1154         log_rdma_event(INFO, "wait_for_completion_timeout rc=%d\n", rc);
1155
1156         if (info->negotiate_done)
1157                 return 0;
1158
1159         if (rc == 0)
1160                 rc = -ETIMEDOUT;
1161         else if (rc == -ERESTARTSYS)
1162                 rc = -EINTR;
1163         else
1164                 rc = -ENOTCONN;
1165
1166         return rc;
1167 }
1168
1169 static void put_empty_packet(
1170                 struct smbd_connection *info, struct smbd_response *response)
1171 {
1172         spin_lock(&info->empty_packet_queue_lock);
1173         list_add_tail(&response->list, &info->empty_packet_queue);
1174         info->count_empty_packet_queue++;
1175         spin_unlock(&info->empty_packet_queue_lock);
1176
1177         queue_work(info->workqueue, &info->post_send_credits_work);
1178 }
1179
1180 /*
1181  * Implement Connection.FragmentReassemblyBuffer defined in [MS-SMBD] 3.1.1.1
1182  * This is a queue for reassembling upper layer payload and present to upper
1183  * layer. All the inncoming payload go to the reassembly queue, regardless of
1184  * if reassembly is required. The uuper layer code reads from the queue for all
1185  * incoming payloads.
1186  * Put a received packet to the reassembly queue
1187  * response: the packet received
1188  * data_length: the size of payload in this packet
1189  */
1190 static void enqueue_reassembly(
1191         struct smbd_connection *info,
1192         struct smbd_response *response,
1193         int data_length)
1194 {
1195         spin_lock(&info->reassembly_queue_lock);
1196         list_add_tail(&response->list, &info->reassembly_queue);
1197         info->reassembly_queue_length++;
1198         /*
1199          * Make sure reassembly_data_length is updated after list and
1200          * reassembly_queue_length are updated. On the dequeue side
1201          * reassembly_data_length is checked without a lock to determine
1202          * if reassembly_queue_length and list is up to date
1203          */
1204         virt_wmb();
1205         info->reassembly_data_length += data_length;
1206         spin_unlock(&info->reassembly_queue_lock);
1207         info->count_reassembly_queue++;
1208         info->count_enqueue_reassembly_queue++;
1209 }
1210
1211 /*
1212  * Get the first entry at the front of reassembly queue
1213  * Caller is responsible for locking
1214  * return value: the first entry if any, NULL if queue is empty
1215  */
1216 static struct smbd_response *_get_first_reassembly(struct smbd_connection *info)
1217 {
1218         struct smbd_response *ret = NULL;
1219
1220         if (!list_empty(&info->reassembly_queue)) {
1221                 ret = list_first_entry(
1222                         &info->reassembly_queue,
1223                         struct smbd_response, list);
1224         }
1225         return ret;
1226 }
1227
1228 static struct smbd_response *get_empty_queue_buffer(
1229                 struct smbd_connection *info)
1230 {
1231         struct smbd_response *ret = NULL;
1232         unsigned long flags;
1233
1234         spin_lock_irqsave(&info->empty_packet_queue_lock, flags);
1235         if (!list_empty(&info->empty_packet_queue)) {
1236                 ret = list_first_entry(
1237                         &info->empty_packet_queue,
1238                         struct smbd_response, list);
1239                 list_del(&ret->list);
1240                 info->count_empty_packet_queue--;
1241         }
1242         spin_unlock_irqrestore(&info->empty_packet_queue_lock, flags);
1243
1244         return ret;
1245 }
1246
1247 /*
1248  * Get a receive buffer
1249  * For each remote send, we need to post a receive. The receive buffers are
1250  * pre-allocated in advance.
1251  * return value: the receive buffer, NULL if none is available
1252  */
1253 static struct smbd_response *get_receive_buffer(struct smbd_connection *info)
1254 {
1255         struct smbd_response *ret = NULL;
1256         unsigned long flags;
1257
1258         spin_lock_irqsave(&info->receive_queue_lock, flags);
1259         if (!list_empty(&info->receive_queue)) {
1260                 ret = list_first_entry(
1261                         &info->receive_queue,
1262                         struct smbd_response, list);
1263                 list_del(&ret->list);
1264                 info->count_receive_queue--;
1265                 info->count_get_receive_buffer++;
1266         }
1267         spin_unlock_irqrestore(&info->receive_queue_lock, flags);
1268
1269         return ret;
1270 }
1271
1272 /*
1273  * Return a receive buffer
1274  * Upon returning of a receive buffer, we can post new receive and extend
1275  * more receive credits to remote peer. This is done immediately after a
1276  * receive buffer is returned.
1277  */
1278 static void put_receive_buffer(
1279         struct smbd_connection *info, struct smbd_response *response)
1280 {
1281         unsigned long flags;
1282
1283         ib_dma_unmap_single(info->id->device, response->sge.addr,
1284                 response->sge.length, DMA_FROM_DEVICE);
1285
1286         spin_lock_irqsave(&info->receive_queue_lock, flags);
1287         list_add_tail(&response->list, &info->receive_queue);
1288         info->count_receive_queue++;
1289         info->count_put_receive_buffer++;
1290         spin_unlock_irqrestore(&info->receive_queue_lock, flags);
1291
1292         queue_work(info->workqueue, &info->post_send_credits_work);
1293 }
1294
1295 /* Preallocate all receive buffer on transport establishment */
1296 static int allocate_receive_buffers(struct smbd_connection *info, int num_buf)
1297 {
1298         int i;
1299         struct smbd_response *response;
1300
1301         INIT_LIST_HEAD(&info->reassembly_queue);
1302         spin_lock_init(&info->reassembly_queue_lock);
1303         info->reassembly_data_length = 0;
1304         info->reassembly_queue_length = 0;
1305
1306         INIT_LIST_HEAD(&info->receive_queue);
1307         spin_lock_init(&info->receive_queue_lock);
1308         info->count_receive_queue = 0;
1309
1310         INIT_LIST_HEAD(&info->empty_packet_queue);
1311         spin_lock_init(&info->empty_packet_queue_lock);
1312         info->count_empty_packet_queue = 0;
1313
1314         init_waitqueue_head(&info->wait_receive_queues);
1315
1316         for (i = 0; i < num_buf; i++) {
1317                 response = mempool_alloc(info->response_mempool, GFP_KERNEL);
1318                 if (!response)
1319                         goto allocate_failed;
1320
1321                 response->info = info;
1322                 list_add_tail(&response->list, &info->receive_queue);
1323                 info->count_receive_queue++;
1324         }
1325
1326         return 0;
1327
1328 allocate_failed:
1329         while (!list_empty(&info->receive_queue)) {
1330                 response = list_first_entry(
1331                                 &info->receive_queue,
1332                                 struct smbd_response, list);
1333                 list_del(&response->list);
1334                 info->count_receive_queue--;
1335
1336                 mempool_free(response, info->response_mempool);
1337         }
1338         return -ENOMEM;
1339 }
1340
1341 static void destroy_receive_buffers(struct smbd_connection *info)
1342 {
1343         struct smbd_response *response;
1344
1345         while ((response = get_receive_buffer(info)))
1346                 mempool_free(response, info->response_mempool);
1347
1348         while ((response = get_empty_queue_buffer(info)))
1349                 mempool_free(response, info->response_mempool);
1350 }
1351
1352 /*
1353  * Check and send an immediate or keep alive packet
1354  * The condition to send those packets are defined in [MS-SMBD] 3.1.1.1
1355  * Connection.KeepaliveRequested and Connection.SendImmediate
1356  * The idea is to extend credits to server as soon as it becomes available
1357  */
1358 static void send_immediate_work(struct work_struct *work)
1359 {
1360         struct smbd_connection *info = container_of(
1361                                         work, struct smbd_connection,
1362                                         send_immediate_work.work);
1363
1364         if (info->keep_alive_requested == KEEP_ALIVE_PENDING ||
1365             info->send_immediate) {
1366                 log_keep_alive(INFO, "send an empty message\n");
1367                 smbd_post_send_empty(info);
1368         }
1369 }
1370
1371 /* Implement idle connection timer [MS-SMBD] 3.1.6.2 */
1372 static void idle_connection_timer(struct work_struct *work)
1373 {
1374         struct smbd_connection *info = container_of(
1375                                         work, struct smbd_connection,
1376                                         idle_timer_work.work);
1377
1378         if (info->keep_alive_requested != KEEP_ALIVE_NONE) {
1379                 log_keep_alive(ERR,
1380                         "error status info->keep_alive_requested=%d\n",
1381                         info->keep_alive_requested);
1382                 smbd_disconnect_rdma_connection(info);
1383                 return;
1384         }
1385
1386         log_keep_alive(INFO, "about to send an empty idle message\n");
1387         smbd_post_send_empty(info);
1388
1389         /* Setup the next idle timeout work */
1390         queue_delayed_work(info->workqueue, &info->idle_timer_work,
1391                         info->keep_alive_interval*HZ);
1392 }
1393
1394 /*
1395  * Destroy the transport and related RDMA and memory resources
1396  * Need to go through all the pending counters and make sure on one is using
1397  * the transport while it is destroyed
1398  */
1399 void smbd_destroy(struct TCP_Server_Info *server)
1400 {
1401         struct smbd_connection *info = server->smbd_conn;
1402         struct smbd_response *response;
1403         unsigned long flags;
1404
1405         if (!info) {
1406                 log_rdma_event(INFO, "rdma session already destroyed\n");
1407                 return;
1408         }
1409
1410         log_rdma_event(INFO, "destroying rdma session\n");
1411         if (info->transport_status != SMBD_DISCONNECTED) {
1412                 rdma_disconnect(server->smbd_conn->id);
1413                 log_rdma_event(INFO, "wait for transport being disconnected\n");
1414                 wait_event_interruptible(
1415                         info->disconn_wait,
1416                         info->transport_status == SMBD_DISCONNECTED);
1417         }
1418
1419         log_rdma_event(INFO, "destroying qp\n");
1420         ib_drain_qp(info->id->qp);
1421         rdma_destroy_qp(info->id);
1422
1423         log_rdma_event(INFO, "cancelling idle timer\n");
1424         cancel_delayed_work_sync(&info->idle_timer_work);
1425         log_rdma_event(INFO, "cancelling send immediate work\n");
1426         cancel_delayed_work_sync(&info->send_immediate_work);
1427
1428         log_rdma_event(INFO, "wait for all send posted to IB to finish\n");
1429         wait_event(info->wait_send_pending,
1430                 atomic_read(&info->send_pending) == 0);
1431         wait_event(info->wait_send_payload_pending,
1432                 atomic_read(&info->send_payload_pending) == 0);
1433
1434         /* It's not posssible for upper layer to get to reassembly */
1435         log_rdma_event(INFO, "drain the reassembly queue\n");
1436         do {
1437                 spin_lock_irqsave(&info->reassembly_queue_lock, flags);
1438                 response = _get_first_reassembly(info);
1439                 if (response) {
1440                         list_del(&response->list);
1441                         spin_unlock_irqrestore(
1442                                 &info->reassembly_queue_lock, flags);
1443                         put_receive_buffer(info, response);
1444                 } else
1445                         spin_unlock_irqrestore(
1446                                 &info->reassembly_queue_lock, flags);
1447         } while (response);
1448         info->reassembly_data_length = 0;
1449
1450         log_rdma_event(INFO, "free receive buffers\n");
1451         wait_event(info->wait_receive_queues,
1452                 info->count_receive_queue + info->count_empty_packet_queue
1453                         == info->receive_credit_max);
1454         destroy_receive_buffers(info);
1455
1456         /*
1457          * For performance reasons, memory registration and deregistration
1458          * are not locked by srv_mutex. It is possible some processes are
1459          * blocked on transport srv_mutex while holding memory registration.
1460          * Release the transport srv_mutex to allow them to hit the failure
1461          * path when sending data, and then release memory registartions.
1462          */
1463         log_rdma_event(INFO, "freeing mr list\n");
1464         wake_up_interruptible_all(&info->wait_mr);
1465         while (atomic_read(&info->mr_used_count)) {
1466                 mutex_unlock(&server->srv_mutex);
1467                 msleep(1000);
1468                 mutex_lock(&server->srv_mutex);
1469         }
1470         destroy_mr_list(info);
1471
1472         ib_free_cq(info->send_cq);
1473         ib_free_cq(info->recv_cq);
1474         ib_dealloc_pd(info->pd);
1475         rdma_destroy_id(info->id);
1476
1477         /* free mempools */
1478         mempool_destroy(info->request_mempool);
1479         kmem_cache_destroy(info->request_cache);
1480
1481         mempool_destroy(info->response_mempool);
1482         kmem_cache_destroy(info->response_cache);
1483
1484         info->transport_status = SMBD_DESTROYED;
1485
1486         destroy_workqueue(info->workqueue);
1487         kfree(info);
1488 }
1489
1490 /*
1491  * Reconnect this SMBD connection, called from upper layer
1492  * return value: 0 on success, or actual error code
1493  */
1494 int smbd_reconnect(struct TCP_Server_Info *server)
1495 {
1496         log_rdma_event(INFO, "reconnecting rdma session\n");
1497
1498         if (!server->smbd_conn) {
1499                 log_rdma_event(INFO, "rdma session already destroyed\n");
1500                 goto create_conn;
1501         }
1502
1503         /*
1504          * This is possible if transport is disconnected and we haven't received
1505          * notification from RDMA, but upper layer has detected timeout
1506          */
1507         if (server->smbd_conn->transport_status == SMBD_CONNECTED) {
1508                 log_rdma_event(INFO, "disconnecting transport\n");
1509                 smbd_destroy(server);
1510         }
1511
1512 create_conn:
1513         log_rdma_event(INFO, "creating rdma session\n");
1514         server->smbd_conn = smbd_get_connection(
1515                 server, (struct sockaddr *) &server->dstaddr);
1516         log_rdma_event(INFO, "created rdma session info=%p\n",
1517                 server->smbd_conn);
1518
1519         return server->smbd_conn ? 0 : -ENOENT;
1520 }
1521
1522 static void destroy_caches_and_workqueue(struct smbd_connection *info)
1523 {
1524         destroy_receive_buffers(info);
1525         destroy_workqueue(info->workqueue);
1526         mempool_destroy(info->response_mempool);
1527         kmem_cache_destroy(info->response_cache);
1528         mempool_destroy(info->request_mempool);
1529         kmem_cache_destroy(info->request_cache);
1530 }
1531
1532 #define MAX_NAME_LEN    80
1533 static int allocate_caches_and_workqueue(struct smbd_connection *info)
1534 {
1535         char name[MAX_NAME_LEN];
1536         int rc;
1537
1538         scnprintf(name, MAX_NAME_LEN, "smbd_request_%p", info);
1539         info->request_cache =
1540                 kmem_cache_create(
1541                         name,
1542                         sizeof(struct smbd_request) +
1543                                 sizeof(struct smbd_data_transfer),
1544                         0, SLAB_HWCACHE_ALIGN, NULL);
1545         if (!info->request_cache)
1546                 return -ENOMEM;
1547
1548         info->request_mempool =
1549                 mempool_create(info->send_credit_target, mempool_alloc_slab,
1550                         mempool_free_slab, info->request_cache);
1551         if (!info->request_mempool)
1552                 goto out1;
1553
1554         scnprintf(name, MAX_NAME_LEN, "smbd_response_%p", info);
1555         info->response_cache =
1556                 kmem_cache_create(
1557                         name,
1558                         sizeof(struct smbd_response) +
1559                                 info->max_receive_size,
1560                         0, SLAB_HWCACHE_ALIGN, NULL);
1561         if (!info->response_cache)
1562                 goto out2;
1563
1564         info->response_mempool =
1565                 mempool_create(info->receive_credit_max, mempool_alloc_slab,
1566                        mempool_free_slab, info->response_cache);
1567         if (!info->response_mempool)
1568                 goto out3;
1569
1570         scnprintf(name, MAX_NAME_LEN, "smbd_%p", info);
1571         info->workqueue = create_workqueue(name);
1572         if (!info->workqueue)
1573                 goto out4;
1574
1575         rc = allocate_receive_buffers(info, info->receive_credit_max);
1576         if (rc) {
1577                 log_rdma_event(ERR, "failed to allocate receive buffers\n");
1578                 goto out5;
1579         }
1580
1581         return 0;
1582
1583 out5:
1584         destroy_workqueue(info->workqueue);
1585 out4:
1586         mempool_destroy(info->response_mempool);
1587 out3:
1588         kmem_cache_destroy(info->response_cache);
1589 out2:
1590         mempool_destroy(info->request_mempool);
1591 out1:
1592         kmem_cache_destroy(info->request_cache);
1593         return -ENOMEM;
1594 }
1595
1596 /* Create a SMBD connection, called by upper layer */
1597 static struct smbd_connection *_smbd_get_connection(
1598         struct TCP_Server_Info *server, struct sockaddr *dstaddr, int port)
1599 {
1600         int rc;
1601         struct smbd_connection *info;
1602         struct rdma_conn_param conn_param;
1603         struct ib_qp_init_attr qp_attr;
1604         struct sockaddr_in *addr_in = (struct sockaddr_in *) dstaddr;
1605         struct ib_port_immutable port_immutable;
1606         u32 ird_ord_hdr[2];
1607
1608         info = kzalloc(sizeof(struct smbd_connection), GFP_KERNEL);
1609         if (!info)
1610                 return NULL;
1611
1612         info->transport_status = SMBD_CONNECTING;
1613         rc = smbd_ia_open(info, dstaddr, port);
1614         if (rc) {
1615                 log_rdma_event(INFO, "smbd_ia_open rc=%d\n", rc);
1616                 goto create_id_failed;
1617         }
1618
1619         if (smbd_send_credit_target > info->id->device->attrs.max_cqe ||
1620             smbd_send_credit_target > info->id->device->attrs.max_qp_wr) {
1621                 log_rdma_event(ERR,
1622                         "consider lowering send_credit_target = %d. "
1623                         "Possible CQE overrun, device "
1624                         "reporting max_cpe %d max_qp_wr %d\n",
1625                         smbd_send_credit_target,
1626                         info->id->device->attrs.max_cqe,
1627                         info->id->device->attrs.max_qp_wr);
1628                 goto config_failed;
1629         }
1630
1631         if (smbd_receive_credit_max > info->id->device->attrs.max_cqe ||
1632             smbd_receive_credit_max > info->id->device->attrs.max_qp_wr) {
1633                 log_rdma_event(ERR,
1634                         "consider lowering receive_credit_max = %d. "
1635                         "Possible CQE overrun, device "
1636                         "reporting max_cpe %d max_qp_wr %d\n",
1637                         smbd_receive_credit_max,
1638                         info->id->device->attrs.max_cqe,
1639                         info->id->device->attrs.max_qp_wr);
1640                 goto config_failed;
1641         }
1642
1643         info->receive_credit_max = smbd_receive_credit_max;
1644         info->send_credit_target = smbd_send_credit_target;
1645         info->max_send_size = smbd_max_send_size;
1646         info->max_fragmented_recv_size = smbd_max_fragmented_recv_size;
1647         info->max_receive_size = smbd_max_receive_size;
1648         info->keep_alive_interval = smbd_keep_alive_interval;
1649
1650         if (info->id->device->attrs.max_send_sge < SMBDIRECT_MAX_SGE) {
1651                 log_rdma_event(ERR,
1652                         "warning: device max_send_sge = %d too small\n",
1653                         info->id->device->attrs.max_send_sge);
1654                 log_rdma_event(ERR, "Queue Pair creation may fail\n");
1655         }
1656         if (info->id->device->attrs.max_recv_sge < SMBDIRECT_MAX_SGE) {
1657                 log_rdma_event(ERR,
1658                         "warning: device max_recv_sge = %d too small\n",
1659                         info->id->device->attrs.max_recv_sge);
1660                 log_rdma_event(ERR, "Queue Pair creation may fail\n");
1661         }
1662
1663         info->send_cq = NULL;
1664         info->recv_cq = NULL;
1665         info->send_cq = ib_alloc_cq(info->id->device, info,
1666                         info->send_credit_target, 0, IB_POLL_SOFTIRQ);
1667         if (IS_ERR(info->send_cq)) {
1668                 info->send_cq = NULL;
1669                 goto alloc_cq_failed;
1670         }
1671
1672         info->recv_cq = ib_alloc_cq(info->id->device, info,
1673                         info->receive_credit_max, 0, IB_POLL_SOFTIRQ);
1674         if (IS_ERR(info->recv_cq)) {
1675                 info->recv_cq = NULL;
1676                 goto alloc_cq_failed;
1677         }
1678
1679         memset(&qp_attr, 0, sizeof(qp_attr));
1680         qp_attr.event_handler = smbd_qp_async_error_upcall;
1681         qp_attr.qp_context = info;
1682         qp_attr.cap.max_send_wr = info->send_credit_target;
1683         qp_attr.cap.max_recv_wr = info->receive_credit_max;
1684         qp_attr.cap.max_send_sge = SMBDIRECT_MAX_SGE;
1685         qp_attr.cap.max_recv_sge = SMBDIRECT_MAX_SGE;
1686         qp_attr.cap.max_inline_data = 0;
1687         qp_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
1688         qp_attr.qp_type = IB_QPT_RC;
1689         qp_attr.send_cq = info->send_cq;
1690         qp_attr.recv_cq = info->recv_cq;
1691         qp_attr.port_num = ~0;
1692
1693         rc = rdma_create_qp(info->id, info->pd, &qp_attr);
1694         if (rc) {
1695                 log_rdma_event(ERR, "rdma_create_qp failed %i\n", rc);
1696                 goto create_qp_failed;
1697         }
1698
1699         memset(&conn_param, 0, sizeof(conn_param));
1700         conn_param.initiator_depth = 0;
1701
1702         conn_param.responder_resources =
1703                 info->id->device->attrs.max_qp_rd_atom
1704                         < SMBD_CM_RESPONDER_RESOURCES ?
1705                 info->id->device->attrs.max_qp_rd_atom :
1706                 SMBD_CM_RESPONDER_RESOURCES;
1707         info->responder_resources = conn_param.responder_resources;
1708         log_rdma_mr(INFO, "responder_resources=%d\n",
1709                 info->responder_resources);
1710
1711         /* Need to send IRD/ORD in private data for iWARP */
1712         info->id->device->ops.get_port_immutable(
1713                 info->id->device, info->id->port_num, &port_immutable);
1714         if (port_immutable.core_cap_flags & RDMA_CORE_PORT_IWARP) {
1715                 ird_ord_hdr[0] = info->responder_resources;
1716                 ird_ord_hdr[1] = 1;
1717                 conn_param.private_data = ird_ord_hdr;
1718                 conn_param.private_data_len = sizeof(ird_ord_hdr);
1719         } else {
1720                 conn_param.private_data = NULL;
1721                 conn_param.private_data_len = 0;
1722         }
1723
1724         conn_param.retry_count = SMBD_CM_RETRY;
1725         conn_param.rnr_retry_count = SMBD_CM_RNR_RETRY;
1726         conn_param.flow_control = 0;
1727
1728         log_rdma_event(INFO, "connecting to IP %pI4 port %d\n",
1729                 &addr_in->sin_addr, port);
1730
1731         init_waitqueue_head(&info->conn_wait);
1732         init_waitqueue_head(&info->disconn_wait);
1733         init_waitqueue_head(&info->wait_reassembly_queue);
1734         rc = rdma_connect(info->id, &conn_param);
1735         if (rc) {
1736                 log_rdma_event(ERR, "rdma_connect() failed with %i\n", rc);
1737                 goto rdma_connect_failed;
1738         }
1739
1740         wait_event_interruptible(
1741                 info->conn_wait, info->transport_status != SMBD_CONNECTING);
1742
1743         if (info->transport_status != SMBD_CONNECTED) {
1744                 log_rdma_event(ERR, "rdma_connect failed port=%d\n", port);
1745                 goto rdma_connect_failed;
1746         }
1747
1748         log_rdma_event(INFO, "rdma_connect connected\n");
1749
1750         rc = allocate_caches_and_workqueue(info);
1751         if (rc) {
1752                 log_rdma_event(ERR, "cache allocation failed\n");
1753                 goto allocate_cache_failed;
1754         }
1755
1756         init_waitqueue_head(&info->wait_send_queue);
1757         INIT_DELAYED_WORK(&info->idle_timer_work, idle_connection_timer);
1758         INIT_DELAYED_WORK(&info->send_immediate_work, send_immediate_work);
1759         queue_delayed_work(info->workqueue, &info->idle_timer_work,
1760                 info->keep_alive_interval*HZ);
1761
1762         init_waitqueue_head(&info->wait_send_pending);
1763         atomic_set(&info->send_pending, 0);
1764
1765         init_waitqueue_head(&info->wait_send_payload_pending);
1766         atomic_set(&info->send_payload_pending, 0);
1767
1768         INIT_WORK(&info->disconnect_work, smbd_disconnect_rdma_work);
1769         INIT_WORK(&info->recv_done_work, smbd_recv_done_work);
1770         INIT_WORK(&info->post_send_credits_work, smbd_post_send_credits);
1771         info->new_credits_offered = 0;
1772         spin_lock_init(&info->lock_new_credits_offered);
1773
1774         rc = smbd_negotiate(info);
1775         if (rc) {
1776                 log_rdma_event(ERR, "smbd_negotiate rc=%d\n", rc);
1777                 goto negotiation_failed;
1778         }
1779
1780         rc = allocate_mr_list(info);
1781         if (rc) {
1782                 log_rdma_mr(ERR, "memory registration allocation failed\n");
1783                 goto allocate_mr_failed;
1784         }
1785
1786         return info;
1787
1788 allocate_mr_failed:
1789         /* At this point, need to a full transport shutdown */
1790         smbd_destroy(server);
1791         return NULL;
1792
1793 negotiation_failed:
1794         cancel_delayed_work_sync(&info->idle_timer_work);
1795         destroy_caches_and_workqueue(info);
1796         info->transport_status = SMBD_NEGOTIATE_FAILED;
1797         init_waitqueue_head(&info->conn_wait);
1798         rdma_disconnect(info->id);
1799         wait_event(info->conn_wait,
1800                 info->transport_status == SMBD_DISCONNECTED);
1801
1802 allocate_cache_failed:
1803 rdma_connect_failed:
1804         rdma_destroy_qp(info->id);
1805
1806 create_qp_failed:
1807 alloc_cq_failed:
1808         if (info->send_cq)
1809                 ib_free_cq(info->send_cq);
1810         if (info->recv_cq)
1811                 ib_free_cq(info->recv_cq);
1812
1813 config_failed:
1814         ib_dealloc_pd(info->pd);
1815         rdma_destroy_id(info->id);
1816
1817 create_id_failed:
1818         kfree(info);
1819         return NULL;
1820 }
1821
1822 struct smbd_connection *smbd_get_connection(
1823         struct TCP_Server_Info *server, struct sockaddr *dstaddr)
1824 {
1825         struct smbd_connection *ret;
1826         int port = SMBD_PORT;
1827
1828 try_again:
1829         ret = _smbd_get_connection(server, dstaddr, port);
1830
1831         /* Try SMB_PORT if SMBD_PORT doesn't work */
1832         if (!ret && port == SMBD_PORT) {
1833                 port = SMB_PORT;
1834                 goto try_again;
1835         }
1836         return ret;
1837 }
1838
1839 /*
1840  * Receive data from receive reassembly queue
1841  * All the incoming data packets are placed in reassembly queue
1842  * buf: the buffer to read data into
1843  * size: the length of data to read
1844  * return value: actual data read
1845  * Note: this implementation copies the data from reassebmly queue to receive
1846  * buffers used by upper layer. This is not the optimal code path. A better way
1847  * to do it is to not have upper layer allocate its receive buffers but rather
1848  * borrow the buffer from reassembly queue, and return it after data is
1849  * consumed. But this will require more changes to upper layer code, and also
1850  * need to consider packet boundaries while they still being reassembled.
1851  */
1852 static int smbd_recv_buf(struct smbd_connection *info, char *buf,
1853                 unsigned int size)
1854 {
1855         struct smbd_response *response;
1856         struct smbd_data_transfer *data_transfer;
1857         int to_copy, to_read, data_read, offset;
1858         u32 data_length, remaining_data_length, data_offset;
1859         int rc;
1860
1861 again:
1862         /*
1863          * No need to hold the reassembly queue lock all the time as we are
1864          * the only one reading from the front of the queue. The transport
1865          * may add more entries to the back of the queue at the same time
1866          */
1867         log_read(INFO, "size=%d info->reassembly_data_length=%d\n", size,
1868                 info->reassembly_data_length);
1869         if (info->reassembly_data_length >= size) {
1870                 int queue_length;
1871                 int queue_removed = 0;
1872
1873                 /*
1874                  * Need to make sure reassembly_data_length is read before
1875                  * reading reassembly_queue_length and calling
1876                  * _get_first_reassembly. This call is lock free
1877                  * as we never read at the end of the queue which are being
1878                  * updated in SOFTIRQ as more data is received
1879                  */
1880                 virt_rmb();
1881                 queue_length = info->reassembly_queue_length;
1882                 data_read = 0;
1883                 to_read = size;
1884                 offset = info->first_entry_offset;
1885                 while (data_read < size) {
1886                         response = _get_first_reassembly(info);
1887                         data_transfer = smbd_response_payload(response);
1888                         data_length = le32_to_cpu(data_transfer->data_length);
1889                         remaining_data_length =
1890                                 le32_to_cpu(
1891                                         data_transfer->remaining_data_length);
1892                         data_offset = le32_to_cpu(data_transfer->data_offset);
1893
1894                         /*
1895                          * The upper layer expects RFC1002 length at the
1896                          * beginning of the payload. Return it to indicate
1897                          * the total length of the packet. This minimize the
1898                          * change to upper layer packet processing logic. This
1899                          * will be eventually remove when an intermediate
1900                          * transport layer is added
1901                          */
1902                         if (response->first_segment && size == 4) {
1903                                 unsigned int rfc1002_len =
1904                                         data_length + remaining_data_length;
1905                                 *((__be32 *)buf) = cpu_to_be32(rfc1002_len);
1906                                 data_read = 4;
1907                                 response->first_segment = false;
1908                                 log_read(INFO, "returning rfc1002 length %d\n",
1909                                         rfc1002_len);
1910                                 goto read_rfc1002_done;
1911                         }
1912
1913                         to_copy = min_t(int, data_length - offset, to_read);
1914                         memcpy(
1915                                 buf + data_read,
1916                                 (char *)data_transfer + data_offset + offset,
1917                                 to_copy);
1918
1919                         /* move on to the next buffer? */
1920                         if (to_copy == data_length - offset) {
1921                                 queue_length--;
1922                                 /*
1923                                  * No need to lock if we are not at the
1924                                  * end of the queue
1925                                  */
1926                                 if (queue_length)
1927                                         list_del(&response->list);
1928                                 else {
1929                                         spin_lock_irq(
1930                                                 &info->reassembly_queue_lock);
1931                                         list_del(&response->list);
1932                                         spin_unlock_irq(
1933                                                 &info->reassembly_queue_lock);
1934                                 }
1935                                 queue_removed++;
1936                                 info->count_reassembly_queue--;
1937                                 info->count_dequeue_reassembly_queue++;
1938                                 put_receive_buffer(info, response);
1939                                 offset = 0;
1940                                 log_read(INFO, "put_receive_buffer offset=0\n");
1941                         } else
1942                                 offset += to_copy;
1943
1944                         to_read -= to_copy;
1945                         data_read += to_copy;
1946
1947                         log_read(INFO, "_get_first_reassembly memcpy %d bytes "
1948                                 "data_transfer_length-offset=%d after that "
1949                                 "to_read=%d data_read=%d offset=%d\n",
1950                                 to_copy, data_length - offset,
1951                                 to_read, data_read, offset);
1952                 }
1953
1954                 spin_lock_irq(&info->reassembly_queue_lock);
1955                 info->reassembly_data_length -= data_read;
1956                 info->reassembly_queue_length -= queue_removed;
1957                 spin_unlock_irq(&info->reassembly_queue_lock);
1958
1959                 info->first_entry_offset = offset;
1960                 log_read(INFO, "returning to thread data_read=%d "
1961                         "reassembly_data_length=%d first_entry_offset=%d\n",
1962                         data_read, info->reassembly_data_length,
1963                         info->first_entry_offset);
1964 read_rfc1002_done:
1965                 return data_read;
1966         }
1967
1968         log_read(INFO, "wait_event on more data\n");
1969         rc = wait_event_interruptible(
1970                 info->wait_reassembly_queue,
1971                 info->reassembly_data_length >= size ||
1972                         info->transport_status != SMBD_CONNECTED);
1973         /* Don't return any data if interrupted */
1974         if (rc)
1975                 return rc;
1976
1977         if (info->transport_status != SMBD_CONNECTED) {
1978                 log_read(ERR, "disconnected\n");
1979                 return 0;
1980         }
1981
1982         goto again;
1983 }
1984
1985 /*
1986  * Receive a page from receive reassembly queue
1987  * page: the page to read data into
1988  * to_read: the length of data to read
1989  * return value: actual data read
1990  */
1991 static int smbd_recv_page(struct smbd_connection *info,
1992                 struct page *page, unsigned int page_offset,
1993                 unsigned int to_read)
1994 {
1995         int ret;
1996         char *to_address;
1997         void *page_address;
1998
1999         /* make sure we have the page ready for read */
2000         ret = wait_event_interruptible(
2001                 info->wait_reassembly_queue,
2002                 info->reassembly_data_length >= to_read ||
2003                         info->transport_status != SMBD_CONNECTED);
2004         if (ret)
2005                 return ret;
2006
2007         /* now we can read from reassembly queue and not sleep */
2008         page_address = kmap_atomic(page);
2009         to_address = (char *) page_address + page_offset;
2010
2011         log_read(INFO, "reading from page=%p address=%p to_read=%d\n",
2012                 page, to_address, to_read);
2013
2014         ret = smbd_recv_buf(info, to_address, to_read);
2015         kunmap_atomic(page_address);
2016
2017         return ret;
2018 }
2019
2020 /*
2021  * Receive data from transport
2022  * msg: a msghdr point to the buffer, can be ITER_KVEC or ITER_BVEC
2023  * return: total bytes read, or 0. SMB Direct will not do partial read.
2024  */
2025 int smbd_recv(struct smbd_connection *info, struct msghdr *msg)
2026 {
2027         char *buf;
2028         struct page *page;
2029         unsigned int to_read, page_offset;
2030         int rc;
2031
2032         if (iov_iter_rw(&msg->msg_iter) == WRITE) {
2033                 /* It's a bug in upper layer to get there */
2034                 cifs_dbg(VFS, "CIFS: invalid msg iter dir %u\n",
2035                          iov_iter_rw(&msg->msg_iter));
2036                 rc = -EINVAL;
2037                 goto out;
2038         }
2039
2040         switch (iov_iter_type(&msg->msg_iter)) {
2041         case ITER_KVEC:
2042                 buf = msg->msg_iter.kvec->iov_base;
2043                 to_read = msg->msg_iter.kvec->iov_len;
2044                 rc = smbd_recv_buf(info, buf, to_read);
2045                 break;
2046
2047         case ITER_BVEC:
2048                 page = msg->msg_iter.bvec->bv_page;
2049                 page_offset = msg->msg_iter.bvec->bv_offset;
2050                 to_read = msg->msg_iter.bvec->bv_len;
2051                 rc = smbd_recv_page(info, page, page_offset, to_read);
2052                 break;
2053
2054         default:
2055                 /* It's a bug in upper layer to get there */
2056                 cifs_dbg(VFS, "CIFS: invalid msg type %d\n",
2057                          iov_iter_type(&msg->msg_iter));
2058                 rc = -EINVAL;
2059         }
2060
2061 out:
2062         /* SMBDirect will read it all or nothing */
2063         if (rc > 0)
2064                 msg->msg_iter.count = 0;
2065         return rc;
2066 }
2067
2068 /*
2069  * Send data to transport
2070  * Each rqst is transported as a SMBDirect payload
2071  * rqst: the data to write
2072  * return value: 0 if successfully write, otherwise error code
2073  */
2074 int smbd_send(struct TCP_Server_Info *server, struct smb_rqst *rqst)
2075 {
2076         struct smbd_connection *info = server->smbd_conn;
2077         struct kvec vec;
2078         int nvecs;
2079         int size;
2080         unsigned int buflen, remaining_data_length;
2081         int start, i, j;
2082         int max_iov_size =
2083                 info->max_send_size - sizeof(struct smbd_data_transfer);
2084         struct kvec *iov;
2085         int rc;
2086
2087         if (info->transport_status != SMBD_CONNECTED) {
2088                 rc = -ENODEV;
2089                 goto done;
2090         }
2091
2092         /*
2093          * Skip the RFC1002 length defined in MS-SMB2 section 2.1
2094          * It is used only for TCP transport in the iov[0]
2095          * In future we may want to add a transport layer under protocol
2096          * layer so this will only be issued to TCP transport
2097          */
2098
2099         if (rqst->rq_iov[0].iov_len != 4) {
2100                 log_write(ERR, "expected the pdu length in 1st iov, but got %zu\n", rqst->rq_iov[0].iov_len);
2101                 return -EINVAL;
2102         }
2103
2104         /*
2105          * Add in the page array if there is one. The caller needs to set
2106          * rq_tailsz to PAGE_SIZE when the buffer has multiple pages and
2107          * ends at page boundary
2108          */
2109         buflen = smb_rqst_len(server, rqst);
2110
2111         if (buflen + sizeof(struct smbd_data_transfer) >
2112                 info->max_fragmented_send_size) {
2113                 log_write(ERR, "payload size %d > max size %d\n",
2114                         buflen, info->max_fragmented_send_size);
2115                 rc = -EINVAL;
2116                 goto done;
2117         }
2118
2119         iov = &rqst->rq_iov[1];
2120
2121         cifs_dbg(FYI, "Sending smb (RDMA): smb_len=%u\n", buflen);
2122         for (i = 0; i < rqst->rq_nvec-1; i++)
2123                 dump_smb(iov[i].iov_base, iov[i].iov_len);
2124
2125         remaining_data_length = buflen;
2126
2127         log_write(INFO, "rqst->rq_nvec=%d rqst->rq_npages=%d rq_pagesz=%d "
2128                 "rq_tailsz=%d buflen=%d\n",
2129                 rqst->rq_nvec, rqst->rq_npages, rqst->rq_pagesz,
2130                 rqst->rq_tailsz, buflen);
2131
2132         start = i = iov[0].iov_len ? 0 : 1;
2133         buflen = 0;
2134         while (true) {
2135                 buflen += iov[i].iov_len;
2136                 if (buflen > max_iov_size) {
2137                         if (i > start) {
2138                                 remaining_data_length -=
2139                                         (buflen-iov[i].iov_len);
2140                                 log_write(INFO, "sending iov[] from start=%d "
2141                                         "i=%d nvecs=%d "
2142                                         "remaining_data_length=%d\n",
2143                                         start, i, i-start,
2144                                         remaining_data_length);
2145                                 rc = smbd_post_send_data(
2146                                         info, &iov[start], i-start,
2147                                         remaining_data_length);
2148                                 if (rc)
2149                                         goto done;
2150                         } else {
2151                                 /* iov[start] is too big, break it */
2152                                 nvecs = (buflen+max_iov_size-1)/max_iov_size;
2153                                 log_write(INFO, "iov[%d] iov_base=%p buflen=%d"
2154                                         " break to %d vectors\n",
2155                                         start, iov[start].iov_base,
2156                                         buflen, nvecs);
2157                                 for (j = 0; j < nvecs; j++) {
2158                                         vec.iov_base =
2159                                                 (char *)iov[start].iov_base +
2160                                                 j*max_iov_size;
2161                                         vec.iov_len = max_iov_size;
2162                                         if (j == nvecs-1)
2163                                                 vec.iov_len =
2164                                                         buflen -
2165                                                         max_iov_size*(nvecs-1);
2166                                         remaining_data_length -= vec.iov_len;
2167                                         log_write(INFO,
2168                                                 "sending vec j=%d iov_base=%p"
2169                                                 " iov_len=%zu "
2170                                                 "remaining_data_length=%d\n",
2171                                                 j, vec.iov_base, vec.iov_len,
2172                                                 remaining_data_length);
2173                                         rc = smbd_post_send_data(
2174                                                 info, &vec, 1,
2175                                                 remaining_data_length);
2176                                         if (rc)
2177                                                 goto done;
2178                                 }
2179                                 i++;
2180                                 if (i == rqst->rq_nvec-1)
2181                                         break;
2182                         }
2183                         start = i;
2184                         buflen = 0;
2185                 } else {
2186                         i++;
2187                         if (i == rqst->rq_nvec-1) {
2188                                 /* send out all remaining vecs */
2189                                 remaining_data_length -= buflen;
2190                                 log_write(INFO,
2191                                         "sending iov[] from start=%d i=%d "
2192                                         "nvecs=%d remaining_data_length=%d\n",
2193                                         start, i, i-start,
2194                                         remaining_data_length);
2195                                 rc = smbd_post_send_data(info, &iov[start],
2196                                         i-start, remaining_data_length);
2197                                 if (rc)
2198                                         goto done;
2199                                 break;
2200                         }
2201                 }
2202                 log_write(INFO, "looping i=%d buflen=%d\n", i, buflen);
2203         }
2204
2205         /* now sending pages if there are any */
2206         for (i = 0; i < rqst->rq_npages; i++) {
2207                 unsigned int offset;
2208
2209                 rqst_page_get_length(rqst, i, &buflen, &offset);
2210                 nvecs = (buflen + max_iov_size - 1) / max_iov_size;
2211                 log_write(INFO, "sending pages buflen=%d nvecs=%d\n",
2212                         buflen, nvecs);
2213                 for (j = 0; j < nvecs; j++) {
2214                         size = max_iov_size;
2215                         if (j == nvecs-1)
2216                                 size = buflen - j*max_iov_size;
2217                         remaining_data_length -= size;
2218                         log_write(INFO, "sending pages i=%d offset=%d size=%d"
2219                                 " remaining_data_length=%d\n",
2220                                 i, j*max_iov_size+offset, size,
2221                                 remaining_data_length);
2222                         rc = smbd_post_send_page(
2223                                 info, rqst->rq_pages[i],
2224                                 j*max_iov_size + offset,
2225                                 size, remaining_data_length);
2226                         if (rc)
2227                                 goto done;
2228                 }
2229         }
2230
2231 done:
2232         /*
2233          * As an optimization, we don't wait for individual I/O to finish
2234          * before sending the next one.
2235          * Send them all and wait for pending send count to get to 0
2236          * that means all the I/Os have been out and we are good to return
2237          */
2238
2239         wait_event(info->wait_send_payload_pending,
2240                 atomic_read(&info->send_payload_pending) == 0);
2241
2242         return rc;
2243 }
2244
2245 static void register_mr_done(struct ib_cq *cq, struct ib_wc *wc)
2246 {
2247         struct smbd_mr *mr;
2248         struct ib_cqe *cqe;
2249
2250         if (wc->status) {
2251                 log_rdma_mr(ERR, "status=%d\n", wc->status);
2252                 cqe = wc->wr_cqe;
2253                 mr = container_of(cqe, struct smbd_mr, cqe);
2254                 smbd_disconnect_rdma_connection(mr->conn);
2255         }
2256 }
2257
2258 /*
2259  * The work queue function that recovers MRs
2260  * We need to call ib_dereg_mr() and ib_alloc_mr() before this MR can be used
2261  * again. Both calls are slow, so finish them in a workqueue. This will not
2262  * block I/O path.
2263  * There is one workqueue that recovers MRs, there is no need to lock as the
2264  * I/O requests calling smbd_register_mr will never update the links in the
2265  * mr_list.
2266  */
2267 static void smbd_mr_recovery_work(struct work_struct *work)
2268 {
2269         struct smbd_connection *info =
2270                 container_of(work, struct smbd_connection, mr_recovery_work);
2271         struct smbd_mr *smbdirect_mr;
2272         int rc;
2273
2274         list_for_each_entry(smbdirect_mr, &info->mr_list, list) {
2275                 if (smbdirect_mr->state == MR_INVALIDATED)
2276                         ib_dma_unmap_sg(
2277                                 info->id->device, smbdirect_mr->sgl,
2278                                 smbdirect_mr->sgl_count,
2279                                 smbdirect_mr->dir);
2280                 else if (smbdirect_mr->state == MR_ERROR) {
2281
2282                         /* recover this MR entry */
2283                         rc = ib_dereg_mr(smbdirect_mr->mr);
2284                         if (rc) {
2285                                 log_rdma_mr(ERR,
2286                                         "ib_dereg_mr failed rc=%x\n",
2287                                         rc);
2288                                 smbd_disconnect_rdma_connection(info);
2289                                 continue;
2290                         }
2291
2292                         smbdirect_mr->mr = ib_alloc_mr(
2293                                 info->pd, info->mr_type,
2294                                 info->max_frmr_depth);
2295                         if (IS_ERR(smbdirect_mr->mr)) {
2296                                 log_rdma_mr(ERR,
2297                                         "ib_alloc_mr failed mr_type=%x "
2298                                         "max_frmr_depth=%x\n",
2299                                         info->mr_type,
2300                                         info->max_frmr_depth);
2301                                 smbd_disconnect_rdma_connection(info);
2302                                 continue;
2303                         }
2304                 } else
2305                         /* This MR is being used, don't recover it */
2306                         continue;
2307
2308                 smbdirect_mr->state = MR_READY;
2309
2310                 /* smbdirect_mr->state is updated by this function
2311                  * and is read and updated by I/O issuing CPUs trying
2312                  * to get a MR, the call to atomic_inc_return
2313                  * implicates a memory barrier and guarantees this
2314                  * value is updated before waking up any calls to
2315                  * get_mr() from the I/O issuing CPUs
2316                  */
2317                 if (atomic_inc_return(&info->mr_ready_count) == 1)
2318                         wake_up_interruptible(&info->wait_mr);
2319         }
2320 }
2321
2322 static void destroy_mr_list(struct smbd_connection *info)
2323 {
2324         struct smbd_mr *mr, *tmp;
2325
2326         cancel_work_sync(&info->mr_recovery_work);
2327         list_for_each_entry_safe(mr, tmp, &info->mr_list, list) {
2328                 if (mr->state == MR_INVALIDATED)
2329                         ib_dma_unmap_sg(info->id->device, mr->sgl,
2330                                 mr->sgl_count, mr->dir);
2331                 ib_dereg_mr(mr->mr);
2332                 kfree(mr->sgl);
2333                 kfree(mr);
2334         }
2335 }
2336
2337 /*
2338  * Allocate MRs used for RDMA read/write
2339  * The number of MRs will not exceed hardware capability in responder_resources
2340  * All MRs are kept in mr_list. The MR can be recovered after it's used
2341  * Recovery is done in smbd_mr_recovery_work. The content of list entry changes
2342  * as MRs are used and recovered for I/O, but the list links will not change
2343  */
2344 static int allocate_mr_list(struct smbd_connection *info)
2345 {
2346         int i;
2347         struct smbd_mr *smbdirect_mr, *tmp;
2348
2349         INIT_LIST_HEAD(&info->mr_list);
2350         init_waitqueue_head(&info->wait_mr);
2351         spin_lock_init(&info->mr_list_lock);
2352         atomic_set(&info->mr_ready_count, 0);
2353         atomic_set(&info->mr_used_count, 0);
2354         init_waitqueue_head(&info->wait_for_mr_cleanup);
2355         /* Allocate more MRs (2x) than hardware responder_resources */
2356         for (i = 0; i < info->responder_resources * 2; i++) {
2357                 smbdirect_mr = kzalloc(sizeof(*smbdirect_mr), GFP_KERNEL);
2358                 if (!smbdirect_mr)
2359                         goto out;
2360                 smbdirect_mr->mr = ib_alloc_mr(info->pd, info->mr_type,
2361                                         info->max_frmr_depth);
2362                 if (IS_ERR(smbdirect_mr->mr)) {
2363                         log_rdma_mr(ERR, "ib_alloc_mr failed mr_type=%x "
2364                                 "max_frmr_depth=%x\n",
2365                                 info->mr_type, info->max_frmr_depth);
2366                         goto out;
2367                 }
2368                 smbdirect_mr->sgl = kcalloc(
2369                                         info->max_frmr_depth,
2370                                         sizeof(struct scatterlist),
2371                                         GFP_KERNEL);
2372                 if (!smbdirect_mr->sgl) {
2373                         log_rdma_mr(ERR, "failed to allocate sgl\n");
2374                         ib_dereg_mr(smbdirect_mr->mr);
2375                         goto out;
2376                 }
2377                 smbdirect_mr->state = MR_READY;
2378                 smbdirect_mr->conn = info;
2379
2380                 list_add_tail(&smbdirect_mr->list, &info->mr_list);
2381                 atomic_inc(&info->mr_ready_count);
2382         }
2383         INIT_WORK(&info->mr_recovery_work, smbd_mr_recovery_work);
2384         return 0;
2385
2386 out:
2387         kfree(smbdirect_mr);
2388
2389         list_for_each_entry_safe(smbdirect_mr, tmp, &info->mr_list, list) {
2390                 ib_dereg_mr(smbdirect_mr->mr);
2391                 kfree(smbdirect_mr->sgl);
2392                 kfree(smbdirect_mr);
2393         }
2394         return -ENOMEM;
2395 }
2396
2397 /*
2398  * Get a MR from mr_list. This function waits until there is at least one
2399  * MR available in the list. It may access the list while the
2400  * smbd_mr_recovery_work is recovering the MR list. This doesn't need a lock
2401  * as they never modify the same places. However, there may be several CPUs
2402  * issueing I/O trying to get MR at the same time, mr_list_lock is used to
2403  * protect this situation.
2404  */
2405 static struct smbd_mr *get_mr(struct smbd_connection *info)
2406 {
2407         struct smbd_mr *ret;
2408         int rc;
2409 again:
2410         rc = wait_event_interruptible(info->wait_mr,
2411                 atomic_read(&info->mr_ready_count) ||
2412                 info->transport_status != SMBD_CONNECTED);
2413         if (rc) {
2414                 log_rdma_mr(ERR, "wait_event_interruptible rc=%x\n", rc);
2415                 return NULL;
2416         }
2417
2418         if (info->transport_status != SMBD_CONNECTED) {
2419                 log_rdma_mr(ERR, "info->transport_status=%x\n",
2420                         info->transport_status);
2421                 return NULL;
2422         }
2423
2424         spin_lock(&info->mr_list_lock);
2425         list_for_each_entry(ret, &info->mr_list, list) {
2426                 if (ret->state == MR_READY) {
2427                         ret->state = MR_REGISTERED;
2428                         spin_unlock(&info->mr_list_lock);
2429                         atomic_dec(&info->mr_ready_count);
2430                         atomic_inc(&info->mr_used_count);
2431                         return ret;
2432                 }
2433         }
2434
2435         spin_unlock(&info->mr_list_lock);
2436         /*
2437          * It is possible that we could fail to get MR because other processes may
2438          * try to acquire a MR at the same time. If this is the case, retry it.
2439          */
2440         goto again;
2441 }
2442
2443 /*
2444  * Register memory for RDMA read/write
2445  * pages[]: the list of pages to register memory with
2446  * num_pages: the number of pages to register
2447  * tailsz: if non-zero, the bytes to register in the last page
2448  * writing: true if this is a RDMA write (SMB read), false for RDMA read
2449  * need_invalidate: true if this MR needs to be locally invalidated after I/O
2450  * return value: the MR registered, NULL if failed.
2451  */
2452 struct smbd_mr *smbd_register_mr(
2453         struct smbd_connection *info, struct page *pages[], int num_pages,
2454         int offset, int tailsz, bool writing, bool need_invalidate)
2455 {
2456         struct smbd_mr *smbdirect_mr;
2457         int rc, i;
2458         enum dma_data_direction dir;
2459         struct ib_reg_wr *reg_wr;
2460
2461         if (num_pages > info->max_frmr_depth) {
2462                 log_rdma_mr(ERR, "num_pages=%d max_frmr_depth=%d\n",
2463                         num_pages, info->max_frmr_depth);
2464                 return NULL;
2465         }
2466
2467         smbdirect_mr = get_mr(info);
2468         if (!smbdirect_mr) {
2469                 log_rdma_mr(ERR, "get_mr returning NULL\n");
2470                 return NULL;
2471         }
2472         smbdirect_mr->need_invalidate = need_invalidate;
2473         smbdirect_mr->sgl_count = num_pages;
2474         sg_init_table(smbdirect_mr->sgl, num_pages);
2475
2476         log_rdma_mr(INFO, "num_pages=0x%x offset=0x%x tailsz=0x%x\n",
2477                         num_pages, offset, tailsz);
2478
2479         if (num_pages == 1) {
2480                 sg_set_page(&smbdirect_mr->sgl[0], pages[0], tailsz, offset);
2481                 goto skip_multiple_pages;
2482         }
2483
2484         /* We have at least two pages to register */
2485         sg_set_page(
2486                 &smbdirect_mr->sgl[0], pages[0], PAGE_SIZE - offset, offset);
2487         i = 1;
2488         while (i < num_pages - 1) {
2489                 sg_set_page(&smbdirect_mr->sgl[i], pages[i], PAGE_SIZE, 0);
2490                 i++;
2491         }
2492         sg_set_page(&smbdirect_mr->sgl[i], pages[i],
2493                 tailsz ? tailsz : PAGE_SIZE, 0);
2494
2495 skip_multiple_pages:
2496         dir = writing ? DMA_FROM_DEVICE : DMA_TO_DEVICE;
2497         smbdirect_mr->dir = dir;
2498         rc = ib_dma_map_sg(info->id->device, smbdirect_mr->sgl, num_pages, dir);
2499         if (!rc) {
2500                 log_rdma_mr(ERR, "ib_dma_map_sg num_pages=%x dir=%x rc=%x\n",
2501                         num_pages, dir, rc);
2502                 goto dma_map_error;
2503         }
2504
2505         rc = ib_map_mr_sg(smbdirect_mr->mr, smbdirect_mr->sgl, num_pages,
2506                 NULL, PAGE_SIZE);
2507         if (rc != num_pages) {
2508                 log_rdma_mr(ERR,
2509                         "ib_map_mr_sg failed rc = %d num_pages = %x\n",
2510                         rc, num_pages);
2511                 goto map_mr_error;
2512         }
2513
2514         ib_update_fast_reg_key(smbdirect_mr->mr,
2515                 ib_inc_rkey(smbdirect_mr->mr->rkey));
2516         reg_wr = &smbdirect_mr->wr;
2517         reg_wr->wr.opcode = IB_WR_REG_MR;
2518         smbdirect_mr->cqe.done = register_mr_done;
2519         reg_wr->wr.wr_cqe = &smbdirect_mr->cqe;
2520         reg_wr->wr.num_sge = 0;
2521         reg_wr->wr.send_flags = IB_SEND_SIGNALED;
2522         reg_wr->mr = smbdirect_mr->mr;
2523         reg_wr->key = smbdirect_mr->mr->rkey;
2524         reg_wr->access = writing ?
2525                         IB_ACCESS_REMOTE_WRITE | IB_ACCESS_LOCAL_WRITE :
2526                         IB_ACCESS_REMOTE_READ;
2527
2528         /*
2529          * There is no need for waiting for complemtion on ib_post_send
2530          * on IB_WR_REG_MR. Hardware enforces a barrier and order of execution
2531          * on the next ib_post_send when we actaully send I/O to remote peer
2532          */
2533         rc = ib_post_send(info->id->qp, &reg_wr->wr, NULL);
2534         if (!rc)
2535                 return smbdirect_mr;
2536
2537         log_rdma_mr(ERR, "ib_post_send failed rc=%x reg_wr->key=%x\n",
2538                 rc, reg_wr->key);
2539
2540         /* If all failed, attempt to recover this MR by setting it MR_ERROR*/
2541 map_mr_error:
2542         ib_dma_unmap_sg(info->id->device, smbdirect_mr->sgl,
2543                 smbdirect_mr->sgl_count, smbdirect_mr->dir);
2544
2545 dma_map_error:
2546         smbdirect_mr->state = MR_ERROR;
2547         if (atomic_dec_and_test(&info->mr_used_count))
2548                 wake_up(&info->wait_for_mr_cleanup);
2549
2550         smbd_disconnect_rdma_connection(info);
2551
2552         return NULL;
2553 }
2554
2555 static void local_inv_done(struct ib_cq *cq, struct ib_wc *wc)
2556 {
2557         struct smbd_mr *smbdirect_mr;
2558         struct ib_cqe *cqe;
2559
2560         cqe = wc->wr_cqe;
2561         smbdirect_mr = container_of(cqe, struct smbd_mr, cqe);
2562         smbdirect_mr->state = MR_INVALIDATED;
2563         if (wc->status != IB_WC_SUCCESS) {
2564                 log_rdma_mr(ERR, "invalidate failed status=%x\n", wc->status);
2565                 smbdirect_mr->state = MR_ERROR;
2566         }
2567         complete(&smbdirect_mr->invalidate_done);
2568 }
2569
2570 /*
2571  * Deregister a MR after I/O is done
2572  * This function may wait if remote invalidation is not used
2573  * and we have to locally invalidate the buffer to prevent data is being
2574  * modified by remote peer after upper layer consumes it
2575  */
2576 int smbd_deregister_mr(struct smbd_mr *smbdirect_mr)
2577 {
2578         struct ib_send_wr *wr;
2579         struct smbd_connection *info = smbdirect_mr->conn;
2580         int rc = 0;
2581
2582         if (smbdirect_mr->need_invalidate) {
2583                 /* Need to finish local invalidation before returning */
2584                 wr = &smbdirect_mr->inv_wr;
2585                 wr->opcode = IB_WR_LOCAL_INV;
2586                 smbdirect_mr->cqe.done = local_inv_done;
2587                 wr->wr_cqe = &smbdirect_mr->cqe;
2588                 wr->num_sge = 0;
2589                 wr->ex.invalidate_rkey = smbdirect_mr->mr->rkey;
2590                 wr->send_flags = IB_SEND_SIGNALED;
2591
2592                 init_completion(&smbdirect_mr->invalidate_done);
2593                 rc = ib_post_send(info->id->qp, wr, NULL);
2594                 if (rc) {
2595                         log_rdma_mr(ERR, "ib_post_send failed rc=%x\n", rc);
2596                         smbd_disconnect_rdma_connection(info);
2597                         goto done;
2598                 }
2599                 wait_for_completion(&smbdirect_mr->invalidate_done);
2600                 smbdirect_mr->need_invalidate = false;
2601         } else
2602                 /*
2603                  * For remote invalidation, just set it to MR_INVALIDATED
2604                  * and defer to mr_recovery_work to recover the MR for next use
2605                  */
2606                 smbdirect_mr->state = MR_INVALIDATED;
2607
2608         /*
2609          * Schedule the work to do MR recovery for future I/Os
2610          * MR recovery is slow and we don't want it to block the current I/O
2611          */
2612         queue_work(info->workqueue, &info->mr_recovery_work);
2613
2614 done:
2615         if (atomic_dec_and_test(&info->mr_used_count))
2616                 wake_up(&info->wait_for_mr_cleanup);
2617
2618         return rc;
2619 }