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 -EAGAIN;
 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_TO_DEVICE);
 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                 rc = -EAGAIN;
 983         } else
 984                 /* Reset timer for idle connection after packet is sent */
 985                 mod_delayed_work(info->workqueue, &info->idle_timer_work,
 986                         info->keep_alive_interval*HZ);
 987
 988         return rc;
 989 }
 990
 991 static int smbd_post_send_sgl(struct smbd_connection *info,
 992         struct scatterlist *sgl, int data_length, int remaining_data_length)
 993 {
 994         int num_sgs;
 995         int i, rc;
 996         struct smbd_request *request;
 997         struct scatterlist *sg;
 998
 999         rc = smbd_create_header(
1000                 info, data_length, remaining_data_length, &request);
1001         if (rc)
1002                 return rc;
1003
1004         num_sgs = sgl ? sg_nents(sgl) : 0;
1005         for_each_sg(sgl, sg, num_sgs, i) {
1006                 request->sge[i+1].addr =
1007                         ib_dma_map_page(info->id->device, sg_page(sg),
1008                                sg->offset, sg->length, DMA_TO_DEVICE);
1009                 if (ib_dma_mapping_error(
1010                                 info->id->device, request->sge[i+1].addr)) {
1011                         rc = -EIO;
1012                         request->sge[i+1].addr = 0;
1013                         goto dma_mapping_failure;
1014                 }
1015                 request->sge[i+1].length = sg->length;
1016                 request->sge[i+1].lkey = info->pd->local_dma_lkey;
1017                 request->num_sge++;
1018         }
1019
1020         rc = smbd_post_send(info, request, data_length);
1021         if (!rc)
1022                 return 0;
1023
1024 dma_mapping_failure:
1025         for (i = 1; i < request->num_sge; i++)
1026                 if (request->sge[i].addr)
1027                         ib_dma_unmap_single(info->id->device,
1028                                             request->sge[i].addr,
1029                                             request->sge[i].length,
1030                                             DMA_TO_DEVICE);
1031         smbd_destroy_header(info, request);
1032         return rc;
1033 }
1034
1035 /*
1036  * Send a page
1037  * page: the page to send
1038  * offset: offset in the page to send
1039  * size: length in the page to send
1040  * remaining_data_length: remaining data to send in this payload
1041  */
1042 static int smbd_post_send_page(struct smbd_connection *info, struct page *page,
1043                 unsigned long offset, size_t size, int remaining_data_length)
1044 {
1045         struct scatterlist sgl;
1046
1047         sg_init_table(&sgl, 1);
1048         sg_set_page(&sgl, page, size, offset);
1049
1050         return smbd_post_send_sgl(info, &sgl, size, remaining_data_length);
1051 }
1052
1053 /*
1054  * Send an empty message
1055  * Empty message is used to extend credits to peer to for keep live
1056  * while there is no upper layer payload to send at the time
1057  */
1058 static int smbd_post_send_empty(struct smbd_connection *info)
1059 {
1060         info->count_send_empty++;
1061         return smbd_post_send_sgl(info, NULL, 0, 0);
1062 }
1063
1064 /*
1065  * Send a data buffer
1066  * iov: the iov array describing the data buffers
1067  * n_vec: number of iov array
1068  * remaining_data_length: remaining data to send following this packet
1069  * in segmented SMBD packet
1070  */
1071 static int smbd_post_send_data(
1072         struct smbd_connection *info, struct kvec *iov, int n_vec,
1073         int remaining_data_length)
1074 {
1075         int i;
1076         u32 data_length = 0;
1077         struct scatterlist sgl[SMBDIRECT_MAX_SGE];
1078
1079         if (n_vec > SMBDIRECT_MAX_SGE) {
1080                 cifs_dbg(VFS, "Can't fit data to SGL, n_vec=%d\n", n_vec);
1081                 return -ENOMEM;
1082         }
1083
1084         sg_init_table(sgl, n_vec);
1085         for (i = 0; i < n_vec; i++) {
1086                 data_length += iov[i].iov_len;
1087                 sg_set_buf(&sgl[i], iov[i].iov_base, iov[i].iov_len);
1088         }
1089
1090         return smbd_post_send_sgl(info, sgl, data_length, remaining_data_length);
1091 }
1092
1093 /*
1094  * Post a receive request to the transport
1095  * The remote peer can only send data when a receive request is posted
1096  * The interaction is controlled by send/receive credit system
1097  */
1098 static int smbd_post_recv(
1099                 struct smbd_connection *info, struct smbd_response *response)
1100 {
1101         struct ib_recv_wr recv_wr;
1102         int rc = -EIO;
1103
1104         response->sge.addr = ib_dma_map_single(
1105                                 info->id->device, response->packet,
1106                                 info->max_receive_size, DMA_FROM_DEVICE);
1107         if (ib_dma_mapping_error(info->id->device, response->sge.addr))
1108                 return rc;
1109
1110         response->sge.length = info->max_receive_size;
1111         response->sge.lkey = info->pd->local_dma_lkey;
1112
1113         response->cqe.done = recv_done;
1114
1115         recv_wr.wr_cqe = &response->cqe;
1116         recv_wr.next = NULL;
1117         recv_wr.sg_list = &response->sge;
1118         recv_wr.num_sge = 1;
1119
1120         rc = ib_post_recv(info->id->qp, &recv_wr, NULL);
1121         if (rc) {
1122                 ib_dma_unmap_single(info->id->device, response->sge.addr,
1123                                     response->sge.length, DMA_FROM_DEVICE);
1124                 smbd_disconnect_rdma_connection(info);
1125                 log_rdma_recv(ERR, "ib_post_recv failed rc=%d\n", rc);
1126         }
1127
1128         return rc;
1129 }
1130
1131 /* Perform SMBD negotiate according to [MS-SMBD] 3.1.5.2 */
1132 static int smbd_negotiate(struct smbd_connection *info)
1133 {
1134         int rc;
1135         struct smbd_response *response = get_receive_buffer(info);
1136
1137         response->type = SMBD_NEGOTIATE_RESP;
1138         rc = smbd_post_recv(info, response);
1139         log_rdma_event(INFO,
1140                 "smbd_post_recv rc=%d iov.addr=%llx iov.length=%x "
1141                 "iov.lkey=%x\n",
1142                 rc, response->sge.addr,
1143                 response->sge.length, response->sge.lkey);
1144         if (rc)
1145                 return rc;
1146
1147         init_completion(&info->negotiate_completion);
1148         info->negotiate_done = false;
1149         rc = smbd_post_send_negotiate_req(info);
1150         if (rc)
1151                 return rc;
1152
1153         rc = wait_for_completion_interruptible_timeout(
1154                 &info->negotiate_completion, SMBD_NEGOTIATE_TIMEOUT * HZ);
1155         log_rdma_event(INFO, "wait_for_completion_timeout rc=%d\n", rc);
1156
1157         if (info->negotiate_done)
1158                 return 0;
1159
1160         if (rc == 0)
1161                 rc = -ETIMEDOUT;
1162         else if (rc == -ERESTARTSYS)
1163                 rc = -EINTR;
1164         else
1165                 rc = -ENOTCONN;
1166
1167         return rc;
1168 }
1169
1170 static void put_empty_packet(
1171                 struct smbd_connection *info, struct smbd_response *response)
1172 {
1173         spin_lock(&info->empty_packet_queue_lock);
1174         list_add_tail(&response->list, &info->empty_packet_queue);
1175         info->count_empty_packet_queue++;
1176         spin_unlock(&info->empty_packet_queue_lock);
1177
1178         queue_work(info->workqueue, &info->post_send_credits_work);
1179 }
1180
1181 /*
1182  * Implement Connection.FragmentReassemblyBuffer defined in [MS-SMBD] 3.1.1.1
1183  * This is a queue for reassembling upper layer payload and present to upper
1184  * layer. All the inncoming payload go to the reassembly queue, regardless of
1185  * if reassembly is required. The uuper layer code reads from the queue for all
1186  * incoming payloads.
1187  * Put a received packet to the reassembly queue
1188  * response: the packet received
1189  * data_length: the size of payload in this packet
1190  */
1191 static void enqueue_reassembly(
1192         struct smbd_connection *info,
1193         struct smbd_response *response,
1194         int data_length)
1195 {
1196         spin_lock(&info->reassembly_queue_lock);
1197         list_add_tail(&response->list, &info->reassembly_queue);
1198         info->reassembly_queue_length++;
1199         /*
1200          * Make sure reassembly_data_length is updated after list and
1201          * reassembly_queue_length are updated. On the dequeue side
1202          * reassembly_data_length is checked without a lock to determine
1203          * if reassembly_queue_length and list is up to date
1204          */
1205         virt_wmb();
1206         info->reassembly_data_length += data_length;
1207         spin_unlock(&info->reassembly_queue_lock);
1208         info->count_reassembly_queue++;
1209         info->count_enqueue_reassembly_queue++;
1210 }
1211
1212 /*
1213  * Get the first entry at the front of reassembly queue
1214  * Caller is responsible for locking
1215  * return value: the first entry if any, NULL if queue is empty
1216  */
1217 static struct smbd_response *_get_first_reassembly(struct smbd_connection *info)
1218 {
1219         struct smbd_response *ret = NULL;
1220
1221         if (!list_empty(&info->reassembly_queue)) {
1222                 ret = list_first_entry(
1223                         &info->reassembly_queue,
1224                         struct smbd_response, list);
1225         }
1226         return ret;
1227 }
1228
1229 static struct smbd_response *get_empty_queue_buffer(
1230                 struct smbd_connection *info)
1231 {
1232         struct smbd_response *ret = NULL;
1233         unsigned long flags;
1234
1235         spin_lock_irqsave(&info->empty_packet_queue_lock, flags);
1236         if (!list_empty(&info->empty_packet_queue)) {
1237                 ret = list_first_entry(
1238                         &info->empty_packet_queue,
1239                         struct smbd_response, list);
1240                 list_del(&ret->list);
1241                 info->count_empty_packet_queue--;
1242         }
1243         spin_unlock_irqrestore(&info->empty_packet_queue_lock, flags);
1244
1245         return ret;
1246 }
1247
1248 /*
1249  * Get a receive buffer
1250  * For each remote send, we need to post a receive. The receive buffers are
1251  * pre-allocated in advance.
1252  * return value: the receive buffer, NULL if none is available
1253  */
1254 static struct smbd_response *get_receive_buffer(struct smbd_connection *info)
1255 {
1256         struct smbd_response *ret = NULL;
1257         unsigned long flags;
1258
1259         spin_lock_irqsave(&info->receive_queue_lock, flags);
1260         if (!list_empty(&info->receive_queue)) {
1261                 ret = list_first_entry(
1262                         &info->receive_queue,
1263                         struct smbd_response, list);
1264                 list_del(&ret->list);
1265                 info->count_receive_queue--;
1266                 info->count_get_receive_buffer++;
1267         }
1268         spin_unlock_irqrestore(&info->receive_queue_lock, flags);
1269
1270         return ret;
1271 }
1272
1273 /*
1274  * Return a receive buffer
1275  * Upon returning of a receive buffer, we can post new receive and extend
1276  * more receive credits to remote peer. This is done immediately after a
1277  * receive buffer is returned.
1278  */
1279 static void put_receive_buffer(
1280         struct smbd_connection *info, struct smbd_response *response)
1281 {
1282         unsigned long flags;
1283
1284         ib_dma_unmap_single(info->id->device, response->sge.addr,
1285                 response->sge.length, DMA_FROM_DEVICE);
1286
1287         spin_lock_irqsave(&info->receive_queue_lock, flags);
1288         list_add_tail(&response->list, &info->receive_queue);
1289         info->count_receive_queue++;
1290         info->count_put_receive_buffer++;
1291         spin_unlock_irqrestore(&info->receive_queue_lock, flags);
1292
1293         queue_work(info->workqueue, &info->post_send_credits_work);
1294 }
1295
1296 /* Preallocate all receive buffer on transport establishment */
1297 static int allocate_receive_buffers(struct smbd_connection *info, int num_buf)
1298 {
1299         int i;
1300         struct smbd_response *response;
1301
1302         INIT_LIST_HEAD(&info->reassembly_queue);
1303         spin_lock_init(&info->reassembly_queue_lock);
1304         info->reassembly_data_length = 0;
1305         info->reassembly_queue_length = 0;
1306
1307         INIT_LIST_HEAD(&info->receive_queue);
1308         spin_lock_init(&info->receive_queue_lock);
1309         info->count_receive_queue = 0;
1310
1311         INIT_LIST_HEAD(&info->empty_packet_queue);
1312         spin_lock_init(&info->empty_packet_queue_lock);
1313         info->count_empty_packet_queue = 0;
1314
1315         init_waitqueue_head(&info->wait_receive_queues);
1316
1317         for (i = 0; i < num_buf; i++) {
1318                 response = mempool_alloc(info->response_mempool, GFP_KERNEL);
1319                 if (!response)
1320                         goto allocate_failed;
1321
1322                 response->info = info;
1323                 list_add_tail(&response->list, &info->receive_queue);
1324                 info->count_receive_queue++;
1325         }
1326
1327         return 0;
1328
1329 allocate_failed:
1330         while (!list_empty(&info->receive_queue)) {
1331                 response = list_first_entry(
1332                                 &info->receive_queue,
1333                                 struct smbd_response, list);
1334                 list_del(&response->list);
1335                 info->count_receive_queue--;
1336
1337                 mempool_free(response, info->response_mempool);
1338         }
1339         return -ENOMEM;
1340 }
1341
1342 static void destroy_receive_buffers(struct smbd_connection *info)
1343 {
1344         struct smbd_response *response;
1345
1346         while ((response = get_receive_buffer(info)))
1347                 mempool_free(response, info->response_mempool);
1348
1349         while ((response = get_empty_queue_buffer(info)))
1350                 mempool_free(response, info->response_mempool);
1351 }
1352
1353 /*
1354  * Check and send an immediate or keep alive packet
1355  * The condition to send those packets are defined in [MS-SMBD] 3.1.1.1
1356  * Connection.KeepaliveRequested and Connection.SendImmediate
1357  * The idea is to extend credits to server as soon as it becomes available
1358  */
1359 static void send_immediate_work(struct work_struct *work)
1360 {
1361         struct smbd_connection *info = container_of(
1362                                         work, struct smbd_connection,
1363                                         send_immediate_work.work);
1364
1365         if (info->keep_alive_requested == KEEP_ALIVE_PENDING ||
1366             info->send_immediate) {
1367                 log_keep_alive(INFO, "send an empty message\n");
1368                 smbd_post_send_empty(info);
1369         }
1370 }
1371
1372 /* Implement idle connection timer [MS-SMBD] 3.1.6.2 */
1373 static void idle_connection_timer(struct work_struct *work)
1374 {
1375         struct smbd_connection *info = container_of(
1376                                         work, struct smbd_connection,
1377                                         idle_timer_work.work);
1378
1379         if (info->keep_alive_requested != KEEP_ALIVE_NONE) {
1380                 log_keep_alive(ERR,
1381                         "error status info->keep_alive_requested=%d\n",
1382                         info->keep_alive_requested);
1383                 smbd_disconnect_rdma_connection(info);
1384                 return;
1385         }
1386
1387         log_keep_alive(INFO, "about to send an empty idle message\n");
1388         smbd_post_send_empty(info);
1389
1390         /* Setup the next idle timeout work */
1391         queue_delayed_work(info->workqueue, &info->idle_timer_work,
1392                         info->keep_alive_interval*HZ);
1393 }
1394
1395 /*
1396  * Destroy the transport and related RDMA and memory resources
1397  * Need to go through all the pending counters and make sure on one is using
1398  * the transport while it is destroyed
1399  */
1400 void smbd_destroy(struct TCP_Server_Info *server)
1401 {
1402         struct smbd_connection *info = server->smbd_conn;
1403         struct smbd_response *response;
1404         unsigned long flags;
1405
1406         if (!info) {
1407                 log_rdma_event(INFO, "rdma session already destroyed\n");
1408                 return;
1409         }
1410
1411         log_rdma_event(INFO, "destroying rdma session\n");
1412         if (info->transport_status != SMBD_DISCONNECTED) {
1413                 rdma_disconnect(server->smbd_conn->id);
1414                 log_rdma_event(INFO, "wait for transport being disconnected\n");
1415                 wait_event_interruptible(
1416                         info->disconn_wait,
1417                         info->transport_status == SMBD_DISCONNECTED);
1418         }
1419
1420         log_rdma_event(INFO, "destroying qp\n");
1421         ib_drain_qp(info->id->qp);
1422         rdma_destroy_qp(info->id);
1423
1424         log_rdma_event(INFO, "cancelling idle timer\n");
1425         cancel_delayed_work_sync(&info->idle_timer_work);
1426         log_rdma_event(INFO, "cancelling send immediate work\n");
1427         cancel_delayed_work_sync(&info->send_immediate_work);
1428
1429         log_rdma_event(INFO, "wait for all send posted to IB to finish\n");
1430         wait_event(info->wait_send_pending,
1431                 atomic_read(&info->send_pending) == 0);
1432         wait_event(info->wait_send_payload_pending,
1433                 atomic_read(&info->send_payload_pending) == 0);
1434
1435         /* It's not posssible for upper layer to get to reassembly */
1436         log_rdma_event(INFO, "drain the reassembly queue\n");
1437         do {
1438                 spin_lock_irqsave(&info->reassembly_queue_lock, flags);
1439                 response = _get_first_reassembly(info);
1440                 if (response) {
1441                         list_del(&response->list);
1442                         spin_unlock_irqrestore(
1443                                 &info->reassembly_queue_lock, flags);
1444                         put_receive_buffer(info, response);
1445                 } else
1446                         spin_unlock_irqrestore(
1447                                 &info->reassembly_queue_lock, flags);
1448         } while (response);
1449         info->reassembly_data_length = 0;
1450
1451         log_rdma_event(INFO, "free receive buffers\n");
1452         wait_event(info->wait_receive_queues,
1453                 info->count_receive_queue + info->count_empty_packet_queue
1454                         == info->receive_credit_max);
1455         destroy_receive_buffers(info);
1456
1457         /*
1458          * For performance reasons, memory registration and deregistration
1459          * are not locked by srv_mutex. It is possible some processes are
1460          * blocked on transport srv_mutex while holding memory registration.
1461          * Release the transport srv_mutex to allow them to hit the failure
1462          * path when sending data, and then release memory registartions.
1463          */
1464         log_rdma_event(INFO, "freeing mr list\n");
1465         wake_up_interruptible_all(&info->wait_mr);
1466         while (atomic_read(&info->mr_used_count)) {
1467                 mutex_unlock(&server->srv_mutex);
1468                 msleep(1000);
1469                 mutex_lock(&server->srv_mutex);
1470         }
1471         destroy_mr_list(info);
1472
1473         ib_free_cq(info->send_cq);
1474         ib_free_cq(info->recv_cq);
1475         ib_dealloc_pd(info->pd);
1476         rdma_destroy_id(info->id);
1477
1478         /* free mempools */
1479         mempool_destroy(info->request_mempool);
1480         kmem_cache_destroy(info->request_cache);
1481
1482         mempool_destroy(info->response_mempool);
1483         kmem_cache_destroy(info->response_cache);
1484
1485         info->transport_status = SMBD_DESTROYED;
1486
1487         destroy_workqueue(info->workqueue);
1488         kfree(info);
1489 }
1490
1491 /*
1492  * Reconnect this SMBD connection, called from upper layer
1493  * return value: 0 on success, or actual error code
1494  */
1495 int smbd_reconnect(struct TCP_Server_Info *server)
1496 {
1497         log_rdma_event(INFO, "reconnecting rdma session\n");
1498
1499         if (!server->smbd_conn) {
1500                 log_rdma_event(INFO, "rdma session already destroyed\n");
1501                 goto create_conn;
1502         }
1503
1504         /*
1505          * This is possible if transport is disconnected and we haven't received
1506          * notification from RDMA, but upper layer has detected timeout
1507          */
1508         if (server->smbd_conn->transport_status == SMBD_CONNECTED) {
1509                 log_rdma_event(INFO, "disconnecting transport\n");
1510                 smbd_destroy(server);
1511         }
1512
1513 create_conn:
1514         log_rdma_event(INFO, "creating rdma session\n");
1515         server->smbd_conn = smbd_get_connection(
1516                 server, (struct sockaddr *) &server->dstaddr);
1517         log_rdma_event(INFO, "created rdma session info=%p\n",
1518                 server->smbd_conn);
1519
1520         return server->smbd_conn ? 0 : -ENOENT;
1521 }
1522
1523 static void destroy_caches_and_workqueue(struct smbd_connection *info)
1524 {
1525         destroy_receive_buffers(info);
1526         destroy_workqueue(info->workqueue);
1527         mempool_destroy(info->response_mempool);
1528         kmem_cache_destroy(info->response_cache);
1529         mempool_destroy(info->request_mempool);
1530         kmem_cache_destroy(info->request_cache);
1531 }
1532
1533 #define MAX_NAME_LEN    80
1534 static int allocate_caches_and_workqueue(struct smbd_connection *info)
1535 {
1536         char name[MAX_NAME_LEN];
1537         int rc;
1538
1539         scnprintf(name, MAX_NAME_LEN, "smbd_request_%p", info);
1540         info->request_cache =
1541                 kmem_cache_create(
1542                         name,
1543                         sizeof(struct smbd_request) +
1544                                 sizeof(struct smbd_data_transfer),
1545                         0, SLAB_HWCACHE_ALIGN, NULL);
1546         if (!info->request_cache)
1547                 return -ENOMEM;
1548
1549         info->request_mempool =
1550                 mempool_create(info->send_credit_target, mempool_alloc_slab,
1551                         mempool_free_slab, info->request_cache);
1552         if (!info->request_mempool)
1553                 goto out1;
1554
1555         scnprintf(name, MAX_NAME_LEN, "smbd_response_%p", info);
1556         info->response_cache =
1557                 kmem_cache_create(
1558                         name,
1559                         sizeof(struct smbd_response) +
1560                                 info->max_receive_size,
1561                         0, SLAB_HWCACHE_ALIGN, NULL);
1562         if (!info->response_cache)
1563                 goto out2;
1564
1565         info->response_mempool =
1566                 mempool_create(info->receive_credit_max, mempool_alloc_slab,
1567                        mempool_free_slab, info->response_cache);
1568         if (!info->response_mempool)
1569                 goto out3;
1570
1571         scnprintf(name, MAX_NAME_LEN, "smbd_%p", info);
1572         info->workqueue = create_workqueue(name);
1573         if (!info->workqueue)
1574                 goto out4;
1575
1576         rc = allocate_receive_buffers(info, info->receive_credit_max);
1577         if (rc) {
1578                 log_rdma_event(ERR, "failed to allocate receive buffers\n");
1579                 goto out5;
1580         }
1581
1582         return 0;
1583
1584 out5:
1585         destroy_workqueue(info->workqueue);
1586 out4:
1587         mempool_destroy(info->response_mempool);
1588 out3:
1589         kmem_cache_destroy(info->response_cache);
1590 out2:
1591         mempool_destroy(info->request_mempool);
1592 out1:
1593         kmem_cache_destroy(info->request_cache);
1594         return -ENOMEM;
1595 }
1596
1597 /* Create a SMBD connection, called by upper layer */
1598 static struct smbd_connection *_smbd_get_connection(
1599         struct TCP_Server_Info *server, struct sockaddr *dstaddr, int port)
1600 {
1601         int rc;
1602         struct smbd_connection *info;
1603         struct rdma_conn_param conn_param;
1604         struct ib_qp_init_attr qp_attr;
1605         struct sockaddr_in *addr_in = (struct sockaddr_in *) dstaddr;
1606         struct ib_port_immutable port_immutable;
1607         u32 ird_ord_hdr[2];
1608
1609         info = kzalloc(sizeof(struct smbd_connection), GFP_KERNEL);
1610         if (!info)
1611                 return NULL;
1612
1613         info->transport_status = SMBD_CONNECTING;
1614         rc = smbd_ia_open(info, dstaddr, port);
1615         if (rc) {
1616                 log_rdma_event(INFO, "smbd_ia_open rc=%d\n", rc);
1617                 goto create_id_failed;
1618         }
1619
1620         if (smbd_send_credit_target > info->id->device->attrs.max_cqe ||
1621             smbd_send_credit_target > info->id->device->attrs.max_qp_wr) {
1622                 log_rdma_event(ERR,
1623                         "consider lowering send_credit_target = %d. "
1624                         "Possible CQE overrun, device "
1625                         "reporting max_cpe %d max_qp_wr %d\n",
1626                         smbd_send_credit_target,
1627                         info->id->device->attrs.max_cqe,
1628                         info->id->device->attrs.max_qp_wr);
1629                 goto config_failed;
1630         }
1631
1632         if (smbd_receive_credit_max > info->id->device->attrs.max_cqe ||
1633             smbd_receive_credit_max > info->id->device->attrs.max_qp_wr) {
1634                 log_rdma_event(ERR,
1635                         "consider lowering receive_credit_max = %d. "
1636                         "Possible CQE overrun, device "
1637                         "reporting max_cpe %d max_qp_wr %d\n",
1638                         smbd_receive_credit_max,
1639                         info->id->device->attrs.max_cqe,
1640                         info->id->device->attrs.max_qp_wr);
1641                 goto config_failed;
1642         }
1643
1644         info->receive_credit_max = smbd_receive_credit_max;
1645         info->send_credit_target = smbd_send_credit_target;
1646         info->max_send_size = smbd_max_send_size;
1647         info->max_fragmented_recv_size = smbd_max_fragmented_recv_size;
1648         info->max_receive_size = smbd_max_receive_size;
1649         info->keep_alive_interval = smbd_keep_alive_interval;
1650
1651         if (info->id->device->attrs.max_send_sge < SMBDIRECT_MAX_SGE) {
1652                 log_rdma_event(ERR,
1653                         "warning: device max_send_sge = %d too small\n",
1654                         info->id->device->attrs.max_send_sge);
1655                 log_rdma_event(ERR, "Queue Pair creation may fail\n");
1656         }
1657         if (info->id->device->attrs.max_recv_sge < SMBDIRECT_MAX_SGE) {
1658                 log_rdma_event(ERR,
1659                         "warning: device max_recv_sge = %d too small\n",
1660                         info->id->device->attrs.max_recv_sge);
1661                 log_rdma_event(ERR, "Queue Pair creation may fail\n");
1662         }
1663
1664         info->send_cq = NULL;
1665         info->recv_cq = NULL;
1666         info->send_cq = ib_alloc_cq(info->id->device, info,
1667                         info->send_credit_target, 0, IB_POLL_SOFTIRQ);
1668         if (IS_ERR(info->send_cq)) {
1669                 info->send_cq = NULL;
1670                 goto alloc_cq_failed;
1671         }
1672
1673         info->recv_cq = ib_alloc_cq(info->id->device, info,
1674                         info->receive_credit_max, 0, IB_POLL_SOFTIRQ);
1675         if (IS_ERR(info->recv_cq)) {
1676                 info->recv_cq = NULL;
1677                 goto alloc_cq_failed;
1678         }
1679
1680         memset(&qp_attr, 0, sizeof(qp_attr));
1681         qp_attr.event_handler = smbd_qp_async_error_upcall;
1682         qp_attr.qp_context = info;
1683         qp_attr.cap.max_send_wr = info->send_credit_target;
1684         qp_attr.cap.max_recv_wr = info->receive_credit_max;
1685         qp_attr.cap.max_send_sge = SMBDIRECT_MAX_SGE;
1686         qp_attr.cap.max_recv_sge = SMBDIRECT_MAX_SGE;
1687         qp_attr.cap.max_inline_data = 0;
1688         qp_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
1689         qp_attr.qp_type = IB_QPT_RC;
1690         qp_attr.send_cq = info->send_cq;
1691         qp_attr.recv_cq = info->recv_cq;
1692         qp_attr.port_num = ~0;
1693
1694         rc = rdma_create_qp(info->id, info->pd, &qp_attr);
1695         if (rc) {
1696                 log_rdma_event(ERR, "rdma_create_qp failed %i\n", rc);
1697                 goto create_qp_failed;
1698         }
1699
1700         memset(&conn_param, 0, sizeof(conn_param));
1701         conn_param.initiator_depth = 0;
1702
1703         conn_param.responder_resources =
1704                 info->id->device->attrs.max_qp_rd_atom
1705                         < SMBD_CM_RESPONDER_RESOURCES ?
1706                 info->id->device->attrs.max_qp_rd_atom :
1707                 SMBD_CM_RESPONDER_RESOURCES;
1708         info->responder_resources = conn_param.responder_resources;
1709         log_rdma_mr(INFO, "responder_resources=%d\n",
1710                 info->responder_resources);
1711
1712         /* Need to send IRD/ORD in private data for iWARP */
1713         info->id->device->ops.get_port_immutable(
1714                 info->id->device, info->id->port_num, &port_immutable);
1715         if (port_immutable.core_cap_flags & RDMA_CORE_PORT_IWARP) {
1716                 ird_ord_hdr[0] = info->responder_resources;
1717                 ird_ord_hdr[1] = 1;
1718                 conn_param.private_data = ird_ord_hdr;
1719                 conn_param.private_data_len = sizeof(ird_ord_hdr);
1720         } else {
1721                 conn_param.private_data = NULL;
1722                 conn_param.private_data_len = 0;
1723         }
1724
1725         conn_param.retry_count = SMBD_CM_RETRY;
1726         conn_param.rnr_retry_count = SMBD_CM_RNR_RETRY;
1727         conn_param.flow_control = 0;
1728
1729         log_rdma_event(INFO, "connecting to IP %pI4 port %d\n",
1730                 &addr_in->sin_addr, port);
1731
1732         init_waitqueue_head(&info->conn_wait);
1733         init_waitqueue_head(&info->disconn_wait);
1734         init_waitqueue_head(&info->wait_reassembly_queue);
1735         rc = rdma_connect(info->id, &conn_param);
1736         if (rc) {
1737                 log_rdma_event(ERR, "rdma_connect() failed with %i\n", rc);
1738                 goto rdma_connect_failed;
1739         }
1740
1741         wait_event_interruptible(
1742                 info->conn_wait, info->transport_status != SMBD_CONNECTING);
1743
1744         if (info->transport_status != SMBD_CONNECTED) {
1745                 log_rdma_event(ERR, "rdma_connect failed port=%d\n", port);
1746                 goto rdma_connect_failed;
1747         }
1748
1749         log_rdma_event(INFO, "rdma_connect connected\n");
1750
1751         rc = allocate_caches_and_workqueue(info);
1752         if (rc) {
1753                 log_rdma_event(ERR, "cache allocation failed\n");
1754                 goto allocate_cache_failed;
1755         }
1756
1757         init_waitqueue_head(&info->wait_send_queue);
1758         INIT_DELAYED_WORK(&info->idle_timer_work, idle_connection_timer);
1759         INIT_DELAYED_WORK(&info->send_immediate_work, send_immediate_work);
1760         queue_delayed_work(info->workqueue, &info->idle_timer_work,
1761                 info->keep_alive_interval*HZ);
1762
1763         init_waitqueue_head(&info->wait_send_pending);
1764         atomic_set(&info->send_pending, 0);
1765
1766         init_waitqueue_head(&info->wait_send_payload_pending);
1767         atomic_set(&info->send_payload_pending, 0);
1768
1769         INIT_WORK(&info->disconnect_work, smbd_disconnect_rdma_work);
1770         INIT_WORK(&info->recv_done_work, smbd_recv_done_work);
1771         INIT_WORK(&info->post_send_credits_work, smbd_post_send_credits);
1772         info->new_credits_offered = 0;
1773         spin_lock_init(&info->lock_new_credits_offered);
1774
1775         rc = smbd_negotiate(info);
1776         if (rc) {
1777                 log_rdma_event(ERR, "smbd_negotiate rc=%d\n", rc);
1778                 goto negotiation_failed;
1779         }
1780
1781         rc = allocate_mr_list(info);
1782         if (rc) {
1783                 log_rdma_mr(ERR, "memory registration allocation failed\n");
1784                 goto allocate_mr_failed;
1785         }
1786
1787         return info;
1788
1789 allocate_mr_failed:
1790         /* At this point, need to a full transport shutdown */
1791         smbd_destroy(server);
1792         return NULL;
1793
1794 negotiation_failed:
1795         cancel_delayed_work_sync(&info->idle_timer_work);
1796         destroy_caches_and_workqueue(info);
1797         info->transport_status = SMBD_NEGOTIATE_FAILED;
1798         init_waitqueue_head(&info->conn_wait);
1799         rdma_disconnect(info->id);
1800         wait_event(info->conn_wait,
1801                 info->transport_status == SMBD_DISCONNECTED);
1802
1803 allocate_cache_failed:
1804 rdma_connect_failed:
1805         rdma_destroy_qp(info->id);
1806
1807 create_qp_failed:
1808 alloc_cq_failed:
1809         if (info->send_cq)
1810                 ib_free_cq(info->send_cq);
1811         if (info->recv_cq)
1812                 ib_free_cq(info->recv_cq);
1813
1814 config_failed:
1815         ib_dealloc_pd(info->pd);
1816         rdma_destroy_id(info->id);
1817
1818 create_id_failed:
1819         kfree(info);
1820         return NULL;
1821 }
1822
1823 struct smbd_connection *smbd_get_connection(
1824         struct TCP_Server_Info *server, struct sockaddr *dstaddr)
1825 {
1826         struct smbd_connection *ret;
1827         int port = SMBD_PORT;
1828
1829 try_again:
1830         ret = _smbd_get_connection(server, dstaddr, port);
1831
1832         /* Try SMB_PORT if SMBD_PORT doesn't work */
1833         if (!ret && port == SMBD_PORT) {
1834                 port = SMB_PORT;
1835                 goto try_again;
1836         }
1837         return ret;
1838 }
1839
1840 /*
1841  * Receive data from receive reassembly queue
1842  * All the incoming data packets are placed in reassembly queue
1843  * buf: the buffer to read data into
1844  * size: the length of data to read
1845  * return value: actual data read
1846  * Note: this implementation copies the data from reassebmly queue to receive
1847  * buffers used by upper layer. This is not the optimal code path. A better way
1848  * to do it is to not have upper layer allocate its receive buffers but rather
1849  * borrow the buffer from reassembly queue, and return it after data is
1850  * consumed. But this will require more changes to upper layer code, and also
1851  * need to consider packet boundaries while they still being reassembled.
1852  */
1853 static int smbd_recv_buf(struct smbd_connection *info, char *buf,
1854                 unsigned int size)
1855 {
1856         struct smbd_response *response;
1857         struct smbd_data_transfer *data_transfer;
1858         int to_copy, to_read, data_read, offset;
1859         u32 data_length, remaining_data_length, data_offset;
1860         int rc;
1861
1862 again:
1863         /*
1864          * No need to hold the reassembly queue lock all the time as we are
1865          * the only one reading from the front of the queue. The transport
1866          * may add more entries to the back of the queue at the same time
1867          */
1868         log_read(INFO, "size=%d info->reassembly_data_length=%d\n", size,
1869                 info->reassembly_data_length);
1870         if (info->reassembly_data_length >= size) {
1871                 int queue_length;
1872                 int queue_removed = 0;
1873
1874                 /*
1875                  * Need to make sure reassembly_data_length is read before
1876                  * reading reassembly_queue_length and calling
1877                  * _get_first_reassembly. This call is lock free
1878                  * as we never read at the end of the queue which are being
1879                  * updated in SOFTIRQ as more data is received
1880                  */
1881                 virt_rmb();
1882                 queue_length = info->reassembly_queue_length;
1883                 data_read = 0;
1884                 to_read = size;
1885                 offset = info->first_entry_offset;
1886                 while (data_read < size) {
1887                         response = _get_first_reassembly(info);
1888                         data_transfer = smbd_response_payload(response);
1889                         data_length = le32_to_cpu(data_transfer->data_length);
1890                         remaining_data_length =
1891                                 le32_to_cpu(
1892                                         data_transfer->remaining_data_length);
1893                         data_offset = le32_to_cpu(data_transfer->data_offset);
1894
1895                         /*
1896                          * The upper layer expects RFC1002 length at the
1897                          * beginning of the payload. Return it to indicate
1898                          * the total length of the packet. This minimize the
1899                          * change to upper layer packet processing logic. This
1900                          * will be eventually remove when an intermediate
1901                          * transport layer is added
1902                          */
1903                         if (response->first_segment && size == 4) {
1904                                 unsigned int rfc1002_len =
1905                                         data_length + remaining_data_length;
1906                                 *((__be32 *)buf) = cpu_to_be32(rfc1002_len);
1907                                 data_read = 4;
1908                                 response->first_segment = false;
1909                                 log_read(INFO, "returning rfc1002 length %d\n",
1910                                         rfc1002_len);
1911                                 goto read_rfc1002_done;
1912                         }
1913
1914                         to_copy = min_t(int, data_length - offset, to_read);
1915                         memcpy(
1916                                 buf + data_read,
1917                                 (char *)data_transfer + data_offset + offset,
1918                                 to_copy);
1919
1920                         /* move on to the next buffer? */
1921                         if (to_copy == data_length - offset) {
1922                                 queue_length--;
1923                                 /*
1924                                  * No need to lock if we are not at the
1925                                  * end of the queue
1926                                  */
1927                                 if (queue_length)
1928                                         list_del(&response->list);
1929                                 else {
1930                                         spin_lock_irq(
1931                                                 &info->reassembly_queue_lock);
1932                                         list_del(&response->list);
1933                                         spin_unlock_irq(
1934                                                 &info->reassembly_queue_lock);
1935                                 }
1936                                 queue_removed++;
1937                                 info->count_reassembly_queue--;
1938                                 info->count_dequeue_reassembly_queue++;
1939                                 put_receive_buffer(info, response);
1940                                 offset = 0;
1941                                 log_read(INFO, "put_receive_buffer offset=0\n");
1942                         } else
1943                                 offset += to_copy;
1944
1945                         to_read -= to_copy;
1946                         data_read += to_copy;
1947
1948                         log_read(INFO, "_get_first_reassembly memcpy %d bytes "
1949                                 "data_transfer_length-offset=%d after that "
1950                                 "to_read=%d data_read=%d offset=%d\n",
1951                                 to_copy, data_length - offset,
1952                                 to_read, data_read, offset);
1953                 }
1954
1955                 spin_lock_irq(&info->reassembly_queue_lock);
1956                 info->reassembly_data_length -= data_read;
1957                 info->reassembly_queue_length -= queue_removed;
1958                 spin_unlock_irq(&info->reassembly_queue_lock);
1959
1960                 info->first_entry_offset = offset;
1961                 log_read(INFO, "returning to thread data_read=%d "
1962                         "reassembly_data_length=%d first_entry_offset=%d\n",
1963                         data_read, info->reassembly_data_length,
1964                         info->first_entry_offset);
1965 read_rfc1002_done:
1966                 return data_read;
1967         }
1968
1969         log_read(INFO, "wait_event on more data\n");
1970         rc = wait_event_interruptible(
1971                 info->wait_reassembly_queue,
1972                 info->reassembly_data_length >= size ||
1973                         info->transport_status != SMBD_CONNECTED);
1974         /* Don't return any data if interrupted */
1975         if (rc)
1976                 return rc;
1977
1978         if (info->transport_status != SMBD_CONNECTED) {
1979                 log_read(ERR, "disconnected\n");
1980                 return 0;
1981         }
1982
1983         goto again;
1984 }
1985
1986 /*
1987  * Receive a page from receive reassembly queue
1988  * page: the page to read data into
1989  * to_read: the length of data to read
1990  * return value: actual data read
1991  */
1992 static int smbd_recv_page(struct smbd_connection *info,
1993                 struct page *page, unsigned int page_offset,
1994                 unsigned int to_read)
1995 {
1996         int ret;
1997         char *to_address;
1998         void *page_address;
1999
2000         /* make sure we have the page ready for read */
2001         ret = wait_event_interruptible(
2002                 info->wait_reassembly_queue,
2003                 info->reassembly_data_length >= to_read ||
2004                         info->transport_status != SMBD_CONNECTED);
2005         if (ret)
2006                 return ret;
2007
2008         /* now we can read from reassembly queue and not sleep */
2009         page_address = kmap_atomic(page);
2010         to_address = (char *) page_address + page_offset;
2011
2012         log_read(INFO, "reading from page=%p address=%p to_read=%d\n",
2013                 page, to_address, to_read);
2014
2015         ret = smbd_recv_buf(info, to_address, to_read);
2016         kunmap_atomic(page_address);
2017
2018         return ret;
2019 }
2020
2021 /*
2022  * Receive data from transport
2023  * msg: a msghdr point to the buffer, can be ITER_KVEC or ITER_BVEC
2024  * return: total bytes read, or 0. SMB Direct will not do partial read.
2025  */
2026 int smbd_recv(struct smbd_connection *info, struct msghdr *msg)
2027 {
2028         char *buf;
2029         struct page *page;
2030         unsigned int to_read, page_offset;
2031         int rc;
2032
2033         if (iov_iter_rw(&msg->msg_iter) == WRITE) {
2034                 /* It's a bug in upper layer to get there */
2035                 cifs_dbg(VFS, "CIFS: invalid msg iter dir %u\n",
2036                          iov_iter_rw(&msg->msg_iter));
2037                 rc = -EINVAL;
2038                 goto out;
2039         }
2040
2041         switch (iov_iter_type(&msg->msg_iter)) {
2042         case ITER_KVEC:
2043                 buf = msg->msg_iter.kvec->iov_base;
2044                 to_read = msg->msg_iter.kvec->iov_len;
2045                 rc = smbd_recv_buf(info, buf, to_read);
2046                 break;
2047
2048         case ITER_BVEC:
2049                 page = msg->msg_iter.bvec->bv_page;
2050                 page_offset = msg->msg_iter.bvec->bv_offset;
2051                 to_read = msg->msg_iter.bvec->bv_len;
2052                 rc = smbd_recv_page(info, page, page_offset, to_read);
2053                 break;
2054
2055         default:
2056                 /* It's a bug in upper layer to get there */
2057                 cifs_dbg(VFS, "CIFS: invalid msg type %d\n",
2058                          iov_iter_type(&msg->msg_iter));
2059                 rc = -EINVAL;
2060         }
2061
2062 out:
2063         /* SMBDirect will read it all or nothing */
2064         if (rc > 0)
2065                 msg->msg_iter.count = 0;
2066         return rc;
2067 }
2068
2069 /*
2070  * Send data to transport
2071  * Each rqst is transported as a SMBDirect payload
2072  * rqst: the data to write
2073  * return value: 0 if successfully write, otherwise error code
2074  */
2075 int smbd_send(struct TCP_Server_Info *server,
2076         int num_rqst, struct smb_rqst *rqst_array)
2077 {
2078         struct smbd_connection *info = server->smbd_conn;
2079         struct kvec vec;
2080         int nvecs;
2081         int size;
2082         unsigned int buflen, remaining_data_length;
2083         int start, i, j;
2084         int max_iov_size =
2085                 info->max_send_size - sizeof(struct smbd_data_transfer);
2086         struct kvec *iov;
2087         int rc;
2088         struct smb_rqst *rqst;
2089         int rqst_idx;
2090
2091         if (info->transport_status != SMBD_CONNECTED) {
2092                 rc = -EAGAIN;
2093                 goto done;
2094         }
2095
2096         /*
2097          * Add in the page array if there is one. The caller needs to set
2098          * rq_tailsz to PAGE_SIZE when the buffer has multiple pages and
2099          * ends at page boundary
2100          */
2101         remaining_data_length = 0;
2102         for (i = 0; i < num_rqst; i++)
2103                 remaining_data_length += smb_rqst_len(server, &rqst_array[i]);
2104
2105         if (remaining_data_length + sizeof(struct smbd_data_transfer) >
2106                 info->max_fragmented_send_size) {
2107                 log_write(ERR, "payload size %d > max size %d\n",
2108                         remaining_data_length, info->max_fragmented_send_size);
2109                 rc = -EINVAL;
2110                 goto done;
2111         }
2112
2113         log_write(INFO, "num_rqst=%d total length=%u\n",
2114                         num_rqst, remaining_data_length);
2115
2116         rqst_idx = 0;
2117 next_rqst:
2118         rqst = &rqst_array[rqst_idx];
2119         iov = rqst->rq_iov;
2120
2121         cifs_dbg(FYI, "Sending smb (RDMA): idx=%d smb_len=%lu\n",
2122                 rqst_idx, smb_rqst_len(server, rqst));
2123         for (i = 0; i < rqst->rq_nvec; i++)
2124                 dump_smb(iov[i].iov_base, iov[i].iov_len);
2125
2126
2127         log_write(INFO, "rqst_idx=%d nvec=%d rqst->rq_npages=%d rq_pagesz=%d "
2128                 "rq_tailsz=%d buflen=%lu\n",
2129                 rqst_idx, rqst->rq_nvec, rqst->rq_npages, rqst->rq_pagesz,
2130                 rqst->rq_tailsz, smb_rqst_len(server, rqst));
2131
2132         start = i = 0;
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)
2181                                         break;
2182                         }
2183                         start = i;
2184                         buflen = 0;
2185                 } else {
2186                         i++;
2187                         if (i == rqst->rq_nvec) {
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         rqst_idx++;
2232         if (rqst_idx < num_rqst)
2233                 goto next_rqst;
2234
2235 done:
2236         /*
2237          * As an optimization, we don't wait for individual I/O to finish
2238          * before sending the next one.
2239          * Send them all and wait for pending send count to get to 0
2240          * that means all the I/Os have been out and we are good to return
2241          */
2242
2243         wait_event(info->wait_send_payload_pending,
2244                 atomic_read(&info->send_payload_pending) == 0);
2245
2246         return rc;
2247 }
2248
2249 static void register_mr_done(struct ib_cq *cq, struct ib_wc *wc)
2250 {
2251         struct smbd_mr *mr;
2252         struct ib_cqe *cqe;
2253
2254         if (wc->status) {
2255                 log_rdma_mr(ERR, "status=%d\n", wc->status);
2256                 cqe = wc->wr_cqe;
2257                 mr = container_of(cqe, struct smbd_mr, cqe);
2258                 smbd_disconnect_rdma_connection(mr->conn);
2259         }
2260 }
2261
2262 /*
2263  * The work queue function that recovers MRs
2264  * We need to call ib_dereg_mr() and ib_alloc_mr() before this MR can be used
2265  * again. Both calls are slow, so finish them in a workqueue. This will not
2266  * block I/O path.
2267  * There is one workqueue that recovers MRs, there is no need to lock as the
2268  * I/O requests calling smbd_register_mr will never update the links in the
2269  * mr_list.
2270  */
2271 static void smbd_mr_recovery_work(struct work_struct *work)
2272 {
2273         struct smbd_connection *info =
2274                 container_of(work, struct smbd_connection, mr_recovery_work);
2275         struct smbd_mr *smbdirect_mr;
2276         int rc;
2277
2278         list_for_each_entry(smbdirect_mr, &info->mr_list, list) {
2279                 if (smbdirect_mr->state == MR_INVALIDATED)
2280                         ib_dma_unmap_sg(
2281                                 info->id->device, smbdirect_mr->sgl,
2282                                 smbdirect_mr->sgl_count,
2283                                 smbdirect_mr->dir);
2284                 else if (smbdirect_mr->state == MR_ERROR) {
2285
2286                         /* recover this MR entry */
2287                         rc = ib_dereg_mr(smbdirect_mr->mr);
2288                         if (rc) {
2289                                 log_rdma_mr(ERR,
2290                                         "ib_dereg_mr failed rc=%x\n",
2291                                         rc);
2292                                 smbd_disconnect_rdma_connection(info);
2293                                 continue;
2294                         }
2295
2296                         smbdirect_mr->mr = ib_alloc_mr(
2297                                 info->pd, info->mr_type,
2298                                 info->max_frmr_depth);
2299                         if (IS_ERR(smbdirect_mr->mr)) {
2300                                 log_rdma_mr(ERR,
2301                                         "ib_alloc_mr failed mr_type=%x "
2302                                         "max_frmr_depth=%x\n",
2303                                         info->mr_type,
2304                                         info->max_frmr_depth);
2305                                 smbd_disconnect_rdma_connection(info);
2306                                 continue;
2307                         }
2308                 } else
2309                         /* This MR is being used, don't recover it */
2310                         continue;
2311
2312                 smbdirect_mr->state = MR_READY;
2313
2314                 /* smbdirect_mr->state is updated by this function
2315                  * and is read and updated by I/O issuing CPUs trying
2316                  * to get a MR, the call to atomic_inc_return
2317                  * implicates a memory barrier and guarantees this
2318                  * value is updated before waking up any calls to
2319                  * get_mr() from the I/O issuing CPUs
2320                  */
2321                 if (atomic_inc_return(&info->mr_ready_count) == 1)
2322                         wake_up_interruptible(&info->wait_mr);
2323         }
2324 }
2325
2326 static void destroy_mr_list(struct smbd_connection *info)
2327 {
2328         struct smbd_mr *mr, *tmp;
2329
2330         cancel_work_sync(&info->mr_recovery_work);
2331         list_for_each_entry_safe(mr, tmp, &info->mr_list, list) {
2332                 if (mr->state == MR_INVALIDATED)
2333                         ib_dma_unmap_sg(info->id->device, mr->sgl,
2334                                 mr->sgl_count, mr->dir);
2335                 ib_dereg_mr(mr->mr);
2336                 kfree(mr->sgl);
2337                 kfree(mr);
2338         }
2339 }
2340
2341 /*
2342  * Allocate MRs used for RDMA read/write
2343  * The number of MRs will not exceed hardware capability in responder_resources
2344  * All MRs are kept in mr_list. The MR can be recovered after it's used
2345  * Recovery is done in smbd_mr_recovery_work. The content of list entry changes
2346  * as MRs are used and recovered for I/O, but the list links will not change
2347  */
2348 static int allocate_mr_list(struct smbd_connection *info)
2349 {
2350         int i;
2351         struct smbd_mr *smbdirect_mr, *tmp;
2352
2353         INIT_LIST_HEAD(&info->mr_list);
2354         init_waitqueue_head(&info->wait_mr);
2355         spin_lock_init(&info->mr_list_lock);
2356         atomic_set(&info->mr_ready_count, 0);
2357         atomic_set(&info->mr_used_count, 0);
2358         init_waitqueue_head(&info->wait_for_mr_cleanup);
2359         /* Allocate more MRs (2x) than hardware responder_resources */
2360         for (i = 0; i < info->responder_resources * 2; i++) {
2361                 smbdirect_mr = kzalloc(sizeof(*smbdirect_mr), GFP_KERNEL);
2362                 if (!smbdirect_mr)
2363                         goto out;
2364                 smbdirect_mr->mr = ib_alloc_mr(info->pd, info->mr_type,
2365                                         info->max_frmr_depth);
2366                 if (IS_ERR(smbdirect_mr->mr)) {
2367                         log_rdma_mr(ERR, "ib_alloc_mr failed mr_type=%x "
2368                                 "max_frmr_depth=%x\n",
2369                                 info->mr_type, info->max_frmr_depth);
2370                         goto out;
2371                 }
2372                 smbdirect_mr->sgl = kcalloc(
2373                                         info->max_frmr_depth,
2374                                         sizeof(struct scatterlist),
2375                                         GFP_KERNEL);
2376                 if (!smbdirect_mr->sgl) {
2377                         log_rdma_mr(ERR, "failed to allocate sgl\n");
2378                         ib_dereg_mr(smbdirect_mr->mr);
2379                         goto out;
2380                 }
2381                 smbdirect_mr->state = MR_READY;
2382                 smbdirect_mr->conn = info;
2383
2384                 list_add_tail(&smbdirect_mr->list, &info->mr_list);
2385                 atomic_inc(&info->mr_ready_count);
2386         }
2387         INIT_WORK(&info->mr_recovery_work, smbd_mr_recovery_work);
2388         return 0;
2389
2390 out:
2391         kfree(smbdirect_mr);
2392
2393         list_for_each_entry_safe(smbdirect_mr, tmp, &info->mr_list, list) {
2394                 ib_dereg_mr(smbdirect_mr->mr);
2395                 kfree(smbdirect_mr->sgl);
2396                 kfree(smbdirect_mr);
2397         }
2398         return -ENOMEM;
2399 }
2400
2401 /*
2402  * Get a MR from mr_list. This function waits until there is at least one
2403  * MR available in the list. It may access the list while the
2404  * smbd_mr_recovery_work is recovering the MR list. This doesn't need a lock
2405  * as they never modify the same places. However, there may be several CPUs
2406  * issueing I/O trying to get MR at the same time, mr_list_lock is used to
2407  * protect this situation.
2408  */
2409 static struct smbd_mr *get_mr(struct smbd_connection *info)
2410 {
2411         struct smbd_mr *ret;
2412         int rc;
2413 again:
2414         rc = wait_event_interruptible(info->wait_mr,
2415                 atomic_read(&info->mr_ready_count) ||
2416                 info->transport_status != SMBD_CONNECTED);
2417         if (rc) {
2418                 log_rdma_mr(ERR, "wait_event_interruptible rc=%x\n", rc);
2419                 return NULL;
2420         }
2421
2422         if (info->transport_status != SMBD_CONNECTED) {
2423                 log_rdma_mr(ERR, "info->transport_status=%x\n",
2424                         info->transport_status);
2425                 return NULL;
2426         }
2427
2428         spin_lock(&info->mr_list_lock);
2429         list_for_each_entry(ret, &info->mr_list, list) {
2430                 if (ret->state == MR_READY) {
2431                         ret->state = MR_REGISTERED;
2432                         spin_unlock(&info->mr_list_lock);
2433                         atomic_dec(&info->mr_ready_count);
2434                         atomic_inc(&info->mr_used_count);
2435                         return ret;
2436                 }
2437         }
2438
2439         spin_unlock(&info->mr_list_lock);
2440         /*
2441          * It is possible that we could fail to get MR because other processes may
2442          * try to acquire a MR at the same time. If this is the case, retry it.
2443          */
2444         goto again;
2445 }
2446
2447 /*
2448  * Register memory for RDMA read/write
2449  * pages[]: the list of pages to register memory with
2450  * num_pages: the number of pages to register
2451  * tailsz: if non-zero, the bytes to register in the last page
2452  * writing: true if this is a RDMA write (SMB read), false for RDMA read
2453  * need_invalidate: true if this MR needs to be locally invalidated after I/O
2454  * return value: the MR registered, NULL if failed.
2455  */
2456 struct smbd_mr *smbd_register_mr(
2457         struct smbd_connection *info, struct page *pages[], int num_pages,
2458         int offset, int tailsz, bool writing, bool need_invalidate)
2459 {
2460         struct smbd_mr *smbdirect_mr;
2461         int rc, i;
2462         enum dma_data_direction dir;
2463         struct ib_reg_wr *reg_wr;
2464
2465         if (num_pages > info->max_frmr_depth) {
2466                 log_rdma_mr(ERR, "num_pages=%d max_frmr_depth=%d\n",
2467                         num_pages, info->max_frmr_depth);
2468                 return NULL;
2469         }
2470
2471         smbdirect_mr = get_mr(info);
2472         if (!smbdirect_mr) {
2473                 log_rdma_mr(ERR, "get_mr returning NULL\n");
2474                 return NULL;
2475         }
2476         smbdirect_mr->need_invalidate = need_invalidate;
2477         smbdirect_mr->sgl_count = num_pages;
2478         sg_init_table(smbdirect_mr->sgl, num_pages);
2479
2480         log_rdma_mr(INFO, "num_pages=0x%x offset=0x%x tailsz=0x%x\n",
2481                         num_pages, offset, tailsz);
2482
2483         if (num_pages == 1) {
2484                 sg_set_page(&smbdirect_mr->sgl[0], pages[0], tailsz, offset);
2485                 goto skip_multiple_pages;
2486         }
2487
2488         /* We have at least two pages to register */
2489         sg_set_page(
2490                 &smbdirect_mr->sgl[0], pages[0], PAGE_SIZE - offset, offset);
2491         i = 1;
2492         while (i < num_pages - 1) {
2493                 sg_set_page(&smbdirect_mr->sgl[i], pages[i], PAGE_SIZE, 0);
2494                 i++;
2495         }
2496         sg_set_page(&smbdirect_mr->sgl[i], pages[i],
2497                 tailsz ? tailsz : PAGE_SIZE, 0);
2498
2499 skip_multiple_pages:
2500         dir = writing ? DMA_FROM_DEVICE : DMA_TO_DEVICE;
2501         smbdirect_mr->dir = dir;
2502         rc = ib_dma_map_sg(info->id->device, smbdirect_mr->sgl, num_pages, dir);
2503         if (!rc) {
2504                 log_rdma_mr(ERR, "ib_dma_map_sg num_pages=%x dir=%x rc=%x\n",
2505                         num_pages, dir, rc);
2506                 goto dma_map_error;
2507         }
2508
2509         rc = ib_map_mr_sg(smbdirect_mr->mr, smbdirect_mr->sgl, num_pages,
2510                 NULL, PAGE_SIZE);
2511         if (rc != num_pages) {
2512                 log_rdma_mr(ERR,
2513                         "ib_map_mr_sg failed rc = %d num_pages = %x\n",
2514                         rc, num_pages);
2515                 goto map_mr_error;
2516         }
2517
2518         ib_update_fast_reg_key(smbdirect_mr->mr,
2519                 ib_inc_rkey(smbdirect_mr->mr->rkey));
2520         reg_wr = &smbdirect_mr->wr;
2521         reg_wr->wr.opcode = IB_WR_REG_MR;
2522         smbdirect_mr->cqe.done = register_mr_done;
2523         reg_wr->wr.wr_cqe = &smbdirect_mr->cqe;
2524         reg_wr->wr.num_sge = 0;
2525         reg_wr->wr.send_flags = IB_SEND_SIGNALED;
2526         reg_wr->mr = smbdirect_mr->mr;
2527         reg_wr->key = smbdirect_mr->mr->rkey;
2528         reg_wr->access = writing ?
2529                         IB_ACCESS_REMOTE_WRITE | IB_ACCESS_LOCAL_WRITE :
2530                         IB_ACCESS_REMOTE_READ;
2531
2532         /*
2533          * There is no need for waiting for complemtion on ib_post_send
2534          * on IB_WR_REG_MR. Hardware enforces a barrier and order of execution
2535          * on the next ib_post_send when we actaully send I/O to remote peer
2536          */
2537         rc = ib_post_send(info->id->qp, &reg_wr->wr, NULL);
2538         if (!rc)
2539                 return smbdirect_mr;
2540
2541         log_rdma_mr(ERR, "ib_post_send failed rc=%x reg_wr->key=%x\n",
2542                 rc, reg_wr->key);
2543
2544         /* If all failed, attempt to recover this MR by setting it MR_ERROR*/
2545 map_mr_error:
2546         ib_dma_unmap_sg(info->id->device, smbdirect_mr->sgl,
2547                 smbdirect_mr->sgl_count, smbdirect_mr->dir);
2548
2549 dma_map_error:
2550         smbdirect_mr->state = MR_ERROR;
2551         if (atomic_dec_and_test(&info->mr_used_count))
2552                 wake_up(&info->wait_for_mr_cleanup);
2553
2554         smbd_disconnect_rdma_connection(info);
2555
2556         return NULL;
2557 }
2558
2559 static void local_inv_done(struct ib_cq *cq, struct ib_wc *wc)
2560 {
2561         struct smbd_mr *smbdirect_mr;
2562         struct ib_cqe *cqe;
2563
2564         cqe = wc->wr_cqe;
2565         smbdirect_mr = container_of(cqe, struct smbd_mr, cqe);
2566         smbdirect_mr->state = MR_INVALIDATED;
2567         if (wc->status != IB_WC_SUCCESS) {
2568                 log_rdma_mr(ERR, "invalidate failed status=%x\n", wc->status);
2569                 smbdirect_mr->state = MR_ERROR;
2570         }
2571         complete(&smbdirect_mr->invalidate_done);
2572 }
2573
2574 /*
2575  * Deregister a MR after I/O is done
2576  * This function may wait if remote invalidation is not used
2577  * and we have to locally invalidate the buffer to prevent data is being
2578  * modified by remote peer after upper layer consumes it
2579  */
2580 int smbd_deregister_mr(struct smbd_mr *smbdirect_mr)
2581 {
2582         struct ib_send_wr *wr;
2583         struct smbd_connection *info = smbdirect_mr->conn;
2584         int rc = 0;
2585
2586         if (smbdirect_mr->need_invalidate) {
2587                 /* Need to finish local invalidation before returning */
2588                 wr = &smbdirect_mr->inv_wr;
2589                 wr->opcode = IB_WR_LOCAL_INV;
2590                 smbdirect_mr->cqe.done = local_inv_done;
2591                 wr->wr_cqe = &smbdirect_mr->cqe;
2592                 wr->num_sge = 0;
2593                 wr->ex.invalidate_rkey = smbdirect_mr->mr->rkey;
2594                 wr->send_flags = IB_SEND_SIGNALED;
2595
2596                 init_completion(&smbdirect_mr->invalidate_done);
2597                 rc = ib_post_send(info->id->qp, wr, NULL);
2598                 if (rc) {
2599                         log_rdma_mr(ERR, "ib_post_send failed rc=%x\n", rc);
2600                         smbd_disconnect_rdma_connection(info);
2601                         goto done;
2602                 }
2603                 wait_for_completion(&smbdirect_mr->invalidate_done);
2604                 smbdirect_mr->need_invalidate = false;
2605         } else
2606                 /*
2607                  * For remote invalidation, just set it to MR_INVALIDATED
2608                  * and defer to mr_recovery_work to recover the MR for next use
2609                  */
2610                 smbdirect_mr->state = MR_INVALIDATED;
2611
2612         /*
2613          * Schedule the work to do MR recovery for future I/Os
2614          * MR recovery is slow and we don't want it to block the current I/O
2615          */
2616         queue_work(info->workqueue, &info->mr_recovery_work);
2617
2618 done:
2619         if (atomic_dec_and_test(&info->mr_used_count))
2620                 wake_up(&info->wait_for_mr_cleanup);
2621
2622         return rc;
2623 }