fs/cifs/smbdirect.c

   1 /*
   2  *   Copyright (C) 2017, Microsoft Corporation.
   3  *
   4  *   Author(s): Long Li <longli@microsoft.com>
   5  *
   6  *   This program is free software;  you can redistribute it and/or modify
   7  *   it under the terms of the GNU General Public License as published by
   8  *   the Free Software Foundation; either version 2 of the License, or
   9  *   (at your option) any later version.
  10  *
  11  *   This program is distributed in the hope that it will be useful,
  12  *   but WITHOUT ANY WARRANTY;  without even the implied warranty of
  13  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See
  14  *   the GNU General Public License for more details.
  15  */
  16 #include <linux/module.h>
  17 #include <linux/highmem.h>
  18 #include "smbdirect.h"
  19 #include "cifs_debug.h"
  20
  21 static struct smbd_response *get_empty_queue_buffer(
  22                 struct smbd_connection *info);
  23 static struct smbd_response *get_receive_buffer(
  24                 struct smbd_connection *info);
  25 static void put_receive_buffer(
  26                 struct smbd_connection *info,
  27                 struct smbd_response *response);
  28 static int allocate_receive_buffers(struct smbd_connection *info, int num_buf);
  29 static void destroy_receive_buffers(struct smbd_connection *info);
  30
  31 static void put_empty_packet(
  32                 struct smbd_connection *info, struct smbd_response *response);
  33 static void enqueue_reassembly(
  34                 struct smbd_connection *info,
  35                 struct smbd_response *response, int data_length);
  36 static struct smbd_response *_get_first_reassembly(
  37                 struct smbd_connection *info);
  38
  39 static int smbd_post_recv(
  40                 struct smbd_connection *info,
  41                 struct smbd_response *response);
  42
  43 static int smbd_post_send_empty(struct smbd_connection *info);
  44
  45 /* SMBD version number */
  46 #define SMBD_V1 0x0100
  47
  48 /* Port numbers for SMBD transport */
  49 #define SMB_PORT        445
  50 #define SMBD_PORT       5445
  51
  52 /* Address lookup and resolve timeout in ms */
  53 #define RDMA_RESOLVE_TIMEOUT    5000
  54
  55 /* SMBD negotiation timeout in seconds */
  56 #define SMBD_NEGOTIATE_TIMEOUT  120
  57
  58 /* SMBD minimum receive size and fragmented sized defined in [MS-SMBD] */
  59 #define SMBD_MIN_RECEIVE_SIZE           128
  60 #define SMBD_MIN_FRAGMENTED_SIZE        131072
  61
  62 /*
  63  * Default maximum number of RDMA read/write outstanding on this connection
  64  * This value is possibly decreased during QP creation on hardware limit
  65  */
  66 #define SMBD_CM_RESPONDER_RESOURCES     32
  67
  68 /* Maximum number of retries on data transfer operations */
  69 #define SMBD_CM_RETRY                   6
  70 /* No need to retry on Receiver Not Ready since SMBD manages credits */
  71 #define SMBD_CM_RNR_RETRY               0
  72
  73 /*
  74  * User configurable initial values per SMBD transport connection
  75  * as defined in [MS-SMBD] 3.1.1.1
  76  * Those may change after a SMBD negotiation
  77  */
  78 /* The local peer's maximum number of credits to grant to the peer */
  79 int smbd_receive_credit_max = 255;
  80
  81 /* The remote peer's credit request of local peer */
  82 int smbd_send_credit_target = 255;
  83
  84 /* The maximum single message size can be sent to remote peer */
  85 int smbd_max_send_size = 1364;
  86
  87 /*  The maximum fragmented upper-layer payload receive size supported */
  88 int smbd_max_fragmented_recv_size = 1024 * 1024;
  89
  90 /*  The maximum single-message size which can be received */
  91 int smbd_max_receive_size = 8192;
  92
  93 /* The timeout to initiate send of a keepalive message on idle */
  94 int smbd_keep_alive_interval = 120;
  95
  96 /*
  97  * User configurable initial values for RDMA transport
  98  * The actual values used may be lower and are limited to hardware capabilities
  99  */
 100 /* Default maximum number of SGEs in a RDMA write/read */
 101 int smbd_max_frmr_depth = 2048;
 102
 103 /* If payload is less than this byte, use RDMA send/recv not read/write */
 104 int rdma_readwrite_threshold = 4096;
 105
 106 /* Transport logging functions
 107  * Logging are defined as classes. They can be OR'ed to define the actual
 108  * logging level via module parameter smbd_logging_class
 109  * e.g. cifs.smbd_logging_class=0xa0 will log all log_rdma_recv() and
 110  * log_rdma_event()
 111  */
 112 #define LOG_OUTGOING                    0x1
 113 #define LOG_INCOMING                    0x2
 114 #define LOG_READ                        0x4
 115 #define LOG_WRITE                       0x8
 116 #define LOG_RDMA_SEND                   0x10
 117 #define LOG_RDMA_RECV                   0x20
 118 #define LOG_KEEP_ALIVE                  0x40
 119 #define LOG_RDMA_EVENT                  0x80
 120 #define LOG_RDMA_MR                     0x100
 121 static unsigned int smbd_logging_class;
 122 module_param(smbd_logging_class, uint, 0644);
 123 MODULE_PARM_DESC(smbd_logging_class,
 124         "Logging class for SMBD transport 0x0 to 0x100");
 125
 126 #define ERR             0x0
 127 #define INFO            0x1
 128 static unsigned int smbd_logging_level = ERR;
 129 module_param(smbd_logging_level, uint, 0644);
 130 MODULE_PARM_DESC(smbd_logging_level,
 131         "Logging level for SMBD transport, 0 (default): error, 1: info");
 132
 133 #define log_rdma(level, class, fmt, args...)                            \
 134 do {                                                                    \
 135         if (level <= smbd_logging_level || class & smbd_logging_class)  \
 136                 cifs_dbg(VFS, "%s:%d " fmt, __func__, __LINE__, ##args);\
 137 } while (0)
 138
 139 #define log_outgoing(level, fmt, args...) \
 140                 log_rdma(level, LOG_OUTGOING, fmt, ##args)
 141 #define log_incoming(level, fmt, args...) \
 142                 log_rdma(level, LOG_INCOMING, fmt, ##args)
 143 #define log_read(level, fmt, args...)   log_rdma(level, LOG_READ, fmt, ##args)
 144 #define log_write(level, fmt, args...)  log_rdma(level, LOG_WRITE, fmt, ##args)
 145 #define log_rdma_send(level, fmt, args...) \
 146                 log_rdma(level, LOG_RDMA_SEND, fmt, ##args)
 147 #define log_rdma_recv(level, fmt, args...) \
 148                 log_rdma(level, LOG_RDMA_RECV, fmt, ##args)
 149 #define log_keep_alive(level, fmt, args...) \
 150                 log_rdma(level, LOG_KEEP_ALIVE, fmt, ##args)
 151 #define log_rdma_event(level, fmt, args...) \
 152                 log_rdma(level, LOG_RDMA_EVENT, fmt, ##args)
 153 #define log_rdma_mr(level, fmt, args...) \
 154                 log_rdma(level, LOG_RDMA_MR, fmt, ##args)
 155
 156 /*
 157  * Destroy the transport and related RDMA and memory resources
 158  * Need to go through all the pending counters and make sure on one is using
 159  * the transport while it is destroyed
 160  */
 161 static void smbd_destroy_rdma_work(struct work_struct *work)
 162 {
 163         struct smbd_response *response;
 164         struct smbd_connection *info =
 165                 container_of(work, struct smbd_connection, destroy_work);
 166         unsigned long flags;
 167
 168         log_rdma_event(INFO, "destroying qp\n");
 169         ib_drain_qp(info->id->qp);
 170         rdma_destroy_qp(info->id);
 171
 172         /* Unblock all I/O waiting on the send queue */
 173         wake_up_interruptible_all(&info->wait_send_queue);
 174
 175         log_rdma_event(INFO, "cancelling idle timer\n");
 176         cancel_delayed_work_sync(&info->idle_timer_work);
 177         log_rdma_event(INFO, "cancelling send immediate work\n");
 178         cancel_delayed_work_sync(&info->send_immediate_work);
 179
 180         log_rdma_event(INFO, "wait for all recv to finish\n");
 181         wake_up_interruptible(&info->wait_reassembly_queue);
 182         wait_event(info->wait_smbd_recv_pending,
 183                 info->smbd_recv_pending == 0);
 184
 185         log_rdma_event(INFO, "wait for all send posted to IB to finish\n");
 186         wait_event(info->wait_send_pending,
 187                 atomic_read(&info->send_pending) == 0);
 188         wait_event(info->wait_send_payload_pending,
 189                 atomic_read(&info->send_payload_pending) == 0);
 190
 191         /* It's not posssible for upper layer to get to reassembly */
 192         log_rdma_event(INFO, "drain the reassembly queue\n");
 193         do {
 194                 spin_lock_irqsave(&info->reassembly_queue_lock, flags);
 195                 response = _get_first_reassembly(info);
 196                 if (response) {
 197                         list_del(&response->list);
 198                         spin_unlock_irqrestore(
 199                                 &info->reassembly_queue_lock, flags);
 200                         put_receive_buffer(info, response);
 201                 }
 202         } while (response);
 203         spin_unlock_irqrestore(&info->reassembly_queue_lock, flags);
 204         info->reassembly_data_length = 0;
 205
 206         log_rdma_event(INFO, "free receive buffers\n");
 207         wait_event(info->wait_receive_queues,
 208                 info->count_receive_queue + info->count_empty_packet_queue
 209                         == info->receive_credit_max);
 210         destroy_receive_buffers(info);
 211
 212         ib_free_cq(info->send_cq);
 213         ib_free_cq(info->recv_cq);
 214         ib_dealloc_pd(info->pd);
 215         rdma_destroy_id(info->id);
 216
 217         /* free mempools */
 218         mempool_destroy(info->request_mempool);
 219         kmem_cache_destroy(info->request_cache);
 220
 221         mempool_destroy(info->response_mempool);
 222         kmem_cache_destroy(info->response_cache);
 223
 224         info->transport_status = SMBD_DESTROYED;
 225         wake_up_all(&info->wait_destroy);
 226 }
 227
 228 static int smbd_process_disconnected(struct smbd_connection *info)
 229 {
 230         schedule_work(&info->destroy_work);
 231         return 0;
 232 }
 233
 234 static void smbd_disconnect_rdma_work(struct work_struct *work)
 235 {
 236         struct smbd_connection *info =
 237                 container_of(work, struct smbd_connection, disconnect_work);
 238
 239         if (info->transport_status == SMBD_CONNECTED) {
 240                 info->transport_status = SMBD_DISCONNECTING;
 241                 rdma_disconnect(info->id);
 242         }
 243 }
 244
 245 static void smbd_disconnect_rdma_connection(struct smbd_connection *info)
 246 {
 247         queue_work(info->workqueue, &info->disconnect_work);
 248 }
 249
 250 /* Upcall from RDMA CM */
 251 static int smbd_conn_upcall(
 252                 struct rdma_cm_id *id, struct rdma_cm_event *event)
 253 {
 254         struct smbd_connection *info = id->context;
 255
 256         log_rdma_event(INFO, "event=%d status=%d\n",
 257                 event->event, event->status);
 258
 259         switch (event->event) {
 260         case RDMA_CM_EVENT_ADDR_RESOLVED:
 261         case RDMA_CM_EVENT_ROUTE_RESOLVED:
 262                 info->ri_rc = 0;
 263                 complete(&info->ri_done);
 264                 break;
 265
 266         case RDMA_CM_EVENT_ADDR_ERROR:
 267                 info->ri_rc = -EHOSTUNREACH;
 268                 complete(&info->ri_done);
 269                 break;
 270
 271         case RDMA_CM_EVENT_ROUTE_ERROR:
 272                 info->ri_rc = -ENETUNREACH;
 273                 complete(&info->ri_done);
 274                 break;
 275
 276         case RDMA_CM_EVENT_ESTABLISHED:
 277                 log_rdma_event(INFO, "connected event=%d\n", event->event);
 278                 info->transport_status = SMBD_CONNECTED;
 279                 wake_up_interruptible(&info->conn_wait);
 280                 break;
 281
 282         case RDMA_CM_EVENT_CONNECT_ERROR:
 283         case RDMA_CM_EVENT_UNREACHABLE:
 284         case RDMA_CM_EVENT_REJECTED:
 285                 log_rdma_event(INFO, "connecting failed event=%d\n", event->event);
 286                 info->transport_status = SMBD_DISCONNECTED;
 287                 wake_up_interruptible(&info->conn_wait);
 288                 break;
 289
 290         case RDMA_CM_EVENT_DEVICE_REMOVAL:
 291         case RDMA_CM_EVENT_DISCONNECTED:
 292                 /* This happenes when we fail the negotiation */
 293                 if (info->transport_status == SMBD_NEGOTIATE_FAILED) {
 294                         info->transport_status = SMBD_DISCONNECTED;
 295                         wake_up(&info->conn_wait);
 296                         break;
 297                 }
 298
 299                 info->transport_status = SMBD_DISCONNECTED;
 300                 smbd_process_disconnected(info);
 301                 break;
 302
 303         default:
 304                 break;
 305         }
 306
 307         return 0;
 308 }
 309
 310 /* Upcall from RDMA QP */
 311 static void
 312 smbd_qp_async_error_upcall(struct ib_event *event, void *context)
 313 {
 314         struct smbd_connection *info = context;
 315
 316         log_rdma_event(ERR, "%s on device %s info %p\n",
 317                 ib_event_msg(event->event), event->device->name, info);
 318
 319         switch (event->event) {
 320         case IB_EVENT_CQ_ERR:
 321         case IB_EVENT_QP_FATAL:
 322                 smbd_disconnect_rdma_connection(info);
 323
 324         default:
 325                 break;
 326         }
 327 }
 328
 329 static inline void *smbd_request_payload(struct smbd_request *request)
 330 {
 331         return (void *)request->packet;
 332 }
 333
 334 static inline void *smbd_response_payload(struct smbd_response *response)
 335 {
 336         return (void *)response->packet;
 337 }
 338
 339 /* Called when a RDMA send is done */
 340 static void send_done(struct ib_cq *cq, struct ib_wc *wc)
 341 {
 342         int i;
 343         struct smbd_request *request =
 344                 container_of(wc->wr_cqe, struct smbd_request, cqe);
 345
 346         log_rdma_send(INFO, "smbd_request %p completed wc->status=%d\n",
 347                 request, wc->status);
 348
 349         if (wc->status != IB_WC_SUCCESS || wc->opcode != IB_WC_SEND) {
 350                 log_rdma_send(ERR, "wc->status=%d wc->opcode=%d\n",
 351                         wc->status, wc->opcode);
 352                 smbd_disconnect_rdma_connection(request->info);
 353         }
 354
 355         for (i = 0; i < request->num_sge; i++)
 356                 ib_dma_unmap_single(request->info->id->device,
 357                         request->sge[i].addr,
 358                         request->sge[i].length,
 359                         DMA_TO_DEVICE);
 360
 361         if (request->has_payload) {
 362                 if (atomic_dec_and_test(&request->info->send_payload_pending))
 363                         wake_up(&request->info->wait_send_payload_pending);
 364         } else {
 365                 if (atomic_dec_and_test(&request->info->send_pending))
 366                         wake_up(&request->info->wait_send_pending);
 367         }
 368
 369         mempool_free(request, request->info->request_mempool);
 370 }
 371
 372 static void dump_smbd_negotiate_resp(struct smbd_negotiate_resp *resp)
 373 {
 374         log_rdma_event(INFO, "resp message min_version %u max_version %u "
 375                 "negotiated_version %u credits_requested %u "
 376                 "credits_granted %u status %u max_readwrite_size %u "
 377                 "preferred_send_size %u max_receive_size %u "
 378                 "max_fragmented_size %u\n",
 379                 resp->min_version, resp->max_version, resp->negotiated_version,
 380                 resp->credits_requested, resp->credits_granted, resp->status,
 381                 resp->max_readwrite_size, resp->preferred_send_size,
 382                 resp->max_receive_size, resp->max_fragmented_size);
 383 }
 384
 385 /*
 386  * Process a negotiation response message, according to [MS-SMBD]3.1.5.7
 387  * response, packet_length: the negotiation response message
 388  * return value: true if negotiation is a success, false if failed
 389  */
 390 static bool process_negotiation_response(
 391                 struct smbd_response *response, int packet_length)
 392 {
 393         struct smbd_connection *info = response->info;
 394         struct smbd_negotiate_resp *packet = smbd_response_payload(response);
 395
 396         if (packet_length < sizeof(struct smbd_negotiate_resp)) {
 397                 log_rdma_event(ERR,
 398                         "error: packet_length=%d\n", packet_length);
 399                 return false;
 400         }
 401
 402         if (le16_to_cpu(packet->negotiated_version) != SMBD_V1) {
 403                 log_rdma_event(ERR, "error: negotiated_version=%x\n",
 404                         le16_to_cpu(packet->negotiated_version));
 405                 return false;
 406         }
 407         info->protocol = le16_to_cpu(packet->negotiated_version);
 408
 409         if (packet->credits_requested == 0) {
 410                 log_rdma_event(ERR, "error: credits_requested==0\n");
 411                 return false;
 412         }
 413         info->receive_credit_target = le16_to_cpu(packet->credits_requested);
 414
 415         if (packet->credits_granted == 0) {
 416                 log_rdma_event(ERR, "error: credits_granted==0\n");
 417                 return false;
 418         }
 419         atomic_set(&info->send_credits, le16_to_cpu(packet->credits_granted));
 420
 421         atomic_set(&info->receive_credits, 0);
 422
 423         if (le32_to_cpu(packet->preferred_send_size) > info->max_receive_size) {
 424                 log_rdma_event(ERR, "error: preferred_send_size=%d\n",
 425                         le32_to_cpu(packet->preferred_send_size));
 426                 return false;
 427         }
 428         info->max_receive_size = le32_to_cpu(packet->preferred_send_size);
 429
 430         if (le32_to_cpu(packet->max_receive_size) < SMBD_MIN_RECEIVE_SIZE) {
 431                 log_rdma_event(ERR, "error: max_receive_size=%d\n",
 432                         le32_to_cpu(packet->max_receive_size));
 433                 return false;
 434         }
 435         info->max_send_size = min_t(int, info->max_send_size,
 436                                         le32_to_cpu(packet->max_receive_size));
 437
 438         if (le32_to_cpu(packet->max_fragmented_size) <
 439                         SMBD_MIN_FRAGMENTED_SIZE) {
 440                 log_rdma_event(ERR, "error: max_fragmented_size=%d\n",
 441                         le32_to_cpu(packet->max_fragmented_size));
 442                 return false;
 443         }
 444         info->max_fragmented_send_size =
 445                 le32_to_cpu(packet->max_fragmented_size);
 446
 447         return true;
 448 }
 449
 450 /*
 451  * Check and schedule to send an immediate packet
 452  * This is used to extend credtis to remote peer to keep the transport busy
 453  */
 454 static void check_and_send_immediate(struct smbd_connection *info)
 455 {
 456         if (info->transport_status != SMBD_CONNECTED)
 457                 return;
 458
 459         info->send_immediate = true;
 460
 461         /*
 462          * Promptly send a packet if our peer is running low on receive
 463          * credits
 464          */
 465         if (atomic_read(&info->receive_credits) <
 466                 info->receive_credit_target - 1)
 467                 queue_delayed_work(
 468                         info->workqueue, &info->send_immediate_work, 0);
 469 }
 470
 471 static void smbd_post_send_credits(struct work_struct *work)
 472 {
 473         int ret = 0;
 474         int use_receive_queue = 1;
 475         int rc;
 476         struct smbd_response *response;
 477         struct smbd_connection *info =
 478                 container_of(work, struct smbd_connection,
 479                         post_send_credits_work);
 480
 481         if (info->transport_status != SMBD_CONNECTED) {
 482                 wake_up(&info->wait_receive_queues);
 483                 return;
 484         }
 485
 486         if (info->receive_credit_target >
 487                 atomic_read(&info->receive_credits)) {
 488                 while (true) {
 489                         if (use_receive_queue)
 490                                 response = get_receive_buffer(info);
 491                         else
 492                                 response = get_empty_queue_buffer(info);
 493                         if (!response) {
 494                                 /* now switch to emtpy packet queue */
 495                                 if (use_receive_queue) {
 496                                         use_receive_queue = 0;
 497                                         continue;
 498                                 } else
 499                                         break;
 500                         }
 501
 502                         response->type = SMBD_TRANSFER_DATA;
 503                         response->first_segment = false;
 504                         rc = smbd_post_recv(info, response);
 505                         if (rc) {
 506                                 log_rdma_recv(ERR,
 507                                         "post_recv failed rc=%d\n", rc);
 508                                 put_receive_buffer(info, response);
 509                                 break;
 510                         }
 511
 512                         ret++;
 513                 }
 514         }
 515
 516         spin_lock(&info->lock_new_credits_offered);
 517         info->new_credits_offered += ret;
 518         spin_unlock(&info->lock_new_credits_offered);
 519
 520         atomic_add(ret, &info->receive_credits);
 521
 522         /* Check if we can post new receive and grant credits to peer */
 523         check_and_send_immediate(info);
 524 }
 525
 526 static void smbd_recv_done_work(struct work_struct *work)
 527 {
 528         struct smbd_connection *info =
 529                 container_of(work, struct smbd_connection, recv_done_work);
 530
 531         /*
 532          * We may have new send credits granted from remote peer
 533          * If any sender is blcoked on lack of credets, unblock it
 534          */
 535         if (atomic_read(&info->send_credits))
 536                 wake_up_interruptible(&info->wait_send_queue);
 537
 538         /*
 539          * Check if we need to send something to remote peer to
 540          * grant more credits or respond to KEEP_ALIVE packet
 541          */
 542         check_and_send_immediate(info);
 543 }
 544
 545 /* Called from softirq, when recv is done */
 546 static void recv_done(struct ib_cq *cq, struct ib_wc *wc)
 547 {
 548         struct smbd_data_transfer *data_transfer;
 549         struct smbd_response *response =
 550                 container_of(wc->wr_cqe, struct smbd_response, cqe);
 551         struct smbd_connection *info = response->info;
 552         int data_length = 0;
 553
 554         log_rdma_recv(INFO, "response=%p type=%d wc status=%d wc opcode %d "
 555                       "byte_len=%d pkey_index=%x\n",
 556                 response, response->type, wc->status, wc->opcode,
 557                 wc->byte_len, wc->pkey_index);
 558
 559         if (wc->status != IB_WC_SUCCESS || wc->opcode != IB_WC_RECV) {
 560                 log_rdma_recv(INFO, "wc->status=%d opcode=%d\n",
 561                         wc->status, wc->opcode);
 562                 smbd_disconnect_rdma_connection(info);
 563                 goto error;
 564         }
 565
 566         ib_dma_sync_single_for_cpu(
 567                 wc->qp->device,
 568                 response->sge.addr,
 569                 response->sge.length,
 570                 DMA_FROM_DEVICE);
 571
 572         switch (response->type) {
 573         /* SMBD negotiation response */
 574         case SMBD_NEGOTIATE_RESP:
 575                 dump_smbd_negotiate_resp(smbd_response_payload(response));
 576                 info->full_packet_received = true;
 577                 info->negotiate_done =
 578                         process_negotiation_response(response, wc->byte_len);
 579                 complete(&info->negotiate_completion);
 580                 break;
 581
 582         /* SMBD data transfer packet */
 583         case SMBD_TRANSFER_DATA:
 584                 data_transfer = smbd_response_payload(response);
 585                 data_length = le32_to_cpu(data_transfer->data_length);
 586
 587                 /*
 588                  * If this is a packet with data playload place the data in
 589                  * reassembly queue and wake up the reading thread
 590                  */
 591                 if (data_length) {
 592                         if (info->full_packet_received)
 593                                 response->first_segment = true;
 594
 595                         if (le32_to_cpu(data_transfer->remaining_data_length))
 596                                 info->full_packet_received = false;
 597                         else
 598                                 info->full_packet_received = true;
 599
 600                         enqueue_reassembly(
 601                                 info,
 602                                 response,
 603                                 data_length);
 604                 } else
 605                         put_empty_packet(info, response);
 606
 607                 if (data_length)
 608                         wake_up_interruptible(&info->wait_reassembly_queue);
 609
 610                 atomic_dec(&info->receive_credits);
 611                 info->receive_credit_target =
 612                         le16_to_cpu(data_transfer->credits_requested);
 613                 atomic_add(le16_to_cpu(data_transfer->credits_granted),
 614                         &info->send_credits);
 615
 616                 log_incoming(INFO, "data flags %d data_offset %d "
 617                         "data_length %d remaining_data_length %d\n",
 618                         le16_to_cpu(data_transfer->flags),
 619                         le32_to_cpu(data_transfer->data_offset),
 620                         le32_to_cpu(data_transfer->data_length),
 621                         le32_to_cpu(data_transfer->remaining_data_length));
 622
 623                 /* Send a KEEP_ALIVE response right away if requested */
 624                 info->keep_alive_requested = KEEP_ALIVE_NONE;
 625                 if (le16_to_cpu(data_transfer->flags) &
 626                                 SMB_DIRECT_RESPONSE_REQUESTED) {
 627                         info->keep_alive_requested = KEEP_ALIVE_PENDING;
 628                 }
 629
 630                 queue_work(info->workqueue, &info->recv_done_work);
 631                 return;
 632
 633         default:
 634                 log_rdma_recv(ERR,
 635                         "unexpected response type=%d\n", response->type);
 636         }
 637
 638 error:
 639         put_receive_buffer(info, response);
 640 }
 641
 642 static struct rdma_cm_id *smbd_create_id(
 643                 struct smbd_connection *info,
 644                 struct sockaddr *dstaddr, int port)
 645 {
 646         struct rdma_cm_id *id;
 647         int rc;
 648         __be16 *sport;
 649
 650         id = rdma_create_id(&init_net, smbd_conn_upcall, info,
 651                 RDMA_PS_TCP, IB_QPT_RC);
 652         if (IS_ERR(id)) {
 653                 rc = PTR_ERR(id);
 654                 log_rdma_event(ERR, "rdma_create_id() failed %i\n", rc);
 655                 return id;
 656         }
 657
 658         if (dstaddr->sa_family == AF_INET6)
 659                 sport = &((struct sockaddr_in6 *)dstaddr)->sin6_port;
 660         else
 661                 sport = &((struct sockaddr_in *)dstaddr)->sin_port;
 662
 663         *sport = htons(port);
 664
 665         init_completion(&info->ri_done);
 666         info->ri_rc = -ETIMEDOUT;
 667
 668         rc = rdma_resolve_addr(id, NULL, (struct sockaddr *)dstaddr,
 669                 RDMA_RESOLVE_TIMEOUT);
 670         if (rc) {
 671                 log_rdma_event(ERR, "rdma_resolve_addr() failed %i\n", rc);
 672                 goto out;
 673         }
 674         wait_for_completion_interruptible_timeout(
 675                 &info->ri_done, msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT));
 676         rc = info->ri_rc;
 677         if (rc) {
 678                 log_rdma_event(ERR, "rdma_resolve_addr() completed %i\n", rc);
 679                 goto out;
 680         }
 681
 682         info->ri_rc = -ETIMEDOUT;
 683         rc = rdma_resolve_route(id, RDMA_RESOLVE_TIMEOUT);
 684         if (rc) {
 685                 log_rdma_event(ERR, "rdma_resolve_route() failed %i\n", rc);
 686                 goto out;
 687         }
 688         wait_for_completion_interruptible_timeout(
 689                 &info->ri_done, msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT));
 690         rc = info->ri_rc;
 691         if (rc) {
 692                 log_rdma_event(ERR, "rdma_resolve_route() completed %i\n", rc);
 693                 goto out;
 694         }
 695
 696         return id;
 697
 698 out:
 699         rdma_destroy_id(id);
 700         return ERR_PTR(rc);
 701 }
 702
 703 /*
 704  * Test if FRWR (Fast Registration Work Requests) is supported on the device
 705  * This implementation requries FRWR on RDMA read/write
 706  * return value: true if it is supported
 707  */
 708 static bool frwr_is_supported(struct ib_device_attr *attrs)
 709 {
 710         if (!(attrs->device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS))
 711                 return false;
 712         if (attrs->max_fast_reg_page_list_len == 0)
 713                 return false;
 714         return true;
 715 }
 716
 717 static int smbd_ia_open(
 718                 struct smbd_connection *info,
 719                 struct sockaddr *dstaddr, int port)
 720 {
 721         int rc;
 722
 723         info->id = smbd_create_id(info, dstaddr, port);
 724         if (IS_ERR(info->id)) {
 725                 rc = PTR_ERR(info->id);
 726                 goto out1;
 727         }
 728
 729         if (!frwr_is_supported(&info->id->device->attrs)) {
 730                 log_rdma_event(ERR,
 731                         "Fast Registration Work Requests "
 732                         "(FRWR) is not supported\n");
 733                 log_rdma_event(ERR,
 734                         "Device capability flags = %llx "
 735                         "max_fast_reg_page_list_len = %u\n",
 736                         info->id->device->attrs.device_cap_flags,
 737                         info->id->device->attrs.max_fast_reg_page_list_len);
 738                 rc = -EPROTONOSUPPORT;
 739                 goto out2;
 740         }
 741
 742         info->pd = ib_alloc_pd(info->id->device, 0);
 743         if (IS_ERR(info->pd)) {
 744                 rc = PTR_ERR(info->pd);
 745                 log_rdma_event(ERR, "ib_alloc_pd() returned %d\n", rc);
 746                 goto out2;
 747         }
 748
 749         return 0;
 750
 751 out2:
 752         rdma_destroy_id(info->id);
 753         info->id = NULL;
 754
 755 out1:
 756         return rc;
 757 }
 758
 759 /*
 760  * Send a negotiation request message to the peer
 761  * The negotiation procedure is in [MS-SMBD] 3.1.5.2 and 3.1.5.3
 762  * After negotiation, the transport is connected and ready for
 763  * carrying upper layer SMB payload
 764  */
 765 static int smbd_post_send_negotiate_req(struct smbd_connection *info)
 766 {
 767         struct ib_send_wr send_wr, *send_wr_fail;
 768         int rc = -ENOMEM;
 769         struct smbd_request *request;
 770         struct smbd_negotiate_req *packet;
 771
 772         request = mempool_alloc(info->request_mempool, GFP_KERNEL);
 773         if (!request)
 774                 return rc;
 775
 776         request->info = info;
 777
 778         packet = smbd_request_payload(request);
 779         packet->min_version = cpu_to_le16(SMBD_V1);
 780         packet->max_version = cpu_to_le16(SMBD_V1);
 781         packet->reserved = 0;
 782         packet->credits_requested = cpu_to_le16(info->send_credit_target);
 783         packet->preferred_send_size = cpu_to_le32(info->max_send_size);
 784         packet->max_receive_size = cpu_to_le32(info->max_receive_size);
 785         packet->max_fragmented_size =
 786                 cpu_to_le32(info->max_fragmented_recv_size);
 787
 788         request->num_sge = 1;
 789         request->sge[0].addr = ib_dma_map_single(
 790                                 info->id->device, (void *)packet,
 791                                 sizeof(*packet), DMA_TO_DEVICE);
 792         if (ib_dma_mapping_error(info->id->device, request->sge[0].addr)) {
 793                 rc = -EIO;
 794                 goto dma_mapping_failed;
 795         }
 796
 797         request->sge[0].length = sizeof(*packet);
 798         request->sge[0].lkey = info->pd->local_dma_lkey;
 799
 800         ib_dma_sync_single_for_device(
 801                 info->id->device, request->sge[0].addr,
 802                 request->sge[0].length, DMA_TO_DEVICE);
 803
 804         request->cqe.done = send_done;
 805
 806         send_wr.next = NULL;
 807         send_wr.wr_cqe = &request->cqe;
 808         send_wr.sg_list = request->sge;
 809         send_wr.num_sge = request->num_sge;
 810         send_wr.opcode = IB_WR_SEND;
 811         send_wr.send_flags = IB_SEND_SIGNALED;
 812
 813         log_rdma_send(INFO, "sge addr=%llx length=%x lkey=%x\n",
 814                 request->sge[0].addr,
 815                 request->sge[0].length, request->sge[0].lkey);
 816
 817         request->has_payload = false;
 818         atomic_inc(&info->send_pending);
 819         rc = ib_post_send(info->id->qp, &send_wr, &send_wr_fail);
 820         if (!rc)
 821                 return 0;
 822
 823         /* if we reach here, post send failed */
 824         log_rdma_send(ERR, "ib_post_send failed rc=%d\n", rc);
 825         atomic_dec(&info->send_pending);
 826         ib_dma_unmap_single(info->id->device, request->sge[0].addr,
 827                 request->sge[0].length, DMA_TO_DEVICE);
 828
 829 dma_mapping_failed:
 830         mempool_free(request, info->request_mempool);
 831         return rc;
 832 }
 833
 834 /*
 835  * Extend the credits to remote peer
 836  * This implements [MS-SMBD] 3.1.5.9
 837  * The idea is that we should extend credits to remote peer as quickly as
 838  * it's allowed, to maintain data flow. We allocate as much receive
 839  * buffer as possible, and extend the receive credits to remote peer
 840  * return value: the new credtis being granted.
 841  */
 842 static int manage_credits_prior_sending(struct smbd_connection *info)
 843 {
 844         int new_credits;
 845
 846         spin_lock(&info->lock_new_credits_offered);
 847         new_credits = info->new_credits_offered;
 848         info->new_credits_offered = 0;
 849         spin_unlock(&info->lock_new_credits_offered);
 850
 851         return new_credits;
 852 }
 853
 854 /*
 855  * Check if we need to send a KEEP_ALIVE message
 856  * The idle connection timer triggers a KEEP_ALIVE message when expires
 857  * SMB_DIRECT_RESPONSE_REQUESTED is set in the message flag to have peer send
 858  * back a response.
 859  * return value:
 860  * 1 if SMB_DIRECT_RESPONSE_REQUESTED needs to be set
 861  * 0: otherwise
 862  */
 863 static int manage_keep_alive_before_sending(struct smbd_connection *info)
 864 {
 865         if (info->keep_alive_requested == KEEP_ALIVE_PENDING) {
 866                 info->keep_alive_requested = KEEP_ALIVE_SENT;
 867                 return 1;
 868         }
 869         return 0;
 870 }
 871
 872 /*
 873  * Build and prepare the SMBD packet header
 874  * This function waits for avaialbe send credits and build a SMBD packet
 875  * header. The caller then optional append payload to the packet after
 876  * the header
 877  * intput values
 878  * size: the size of the payload
 879  * remaining_data_length: remaining data to send if this is part of a
 880  * fragmented packet
 881  * output values
 882  * request_out: the request allocated from this function
 883  * return values: 0 on success, otherwise actual error code returned
 884  */
 885 static int smbd_create_header(struct smbd_connection *info,
 886                 int size, int remaining_data_length,
 887                 struct smbd_request **request_out)
 888 {
 889         struct smbd_request *request;
 890         struct smbd_data_transfer *packet;
 891         int header_length;
 892         int rc;
 893
 894         /* Wait for send credits. A SMBD packet needs one credit */
 895         rc = wait_event_interruptible(info->wait_send_queue,
 896                 atomic_read(&info->send_credits) > 0 ||
 897                 info->transport_status != SMBD_CONNECTED);
 898         if (rc)
 899                 return rc;
 900
 901         if (info->transport_status != SMBD_CONNECTED) {
 902                 log_outgoing(ERR, "disconnected not sending\n");
 903                 return -ENOENT;
 904         }
 905         atomic_dec(&info->send_credits);
 906
 907         request = mempool_alloc(info->request_mempool, GFP_KERNEL);
 908         if (!request) {
 909                 rc = -ENOMEM;
 910                 goto err;
 911         }
 912
 913         request->info = info;
 914
 915         /* Fill in the packet header */
 916         packet = smbd_request_payload(request);
 917         packet->credits_requested = cpu_to_le16(info->send_credit_target);
 918         packet->credits_granted =
 919                 cpu_to_le16(manage_credits_prior_sending(info));
 920         info->send_immediate = false;
 921
 922         packet->flags = 0;
 923         if (manage_keep_alive_before_sending(info))
 924                 packet->flags |= cpu_to_le16(SMB_DIRECT_RESPONSE_REQUESTED);
 925
 926         packet->reserved = 0;
 927         if (!size)
 928                 packet->data_offset = 0;
 929         else
 930                 packet->data_offset = cpu_to_le32(24);
 931         packet->data_length = cpu_to_le32(size);
 932         packet->remaining_data_length = cpu_to_le32(remaining_data_length);
 933         packet->padding = 0;
 934
 935         log_outgoing(INFO, "credits_requested=%d credits_granted=%d "
 936                 "data_offset=%d data_length=%d remaining_data_length=%d\n",
 937                 le16_to_cpu(packet->credits_requested),
 938                 le16_to_cpu(packet->credits_granted),
 939                 le32_to_cpu(packet->data_offset),
 940                 le32_to_cpu(packet->data_length),
 941                 le32_to_cpu(packet->remaining_data_length));
 942
 943         /* Map the packet to DMA */
 944         header_length = sizeof(struct smbd_data_transfer);
 945         /* If this is a packet without payload, don't send padding */
 946         if (!size)
 947                 header_length = offsetof(struct smbd_data_transfer, padding);
 948
 949         request->num_sge = 1;
 950         request->sge[0].addr = ib_dma_map_single(info->id->device,
 951                                                  (void *)packet,
 952                                                  header_length,
 953                                                  DMA_BIDIRECTIONAL);
 954         if (ib_dma_mapping_error(info->id->device, request->sge[0].addr)) {
 955                 mempool_free(request, info->request_mempool);
 956                 rc = -EIO;
 957                 goto err;
 958         }
 959
 960         request->sge[0].length = header_length;
 961         request->sge[0].lkey = info->pd->local_dma_lkey;
 962
 963         *request_out = request;
 964         return 0;
 965
 966 err:
 967         atomic_inc(&info->send_credits);
 968         return rc;
 969 }
 970
 971 static void smbd_destroy_header(struct smbd_connection *info,
 972                 struct smbd_request *request)
 973 {
 974
 975         ib_dma_unmap_single(info->id->device,
 976                             request->sge[0].addr,
 977                             request->sge[0].length,
 978                             DMA_TO_DEVICE);
 979         mempool_free(request, info->request_mempool);
 980         atomic_inc(&info->send_credits);
 981 }
 982
 983 /* Post the send request */
 984 static int smbd_post_send(struct smbd_connection *info,
 985                 struct smbd_request *request, bool has_payload)
 986 {
 987         struct ib_send_wr send_wr, *send_wr_fail;
 988         int rc, i;
 989
 990         for (i = 0; i < request->num_sge; i++) {
 991                 log_rdma_send(INFO,
 992                         "rdma_request sge[%d] addr=%llu legnth=%u\n",
 993                         i, request->sge[0].addr, request->sge[0].length);
 994                 ib_dma_sync_single_for_device(
 995                         info->id->device,
 996                         request->sge[i].addr,
 997                         request->sge[i].length,
 998                         DMA_TO_DEVICE);
 999         }
1000
1001         request->cqe.done = send_done;
1002
1003         send_wr.next = NULL;
1004         send_wr.wr_cqe = &request->cqe;
1005         send_wr.sg_list = request->sge;
1006         send_wr.num_sge = request->num_sge;
1007         send_wr.opcode = IB_WR_SEND;
1008         send_wr.send_flags = IB_SEND_SIGNALED;
1009
1010         if (has_payload) {
1011                 request->has_payload = true;
1012                 atomic_inc(&info->send_payload_pending);
1013         } else {
1014                 request->has_payload = false;
1015                 atomic_inc(&info->send_pending);
1016         }
1017
1018         rc = ib_post_send(info->id->qp, &send_wr, &send_wr_fail);
1019         if (rc) {
1020                 log_rdma_send(ERR, "ib_post_send failed rc=%d\n", rc);
1021                 if (has_payload) {
1022                         if (atomic_dec_and_test(&info->send_payload_pending))
1023                                 wake_up(&info->wait_send_payload_pending);
1024                 } else {
1025                         if (atomic_dec_and_test(&info->send_pending))
1026                                 wake_up(&info->wait_send_pending);
1027                 }
1028         } else
1029                 /* Reset timer for idle connection after packet is sent */
1030                 mod_delayed_work(info->workqueue, &info->idle_timer_work,
1031                         info->keep_alive_interval*HZ);
1032
1033         return rc;
1034 }
1035
1036 static int smbd_post_send_sgl(struct smbd_connection *info,
1037         struct scatterlist *sgl, int data_length, int remaining_data_length)
1038 {
1039         int num_sgs;
1040         int i, rc;
1041         struct smbd_request *request;
1042         struct scatterlist *sg;
1043
1044         rc = smbd_create_header(
1045                 info, data_length, remaining_data_length, &request);
1046         if (rc)
1047                 return rc;
1048
1049         num_sgs = sgl ? sg_nents(sgl) : 0;
1050         for_each_sg(sgl, sg, num_sgs, i) {
1051                 request->sge[i+1].addr =
1052                         ib_dma_map_page(info->id->device, sg_page(sg),
1053                                sg->offset, sg->length, DMA_BIDIRECTIONAL);
1054                 if (ib_dma_mapping_error(
1055                                 info->id->device, request->sge[i+1].addr)) {
1056                         rc = -EIO;
1057                         request->sge[i+1].addr = 0;
1058                         goto dma_mapping_failure;
1059                 }
1060                 request->sge[i+1].length = sg->length;
1061                 request->sge[i+1].lkey = info->pd->local_dma_lkey;
1062                 request->num_sge++;
1063         }
1064
1065         rc = smbd_post_send(info, request, data_length);
1066         if (!rc)
1067                 return 0;
1068
1069 dma_mapping_failure:
1070         for (i = 1; i < request->num_sge; i++)
1071                 if (request->sge[i].addr)
1072                         ib_dma_unmap_single(info->id->device,
1073                                             request->sge[i].addr,
1074                                             request->sge[i].length,
1075                                             DMA_TO_DEVICE);
1076         smbd_destroy_header(info, request);
1077         return rc;
1078 }
1079
1080 /*
1081  * Send an empty message
1082  * Empty message is used to extend credits to peer to for keep live
1083  * while there is no upper layer payload to send at the time
1084  */
1085 static int smbd_post_send_empty(struct smbd_connection *info)
1086 {
1087         info->count_send_empty++;
1088         return smbd_post_send_sgl(info, NULL, 0, 0);
1089 }
1090
1091 /*
1092  * Post a receive request to the transport
1093  * The remote peer can only send data when a receive request is posted
1094  * The interaction is controlled by send/receive credit system
1095  */
1096 static int smbd_post_recv(
1097                 struct smbd_connection *info, struct smbd_response *response)
1098 {
1099         struct ib_recv_wr recv_wr, *recv_wr_fail = NULL;
1100         int rc = -EIO;
1101
1102         response->sge.addr = ib_dma_map_single(
1103                                 info->id->device, response->packet,
1104                                 info->max_receive_size, DMA_FROM_DEVICE);
1105         if (ib_dma_mapping_error(info->id->device, response->sge.addr))
1106                 return rc;
1107
1108         response->sge.length = info->max_receive_size;
1109         response->sge.lkey = info->pd->local_dma_lkey;
1110
1111         response->cqe.done = recv_done;
1112
1113         recv_wr.wr_cqe = &response->cqe;
1114         recv_wr.next = NULL;
1115         recv_wr.sg_list = &response->sge;
1116         recv_wr.num_sge = 1;
1117
1118         rc = ib_post_recv(info->id->qp, &recv_wr, &recv_wr_fail);
1119         if (rc) {
1120                 ib_dma_unmap_single(info->id->device, response->sge.addr,
1121                                     response->sge.length, DMA_FROM_DEVICE);
1122
1123                 log_rdma_recv(ERR, "ib_post_recv failed rc=%d\n", rc);
1124         }
1125
1126         return rc;
1127 }
1128
1129 /* Perform SMBD negotiate according to [MS-SMBD] 3.1.5.2 */
1130 static int smbd_negotiate(struct smbd_connection *info)
1131 {
1132         int rc;
1133         struct smbd_response *response = get_receive_buffer(info);
1134
1135         response->type = SMBD_NEGOTIATE_RESP;
1136         rc = smbd_post_recv(info, response);
1137         log_rdma_event(INFO,
1138                 "smbd_post_recv rc=%d iov.addr=%llx iov.length=%x "
1139                 "iov.lkey=%x\n",
1140                 rc, response->sge.addr,
1141                 response->sge.length, response->sge.lkey);
1142         if (rc)
1143                 return rc;
1144
1145         init_completion(&info->negotiate_completion);
1146         info->negotiate_done = false;
1147         rc = smbd_post_send_negotiate_req(info);
1148         if (rc)
1149                 return rc;
1150
1151         rc = wait_for_completion_interruptible_timeout(
1152                 &info->negotiate_completion, SMBD_NEGOTIATE_TIMEOUT * HZ);
1153         log_rdma_event(INFO, "wait_for_completion_timeout rc=%d\n", rc);
1154
1155         if (info->negotiate_done)
1156                 return 0;
1157
1158         if (rc == 0)
1159                 rc = -ETIMEDOUT;
1160         else if (rc == -ERESTARTSYS)
1161                 rc = -EINTR;
1162         else
1163                 rc = -ENOTCONN;
1164
1165         return rc;
1166 }
1167
1168 static void put_empty_packet(
1169                 struct smbd_connection *info, struct smbd_response *response)
1170 {
1171         spin_lock(&info->empty_packet_queue_lock);
1172         list_add_tail(&response->list, &info->empty_packet_queue);
1173         info->count_empty_packet_queue++;
1174         spin_unlock(&info->empty_packet_queue_lock);
1175
1176         queue_work(info->workqueue, &info->post_send_credits_work);
1177 }
1178
1179 /*
1180  * Implement Connection.FragmentReassemblyBuffer defined in [MS-SMBD] 3.1.1.1
1181  * This is a queue for reassembling upper layer payload and present to upper
1182  * layer. All the inncoming payload go to the reassembly queue, regardless of
1183  * if reassembly is required. The uuper layer code reads from the queue for all
1184  * incoming payloads.
1185  * Put a received packet to the reassembly queue
1186  * response: the packet received
1187  * data_length: the size of payload in this packet
1188  */
1189 static void enqueue_reassembly(
1190         struct smbd_connection *info,
1191         struct smbd_response *response,
1192         int data_length)
1193 {
1194         spin_lock(&info->reassembly_queue_lock);
1195         list_add_tail(&response->list, &info->reassembly_queue);
1196         info->reassembly_queue_length++;
1197         /*
1198          * Make sure reassembly_data_length is updated after list and
1199          * reassembly_queue_length are updated. On the dequeue side
1200          * reassembly_data_length is checked without a lock to determine
1201          * if reassembly_queue_length and list is up to date
1202          */
1203         virt_wmb();
1204         info->reassembly_data_length += data_length;
1205         spin_unlock(&info->reassembly_queue_lock);
1206         info->count_reassembly_queue++;
1207         info->count_enqueue_reassembly_queue++;
1208 }
1209
1210 /*
1211  * Get the first entry at the front of reassembly queue
1212  * Caller is responsible for locking
1213  * return value: the first entry if any, NULL if queue is empty
1214  */
1215 static struct smbd_response *_get_first_reassembly(struct smbd_connection *info)
1216 {
1217         struct smbd_response *ret = NULL;
1218
1219         if (!list_empty(&info->reassembly_queue)) {
1220                 ret = list_first_entry(
1221                         &info->reassembly_queue,
1222                         struct smbd_response, list);
1223         }
1224         return ret;
1225 }
1226
1227 static struct smbd_response *get_empty_queue_buffer(
1228                 struct smbd_connection *info)
1229 {
1230         struct smbd_response *ret = NULL;
1231         unsigned long flags;
1232
1233         spin_lock_irqsave(&info->empty_packet_queue_lock, flags);
1234         if (!list_empty(&info->empty_packet_queue)) {
1235                 ret = list_first_entry(
1236                         &info->empty_packet_queue,
1237                         struct smbd_response, list);
1238                 list_del(&ret->list);
1239                 info->count_empty_packet_queue--;
1240         }
1241         spin_unlock_irqrestore(&info->empty_packet_queue_lock, flags);
1242
1243         return ret;
1244 }
1245
1246 /*
1247  * Get a receive buffer
1248  * For each remote send, we need to post a receive. The receive buffers are
1249  * pre-allocated in advance.
1250  * return value: the receive buffer, NULL if none is available
1251  */
1252 static struct smbd_response *get_receive_buffer(struct smbd_connection *info)
1253 {
1254         struct smbd_response *ret = NULL;
1255         unsigned long flags;
1256
1257         spin_lock_irqsave(&info->receive_queue_lock, flags);
1258         if (!list_empty(&info->receive_queue)) {
1259                 ret = list_first_entry(
1260                         &info->receive_queue,
1261                         struct smbd_response, list);
1262                 list_del(&ret->list);
1263                 info->count_receive_queue--;
1264                 info->count_get_receive_buffer++;
1265         }
1266         spin_unlock_irqrestore(&info->receive_queue_lock, flags);
1267
1268         return ret;
1269 }
1270
1271 /*
1272  * Return a receive buffer
1273  * Upon returning of a receive buffer, we can post new receive and extend
1274  * more receive credits to remote peer. This is done immediately after a
1275  * receive buffer is returned.
1276  */
1277 static void put_receive_buffer(
1278         struct smbd_connection *info, struct smbd_response *response)
1279 {
1280         unsigned long flags;
1281
1282         ib_dma_unmap_single(info->id->device, response->sge.addr,
1283                 response->sge.length, DMA_FROM_DEVICE);
1284
1285         spin_lock_irqsave(&info->receive_queue_lock, flags);
1286         list_add_tail(&response->list, &info->receive_queue);
1287         info->count_receive_queue++;
1288         info->count_put_receive_buffer++;
1289         spin_unlock_irqrestore(&info->receive_queue_lock, flags);
1290
1291         queue_work(info->workqueue, &info->post_send_credits_work);
1292 }
1293
1294 /* Preallocate all receive buffer on transport establishment */
1295 static int allocate_receive_buffers(struct smbd_connection *info, int num_buf)
1296 {
1297         int i;
1298         struct smbd_response *response;
1299
1300         INIT_LIST_HEAD(&info->reassembly_queue);
1301         spin_lock_init(&info->reassembly_queue_lock);
1302         info->reassembly_data_length = 0;
1303         info->reassembly_queue_length = 0;
1304
1305         INIT_LIST_HEAD(&info->receive_queue);
1306         spin_lock_init(&info->receive_queue_lock);
1307         info->count_receive_queue = 0;
1308
1309         INIT_LIST_HEAD(&info->empty_packet_queue);
1310         spin_lock_init(&info->empty_packet_queue_lock);
1311         info->count_empty_packet_queue = 0;
1312
1313         init_waitqueue_head(&info->wait_receive_queues);
1314
1315         for (i = 0; i < num_buf; i++) {
1316                 response = mempool_alloc(info->response_mempool, GFP_KERNEL);
1317                 if (!response)
1318                         goto allocate_failed;
1319
1320                 response->info = info;
1321                 list_add_tail(&response->list, &info->receive_queue);
1322                 info->count_receive_queue++;
1323         }
1324
1325         return 0;
1326
1327 allocate_failed:
1328         while (!list_empty(&info->receive_queue)) {
1329                 response = list_first_entry(
1330                                 &info->receive_queue,
1331                                 struct smbd_response, list);
1332                 list_del(&response->list);
1333                 info->count_receive_queue--;
1334
1335                 mempool_free(response, info->response_mempool);
1336         }
1337         return -ENOMEM;
1338 }
1339
1340 static void destroy_receive_buffers(struct smbd_connection *info)
1341 {
1342         struct smbd_response *response;
1343
1344         while ((response = get_receive_buffer(info)))
1345                 mempool_free(response, info->response_mempool);
1346
1347         while ((response = get_empty_queue_buffer(info)))
1348                 mempool_free(response, info->response_mempool);
1349 }
1350
1351 /*
1352  * Check and send an immediate or keep alive packet
1353  * The condition to send those packets are defined in [MS-SMBD] 3.1.1.1
1354  * Connection.KeepaliveRequested and Connection.SendImmediate
1355  * The idea is to extend credits to server as soon as it becomes available
1356  */
1357 static void send_immediate_work(struct work_struct *work)
1358 {
1359         struct smbd_connection *info = container_of(
1360                                         work, struct smbd_connection,
1361                                         send_immediate_work.work);
1362
1363         if (info->keep_alive_requested == KEEP_ALIVE_PENDING ||
1364             info->send_immediate) {
1365                 log_keep_alive(INFO, "send an empty message\n");
1366                 smbd_post_send_empty(info);
1367         }
1368 }
1369
1370 /* Implement idle connection timer [MS-SMBD] 3.1.6.2 */
1371 static void idle_connection_timer(struct work_struct *work)
1372 {
1373         struct smbd_connection *info = container_of(
1374                                         work, struct smbd_connection,
1375                                         idle_timer_work.work);
1376
1377         if (info->keep_alive_requested != KEEP_ALIVE_NONE) {
1378                 log_keep_alive(ERR,
1379                         "error status info->keep_alive_requested=%d\n",
1380                         info->keep_alive_requested);
1381                 smbd_disconnect_rdma_connection(info);
1382                 return;
1383         }
1384
1385         log_keep_alive(INFO, "about to send an empty idle message\n");
1386         smbd_post_send_empty(info);
1387
1388         /* Setup the next idle timeout work */
1389         queue_delayed_work(info->workqueue, &info->idle_timer_work,
1390                         info->keep_alive_interval*HZ);
1391 }
1392
1393 /* Destroy this SMBD connection, called from upper layer */
1394 void smbd_destroy(struct smbd_connection *info)
1395 {
1396         log_rdma_event(INFO, "destroying rdma session\n");
1397
1398         /* Kick off the disconnection process */
1399         smbd_disconnect_rdma_connection(info);
1400
1401         log_rdma_event(INFO, "wait for transport being destroyed\n");
1402         wait_event(info->wait_destroy,
1403                 info->transport_status == SMBD_DESTROYED);
1404
1405         destroy_workqueue(info->workqueue);
1406         kfree(info);
1407 }
1408
1409 /*
1410  * Reconnect this SMBD connection, called from upper layer
1411  * return value: 0 on success, or actual error code
1412  */
1413 int smbd_reconnect(struct TCP_Server_Info *server)
1414 {
1415         log_rdma_event(INFO, "reconnecting rdma session\n");
1416
1417         if (!server->smbd_conn) {
1418                 log_rdma_event(ERR, "rdma session already destroyed\n");
1419                 return -EINVAL;
1420         }
1421
1422         /*
1423          * This is possible if transport is disconnected and we haven't received
1424          * notification from RDMA, but upper layer has detected timeout
1425          */
1426         if (server->smbd_conn->transport_status == SMBD_CONNECTED) {
1427                 log_rdma_event(INFO, "disconnecting transport\n");
1428                 smbd_disconnect_rdma_connection(server->smbd_conn);
1429         }
1430
1431         /* wait until the transport is destroyed */
1432         wait_event(server->smbd_conn->wait_destroy,
1433                 server->smbd_conn->transport_status == SMBD_DESTROYED);
1434
1435         destroy_workqueue(server->smbd_conn->workqueue);
1436         kfree(server->smbd_conn);
1437
1438         log_rdma_event(INFO, "creating rdma session\n");
1439         server->smbd_conn = smbd_get_connection(
1440                 server, (struct sockaddr *) &server->dstaddr);
1441
1442         return server->smbd_conn ? 0 : -ENOENT;
1443 }
1444
1445 static void destroy_caches_and_workqueue(struct smbd_connection *info)
1446 {
1447         destroy_receive_buffers(info);
1448         destroy_workqueue(info->workqueue);
1449         mempool_destroy(info->response_mempool);
1450         kmem_cache_destroy(info->response_cache);
1451         mempool_destroy(info->request_mempool);
1452         kmem_cache_destroy(info->request_cache);
1453 }
1454
1455 #define MAX_NAME_LEN    80
1456 static int allocate_caches_and_workqueue(struct smbd_connection *info)
1457 {
1458         char name[MAX_NAME_LEN];
1459         int rc;
1460
1461         snprintf(name, MAX_NAME_LEN, "smbd_request_%p", info);
1462         info->request_cache =
1463                 kmem_cache_create(
1464                         name,
1465                         sizeof(struct smbd_request) +
1466                                 sizeof(struct smbd_data_transfer),
1467                         0, SLAB_HWCACHE_ALIGN, NULL);
1468         if (!info->request_cache)
1469                 return -ENOMEM;
1470
1471         info->request_mempool =
1472                 mempool_create(info->send_credit_target, mempool_alloc_slab,
1473                         mempool_free_slab, info->request_cache);
1474         if (!info->request_mempool)
1475                 goto out1;
1476
1477         snprintf(name, MAX_NAME_LEN, "smbd_response_%p", info);
1478         info->response_cache =
1479                 kmem_cache_create(
1480                         name,
1481                         sizeof(struct smbd_response) +
1482                                 info->max_receive_size,
1483                         0, SLAB_HWCACHE_ALIGN, NULL);
1484         if (!info->response_cache)
1485                 goto out2;
1486
1487         info->response_mempool =
1488                 mempool_create(info->receive_credit_max, mempool_alloc_slab,
1489                        mempool_free_slab, info->response_cache);
1490         if (!info->response_mempool)
1491                 goto out3;
1492
1493         snprintf(name, MAX_NAME_LEN, "smbd_%p", info);
1494         info->workqueue = create_workqueue(name);
1495         if (!info->workqueue)
1496                 goto out4;
1497
1498         rc = allocate_receive_buffers(info, info->receive_credit_max);
1499         if (rc) {
1500                 log_rdma_event(ERR, "failed to allocate receive buffers\n");
1501                 goto out5;
1502         }
1503
1504         return 0;
1505
1506 out5:
1507         destroy_workqueue(info->workqueue);
1508 out4:
1509         mempool_destroy(info->response_mempool);
1510 out3:
1511         kmem_cache_destroy(info->response_cache);
1512 out2:
1513         mempool_destroy(info->request_mempool);
1514 out1:
1515         kmem_cache_destroy(info->request_cache);
1516         return -ENOMEM;
1517 }
1518
1519 /* Create a SMBD connection, called by upper layer */
1520 struct smbd_connection *_smbd_get_connection(
1521         struct TCP_Server_Info *server, struct sockaddr *dstaddr, int port)
1522 {
1523         int rc;
1524         struct smbd_connection *info;
1525         struct rdma_conn_param conn_param;
1526         struct ib_qp_init_attr qp_attr;
1527         struct sockaddr_in *addr_in = (struct sockaddr_in *) dstaddr;
1528
1529         info = kzalloc(sizeof(struct smbd_connection), GFP_KERNEL);
1530         if (!info)
1531                 return NULL;
1532
1533         info->transport_status = SMBD_CONNECTING;
1534         rc = smbd_ia_open(info, dstaddr, port);
1535         if (rc) {
1536                 log_rdma_event(INFO, "smbd_ia_open rc=%d\n", rc);
1537                 goto create_id_failed;
1538         }
1539
1540         if (smbd_send_credit_target > info->id->device->attrs.max_cqe ||
1541             smbd_send_credit_target > info->id->device->attrs.max_qp_wr) {
1542                 log_rdma_event(ERR,
1543                         "consider lowering send_credit_target = %d. "
1544                         "Possible CQE overrun, device "
1545                         "reporting max_cpe %d max_qp_wr %d\n",
1546                         smbd_send_credit_target,
1547                         info->id->device->attrs.max_cqe,
1548                         info->id->device->attrs.max_qp_wr);
1549                 goto config_failed;
1550         }
1551
1552         if (smbd_receive_credit_max > info->id->device->attrs.max_cqe ||
1553             smbd_receive_credit_max > info->id->device->attrs.max_qp_wr) {
1554                 log_rdma_event(ERR,
1555                         "consider lowering receive_credit_max = %d. "
1556                         "Possible CQE overrun, device "
1557                         "reporting max_cpe %d max_qp_wr %d\n",
1558                         smbd_receive_credit_max,
1559                         info->id->device->attrs.max_cqe,
1560                         info->id->device->attrs.max_qp_wr);
1561                 goto config_failed;
1562         }
1563
1564         info->receive_credit_max = smbd_receive_credit_max;
1565         info->send_credit_target = smbd_send_credit_target;
1566         info->max_send_size = smbd_max_send_size;
1567         info->max_fragmented_recv_size = smbd_max_fragmented_recv_size;
1568         info->max_receive_size = smbd_max_receive_size;
1569         info->keep_alive_interval = smbd_keep_alive_interval;
1570
1571         if (info->id->device->attrs.max_sge < SMBDIRECT_MAX_SGE) {
1572                 log_rdma_event(ERR, "warning: device max_sge = %d too small\n",
1573                         info->id->device->attrs.max_sge);
1574                 log_rdma_event(ERR, "Queue Pair creation may fail\n");
1575         }
1576
1577         info->send_cq = NULL;
1578         info->recv_cq = NULL;
1579         info->send_cq = ib_alloc_cq(info->id->device, info,
1580                         info->send_credit_target, 0, IB_POLL_SOFTIRQ);
1581         if (IS_ERR(info->send_cq)) {
1582                 info->send_cq = NULL;
1583                 goto alloc_cq_failed;
1584         }
1585
1586         info->recv_cq = ib_alloc_cq(info->id->device, info,
1587                         info->receive_credit_max, 0, IB_POLL_SOFTIRQ);
1588         if (IS_ERR(info->recv_cq)) {
1589                 info->recv_cq = NULL;
1590                 goto alloc_cq_failed;
1591         }
1592
1593         memset(&qp_attr, 0, sizeof(qp_attr));
1594         qp_attr.event_handler = smbd_qp_async_error_upcall;
1595         qp_attr.qp_context = info;
1596         qp_attr.cap.max_send_wr = info->send_credit_target;
1597         qp_attr.cap.max_recv_wr = info->receive_credit_max;
1598         qp_attr.cap.max_send_sge = SMBDIRECT_MAX_SGE;
1599         qp_attr.cap.max_recv_sge = SMBDIRECT_MAX_SGE;
1600         qp_attr.cap.max_inline_data = 0;
1601         qp_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
1602         qp_attr.qp_type = IB_QPT_RC;
1603         qp_attr.send_cq = info->send_cq;
1604         qp_attr.recv_cq = info->recv_cq;
1605         qp_attr.port_num = ~0;
1606
1607         rc = rdma_create_qp(info->id, info->pd, &qp_attr);
1608         if (rc) {
1609                 log_rdma_event(ERR, "rdma_create_qp failed %i\n", rc);
1610                 goto create_qp_failed;
1611         }
1612
1613         memset(&conn_param, 0, sizeof(conn_param));
1614         conn_param.initiator_depth = 0;
1615
1616         conn_param.retry_count = SMBD_CM_RETRY;
1617         conn_param.rnr_retry_count = SMBD_CM_RNR_RETRY;
1618         conn_param.flow_control = 0;
1619         init_waitqueue_head(&info->wait_destroy);
1620
1621         log_rdma_event(INFO, "connecting to IP %pI4 port %d\n",
1622                 &addr_in->sin_addr, port);
1623
1624         init_waitqueue_head(&info->conn_wait);
1625         rc = rdma_connect(info->id, &conn_param);
1626         if (rc) {
1627                 log_rdma_event(ERR, "rdma_connect() failed with %i\n", rc);
1628                 goto rdma_connect_failed;
1629         }
1630
1631         wait_event_interruptible(
1632                 info->conn_wait, info->transport_status != SMBD_CONNECTING);
1633
1634         if (info->transport_status != SMBD_CONNECTED) {
1635                 log_rdma_event(ERR, "rdma_connect failed port=%d\n", port);
1636                 goto rdma_connect_failed;
1637         }
1638
1639         log_rdma_event(INFO, "rdma_connect connected\n");
1640
1641         rc = allocate_caches_and_workqueue(info);
1642         if (rc) {
1643                 log_rdma_event(ERR, "cache allocation failed\n");
1644                 goto allocate_cache_failed;
1645         }
1646
1647         init_waitqueue_head(&info->wait_send_queue);
1648         init_waitqueue_head(&info->wait_reassembly_queue);
1649
1650         INIT_DELAYED_WORK(&info->idle_timer_work, idle_connection_timer);
1651         INIT_DELAYED_WORK(&info->send_immediate_work, send_immediate_work);
1652         queue_delayed_work(info->workqueue, &info->idle_timer_work,
1653                 info->keep_alive_interval*HZ);
1654
1655         init_waitqueue_head(&info->wait_smbd_recv_pending);
1656         info->smbd_recv_pending = 0;
1657
1658         init_waitqueue_head(&info->wait_send_pending);
1659         atomic_set(&info->send_pending, 0);
1660
1661         init_waitqueue_head(&info->wait_send_payload_pending);
1662         atomic_set(&info->send_payload_pending, 0);
1663
1664         INIT_WORK(&info->disconnect_work, smbd_disconnect_rdma_work);
1665         INIT_WORK(&info->destroy_work, smbd_destroy_rdma_work);
1666         INIT_WORK(&info->recv_done_work, smbd_recv_done_work);
1667         INIT_WORK(&info->post_send_credits_work, smbd_post_send_credits);
1668         info->new_credits_offered = 0;
1669         spin_lock_init(&info->lock_new_credits_offered);
1670
1671         rc = smbd_negotiate(info);
1672         if (rc) {
1673                 log_rdma_event(ERR, "smbd_negotiate rc=%d\n", rc);
1674                 goto negotiation_failed;
1675         }
1676
1677         return info;
1678
1679 negotiation_failed:
1680         cancel_delayed_work_sync(&info->idle_timer_work);
1681         destroy_caches_and_workqueue(info);
1682         info->transport_status = SMBD_NEGOTIATE_FAILED;
1683         init_waitqueue_head(&info->conn_wait);
1684         rdma_disconnect(info->id);
1685         wait_event(info->conn_wait,
1686                 info->transport_status == SMBD_DISCONNECTED);
1687
1688 allocate_cache_failed:
1689 rdma_connect_failed:
1690         rdma_destroy_qp(info->id);
1691
1692 create_qp_failed:
1693 alloc_cq_failed:
1694         if (info->send_cq)
1695                 ib_free_cq(info->send_cq);
1696         if (info->recv_cq)
1697                 ib_free_cq(info->recv_cq);
1698
1699 config_failed:
1700         ib_dealloc_pd(info->pd);
1701         rdma_destroy_id(info->id);
1702
1703 create_id_failed:
1704         kfree(info);
1705         return NULL;
1706 }
1707
1708 struct smbd_connection *smbd_get_connection(
1709         struct TCP_Server_Info *server, struct sockaddr *dstaddr)
1710 {
1711         struct smbd_connection *ret;
1712         int port = SMBD_PORT;
1713
1714 try_again:
1715         ret = _smbd_get_connection(server, dstaddr, port);
1716
1717         /* Try SMB_PORT if SMBD_PORT doesn't work */
1718         if (!ret && port == SMBD_PORT) {
1719                 port = SMB_PORT;
1720                 goto try_again;
1721         }
1722         return ret;
1723 }
1724
1725 /*
1726  * Receive data from receive reassembly queue
1727  * All the incoming data packets are placed in reassembly queue
1728  * buf: the buffer to read data into
1729  * size: the length of data to read
1730  * return value: actual data read
1731  * Note: this implementation copies the data from reassebmly queue to receive
1732  * buffers used by upper layer. This is not the optimal code path. A better way
1733  * to do it is to not have upper layer allocate its receive buffers but rather
1734  * borrow the buffer from reassembly queue, and return it after data is
1735  * consumed. But this will require more changes to upper layer code, and also
1736  * need to consider packet boundaries while they still being reassembled.
1737  */
1738 int smbd_recv_buf(struct smbd_connection *info, char *buf, unsigned int size)
1739 {
1740         struct smbd_response *response;
1741         struct smbd_data_transfer *data_transfer;
1742         int to_copy, to_read, data_read, offset;
1743         u32 data_length, remaining_data_length, data_offset;
1744         int rc;
1745         unsigned long flags;
1746
1747 again:
1748         if (info->transport_status != SMBD_CONNECTED) {
1749                 log_read(ERR, "disconnected\n");
1750                 return -ENODEV;
1751         }
1752
1753         /*
1754          * No need to hold the reassembly queue lock all the time as we are
1755          * the only one reading from the front of the queue. The transport
1756          * may add more entries to the back of the queue at the same time
1757          */
1758         log_read(INFO, "size=%d info->reassembly_data_length=%d\n", size,
1759                 info->reassembly_data_length);
1760         if (info->reassembly_data_length >= size) {
1761                 int queue_length;
1762                 int queue_removed = 0;
1763
1764                 /*
1765                  * Need to make sure reassembly_data_length is read before
1766                  * reading reassembly_queue_length and calling
1767                  * _get_first_reassembly. This call is lock free
1768                  * as we never read at the end of the queue which are being
1769                  * updated in SOFTIRQ as more data is received
1770                  */
1771                 virt_rmb();
1772                 queue_length = info->reassembly_queue_length;
1773                 data_read = 0;
1774                 to_read = size;
1775                 offset = info->first_entry_offset;
1776                 while (data_read < size) {
1777                         response = _get_first_reassembly(info);
1778                         data_transfer = smbd_response_payload(response);
1779                         data_length = le32_to_cpu(data_transfer->data_length);
1780                         remaining_data_length =
1781                                 le32_to_cpu(
1782                                         data_transfer->remaining_data_length);
1783                         data_offset = le32_to_cpu(data_transfer->data_offset);
1784
1785                         /*
1786                          * The upper layer expects RFC1002 length at the
1787                          * beginning of the payload. Return it to indicate
1788                          * the total length of the packet. This minimize the
1789                          * change to upper layer packet processing logic. This
1790                          * will be eventually remove when an intermediate
1791                          * transport layer is added
1792                          */
1793                         if (response->first_segment && size == 4) {
1794                                 unsigned int rfc1002_len =
1795                                         data_length + remaining_data_length;
1796                                 *((__be32 *)buf) = cpu_to_be32(rfc1002_len);
1797                                 data_read = 4;
1798                                 response->first_segment = false;
1799                                 log_read(INFO, "returning rfc1002 length %d\n",
1800                                         rfc1002_len);
1801                                 goto read_rfc1002_done;
1802                         }
1803
1804                         to_copy = min_t(int, data_length - offset, to_read);
1805                         memcpy(
1806                                 buf + data_read,
1807                                 (char *)data_transfer + data_offset + offset,
1808                                 to_copy);
1809
1810                         /* move on to the next buffer? */
1811                         if (to_copy == data_length - offset) {
1812                                 queue_length--;
1813                                 /*
1814                                  * No need to lock if we are not at the
1815                                  * end of the queue
1816                                  */
1817                                 if (!queue_length)
1818                                         spin_lock_irqsave(
1819                                                 &info->reassembly_queue_lock,
1820                                                 flags);
1821                                 list_del(&response->list);
1822                                 queue_removed++;
1823                                 if (!queue_length)
1824                                         spin_unlock_irqrestore(
1825                                                 &info->reassembly_queue_lock,
1826                                                 flags);
1827
1828                                 info->count_reassembly_queue--;
1829                                 info->count_dequeue_reassembly_queue++;
1830                                 put_receive_buffer(info, response);
1831                                 offset = 0;
1832                                 log_read(INFO, "put_receive_buffer offset=0\n");
1833                         } else
1834                                 offset += to_copy;
1835
1836                         to_read -= to_copy;
1837                         data_read += to_copy;
1838
1839                         log_read(INFO, "_get_first_reassembly memcpy %d bytes "
1840                                 "data_transfer_length-offset=%d after that "
1841                                 "to_read=%d data_read=%d offset=%d\n",
1842                                 to_copy, data_length - offset,
1843                                 to_read, data_read, offset);
1844                 }
1845
1846                 spin_lock_irqsave(&info->reassembly_queue_lock, flags);
1847                 info->reassembly_data_length -= data_read;
1848                 info->reassembly_queue_length -= queue_removed;
1849                 spin_unlock_irqrestore(&info->reassembly_queue_lock, flags);
1850
1851                 info->first_entry_offset = offset;
1852                 log_read(INFO, "returning to thread data_read=%d "
1853                         "reassembly_data_length=%d first_entry_offset=%d\n",
1854                         data_read, info->reassembly_data_length,
1855                         info->first_entry_offset);
1856 read_rfc1002_done:
1857                 return data_read;
1858         }
1859
1860         log_read(INFO, "wait_event on more data\n");
1861         rc = wait_event_interruptible(
1862                 info->wait_reassembly_queue,
1863                 info->reassembly_data_length >= size ||
1864                         info->transport_status != SMBD_CONNECTED);
1865         /* Don't return any data if interrupted */
1866         if (rc)
1867                 return -ENODEV;
1868
1869         goto again;
1870 }
1871
1872 /*
1873  * Receive a page from receive reassembly queue
1874  * page: the page to read data into
1875  * to_read: the length of data to read
1876  * return value: actual data read
1877  */
1878 int smbd_recv_page(struct smbd_connection *info,
1879                 struct page *page, unsigned int to_read)
1880 {
1881         int ret;
1882         char *to_address;
1883
1884         /* make sure we have the page ready for read */
1885         ret = wait_event_interruptible(
1886                 info->wait_reassembly_queue,
1887                 info->reassembly_data_length >= to_read ||
1888                         info->transport_status != SMBD_CONNECTED);
1889         if (ret)
1890                 return 0;
1891
1892         /* now we can read from reassembly queue and not sleep */
1893         to_address = kmap_atomic(page);
1894
1895         log_read(INFO, "reading from page=%p address=%p to_read=%d\n",
1896                 page, to_address, to_read);
1897
1898         ret = smbd_recv_buf(info, to_address, to_read);
1899         kunmap_atomic(to_address);
1900
1901         return ret;
1902 }
1903
1904 /*
1905  * Receive data from transport
1906  * msg: a msghdr point to the buffer, can be ITER_KVEC or ITER_BVEC
1907  * return: total bytes read, or 0. SMB Direct will not do partial read.
1908  */
1909 int smbd_recv(struct smbd_connection *info, struct msghdr *msg)
1910 {
1911         char *buf;
1912         struct page *page;
1913         unsigned int to_read;
1914         int rc;
1915
1916         info->smbd_recv_pending++;
1917
1918         switch (msg->msg_iter.type) {
1919         case READ | ITER_KVEC:
1920                 buf = msg->msg_iter.kvec->iov_base;
1921                 to_read = msg->msg_iter.kvec->iov_len;
1922                 rc = smbd_recv_buf(info, buf, to_read);
1923                 break;
1924
1925         case READ | ITER_BVEC:
1926                 page = msg->msg_iter.bvec->bv_page;
1927                 to_read = msg->msg_iter.bvec->bv_len;
1928                 rc = smbd_recv_page(info, page, to_read);
1929                 break;
1930
1931         default:
1932                 /* It's a bug in upper layer to get there */
1933                 cifs_dbg(VFS, "CIFS: invalid msg type %d\n",
1934                         msg->msg_iter.type);
1935                 rc = -EIO;
1936         }
1937
1938         info->smbd_recv_pending--;
1939         wake_up(&info->wait_smbd_recv_pending);
1940
1941         /* SMBDirect will read it all or nothing */
1942         if (rc > 0)
1943                 msg->msg_iter.count = 0;
1944         return rc;
1945 }