fs/smb/client/smbdirect.c

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