// SPDX-License-Identifier: GPL-2.0-or-later
/*
 *   Copyright (C) 2017, Microsoft Corporation.
 *
 *   Author(s): Long Li <longli@microsoft.com>
 */
#include <linux/module.h>
#include <linux/highmem.h>
#include <linux/folio_queue.h>
#define __SMBDIRECT_SOCKET_DISCONNECT(__sc) smbd_disconnect_rdma_connection(__sc)
#include "../common/smbdirect/smbdirect_pdu.h"
#include "smbdirect.h"
#include "cifs_debug.h"
#include "cifsproto.h"
#include "smb2proto.h"

const struct smbdirect_socket_parameters *smbd_get_parameters(struct smbd_connection *conn)
{
        struct smbdirect_socket *sc = &conn->socket;

        return &sc->parameters;
}

static struct smbdirect_recv_io *get_receive_buffer(
                struct smbdirect_socket *sc);
static void put_receive_buffer(
                struct smbdirect_socket *sc,
                struct smbdirect_recv_io *response);
static int allocate_receive_buffers(struct smbdirect_socket *sc, int num_buf);
static void destroy_receive_buffers(struct smbdirect_socket *sc);

static void enqueue_reassembly(
                struct smbdirect_socket *sc,
                struct smbdirect_recv_io *response, int data_length);
static struct smbdirect_recv_io *_get_first_reassembly(
                struct smbdirect_socket *sc);

static int smbd_post_send(struct smbdirect_socket *sc,
                          struct smbdirect_send_batch *batch,
                          struct smbdirect_send_io *request);

static int smbd_post_recv(
                struct smbdirect_socket *sc,
                struct smbdirect_recv_io *response);

static int smbd_post_send_empty(struct smbdirect_socket *sc);

static void destroy_mr_list(struct smbdirect_socket *sc);
static int allocate_mr_list(struct smbdirect_socket *sc);

struct smb_extract_to_rdma {
        struct ib_sge           *sge;
        unsigned int            nr_sge;
        unsigned int            max_sge;
        struct ib_device        *device;
        u32                     local_dma_lkey;
        enum dma_data_direction direction;
};
static ssize_t smb_extract_iter_to_rdma(struct iov_iter *iter, size_t len,
                                        struct smb_extract_to_rdma *rdma);

/* Port numbers for SMBD transport */
#define SMB_PORT        445
#define SMBD_PORT       5445

/* Address lookup and resolve timeout in ms */
#define RDMA_RESOLVE_TIMEOUT    5000

/* SMBD negotiation timeout in seconds */
#define SMBD_NEGOTIATE_TIMEOUT  120

/* The timeout to wait for a keepalive message from peer in seconds */
#define KEEPALIVE_RECV_TIMEOUT 5

/* SMBD minimum receive size and fragmented sized defined in [MS-SMBD] */
#define SMBD_MIN_RECEIVE_SIZE           128
#define SMBD_MIN_FRAGMENTED_SIZE        131072

/*
 * Default maximum number of RDMA read/write outstanding on this connection
 * This value is possibly decreased during QP creation on hardware limit
 */
#define SMBD_CM_RESPONDER_RESOURCES     32

/* Maximum number of retries on data transfer operations */
#define SMBD_CM_RETRY                   6
/* No need to retry on Receiver Not Ready since SMBD manages credits */
#define SMBD_CM_RNR_RETRY               0

/*
 * User configurable initial values per SMBD transport connection
 * as defined in [MS-SMBD] 3.1.1.1
 * Those may change after a SMBD negotiation
 */
/* The local peer's maximum number of credits to grant to the peer */
int smbd_receive_credit_max = 255;

/* The remote peer's credit request of local peer */
int smbd_send_credit_target = 255;

/* The maximum single message size can be sent to remote peer */
int smbd_max_send_size = 1364;

/*
 * The maximum fragmented upper-layer payload receive size supported
 *
 * Assume max_payload_per_credit is
 * smbd_max_receive_size - 24 = 1340
 *
 * The maximum number would be
 * smbd_receive_credit_max * max_payload_per_credit
 *
 *                       1340 * 255 = 341700 (0x536C4)
 *
 * The minimum value from the spec is 131072 (0x20000)
 *
 * For now we use the logic we used in ksmbd before:
 *                 (1364 * 255) / 2 = 173910 (0x2A756)
 */
int smbd_max_fragmented_recv_size = (1364 * 255) / 2;

/*  The maximum single-message size which can be received */
int smbd_max_receive_size = 1364;

/* The timeout to initiate send of a keepalive message on idle */
int smbd_keep_alive_interval = 120;

/*
 * User configurable initial values for RDMA transport
 * The actual values used may be lower and are limited to hardware capabilities
 */
/* Default maximum number of pages in a single RDMA write/read */
int smbd_max_frmr_depth = 2048;

/* If payload is less than this byte, use RDMA send/recv not read/write */
int rdma_readwrite_threshold = 4096;

/* Transport logging functions
 * Logging are defined as classes. They can be OR'ed to define the actual
 * logging level via module parameter smbd_logging_class
 * e.g. cifs.smbd_logging_class=0xa0 will log all log_rdma_recv() and
 * log_rdma_event()
 */
#define LOG_OUTGOING                    0x1
#define LOG_INCOMING                    0x2
#define LOG_READ                        0x4
#define LOG_WRITE                       0x8
#define LOG_RDMA_SEND                   0x10
#define LOG_RDMA_RECV                   0x20
#define LOG_KEEP_ALIVE                  0x40
#define LOG_RDMA_EVENT                  0x80
#define LOG_RDMA_MR                     0x100
static unsigned int smbd_logging_class;
module_param(smbd_logging_class, uint, 0644);
MODULE_PARM_DESC(smbd_logging_class,
        "Logging class for SMBD transport 0x0 to 0x100");

#define ERR             0x0
#define INFO            0x1
static unsigned int smbd_logging_level = ERR;
module_param(smbd_logging_level, uint, 0644);
MODULE_PARM_DESC(smbd_logging_level,
        "Logging level for SMBD transport, 0 (default): error, 1: info");

#define log_rdma(level, class, fmt, args...)                            \
do {                                                                    \
        if (level <= smbd_logging_level || class & smbd_logging_class)  \
                cifs_dbg(VFS, "%s:%d " fmt, __func__, __LINE__, ##args);\
} while (0)

#define log_outgoing(level, fmt, args...) \
                log_rdma(level, LOG_OUTGOING, fmt, ##args)
#define log_incoming(level, fmt, args...) \
                log_rdma(level, LOG_INCOMING, fmt, ##args)
#define log_read(level, fmt, args...)   log_rdma(level, LOG_READ, fmt, ##args)
#define log_write(level, fmt, args...)  log_rdma(level, LOG_WRITE, fmt, ##args)
#define log_rdma_send(level, fmt, args...) \
                log_rdma(level, LOG_RDMA_SEND, fmt, ##args)
#define log_rdma_recv(level, fmt, args...) \
                log_rdma(level, LOG_RDMA_RECV, fmt, ##args)
#define log_keep_alive(level, fmt, args...) \
                log_rdma(level, LOG_KEEP_ALIVE, fmt, ##args)
#define log_rdma_event(level, fmt, args...) \
                log_rdma(level, LOG_RDMA_EVENT, fmt, ##args)
#define log_rdma_mr(level, fmt, args...) \
                log_rdma(level, LOG_RDMA_MR, fmt, ##args)

static void smbd_disconnect_wake_up_all(struct smbdirect_socket *sc)
{
        /*
         * Wake up all waiters in all wait queues
         * in order to notice the broken connection.
         */
        wake_up_all(&sc->status_wait);
        wake_up_all(&sc->send_io.lcredits.wait_queue);
        wake_up_all(&sc->send_io.credits.wait_queue);
        wake_up_all(&sc->send_io.pending.dec_wait_queue);
        wake_up_all(&sc->send_io.pending.zero_wait_queue);
        wake_up_all(&sc->recv_io.reassembly.wait_queue);
        wake_up_all(&sc->mr_io.ready.wait_queue);
        wake_up_all(&sc->mr_io.cleanup.wait_queue);
}

static void smbd_disconnect_rdma_work(struct work_struct *work)
{
        struct smbdirect_socket *sc =
                container_of(work, struct smbdirect_socket, disconnect_work);

        if (sc->first_error == 0)
                sc->first_error = -ECONNABORTED;

        /*
         * make sure this and other work is not queued again
         * but here we don't block and avoid
         * disable[_delayed]_work_sync()
         */
        disable_work(&sc->disconnect_work);
        disable_work(&sc->recv_io.posted.refill_work);
        disable_work(&sc->mr_io.recovery_work);
        disable_work(&sc->idle.immediate_work);
        disable_delayed_work(&sc->idle.timer_work);

        switch (sc->status) {
        case SMBDIRECT_SOCKET_NEGOTIATE_NEEDED:
        case SMBDIRECT_SOCKET_NEGOTIATE_RUNNING:
        case SMBDIRECT_SOCKET_NEGOTIATE_FAILED:
        case SMBDIRECT_SOCKET_CONNECTED:
        case SMBDIRECT_SOCKET_ERROR:
                sc->status = SMBDIRECT_SOCKET_DISCONNECTING;
                rdma_disconnect(sc->rdma.cm_id);
                break;

        case SMBDIRECT_SOCKET_CREATED:
        case SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED:
        case SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING:
        case SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED:
        case SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED:
        case SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING:
        case SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED:
        case SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED:
        case SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING:
        case SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED:
                /*
                 * rdma_connect() never reached
                 * RDMA_CM_EVENT_ESTABLISHED
                 */
                sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
                break;

        case SMBDIRECT_SOCKET_DISCONNECTING:
        case SMBDIRECT_SOCKET_DISCONNECTED:
        case SMBDIRECT_SOCKET_DESTROYED:
                break;
        }

        /*
         * Wake up all waiters in all wait queues
         * in order to notice the broken connection.
         */
        smbd_disconnect_wake_up_all(sc);
}

static void smbd_disconnect_rdma_connection(struct smbdirect_socket *sc)
{
        if (sc->first_error == 0)
                sc->first_error = -ECONNABORTED;

        /*
         * make sure other work (than disconnect_work) is
         * not queued again but here we don't block and avoid
         * disable[_delayed]_work_sync()
         */
        disable_work(&sc->recv_io.posted.refill_work);
        disable_work(&sc->mr_io.recovery_work);
        disable_work(&sc->idle.immediate_work);
        disable_delayed_work(&sc->idle.timer_work);

        switch (sc->status) {
        case SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED:
        case SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED:
        case SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED:
        case SMBDIRECT_SOCKET_NEGOTIATE_FAILED:
        case SMBDIRECT_SOCKET_ERROR:
        case SMBDIRECT_SOCKET_DISCONNECTING:
        case SMBDIRECT_SOCKET_DISCONNECTED:
        case SMBDIRECT_SOCKET_DESTROYED:
                /*
                 * Keep the current error status
                 */
                break;

        case SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED:
        case SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING:
                sc->status = SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED;
                break;

        case SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED:
        case SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING:
                sc->status = SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED;
                break;

        case SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED:
        case SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING:
                sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED;
                break;

        case SMBDIRECT_SOCKET_NEGOTIATE_NEEDED:
        case SMBDIRECT_SOCKET_NEGOTIATE_RUNNING:
                sc->status = SMBDIRECT_SOCKET_NEGOTIATE_FAILED;
                break;

        case SMBDIRECT_SOCKET_CREATED:
                sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
                break;

        case SMBDIRECT_SOCKET_CONNECTED:
                sc->status = SMBDIRECT_SOCKET_ERROR;
                break;
        }

        /*
         * Wake up all waiters in all wait queues
         * in order to notice the broken connection.
         */
        smbd_disconnect_wake_up_all(sc);

        queue_work(sc->workqueue, &sc->disconnect_work);
}

/* Upcall from RDMA CM */
static int smbd_conn_upcall(
                struct rdma_cm_id *id, struct rdma_cm_event *event)
{
        struct smbdirect_socket *sc = id->context;
        struct smbdirect_socket_parameters *sp = &sc->parameters;
        const char *event_name = rdma_event_msg(event->event);
        u8 peer_initiator_depth;
        u8 peer_responder_resources;

        log_rdma_event(INFO, "event=%s status=%d\n",
                event_name, event->status);

        switch (event->event) {
        case RDMA_CM_EVENT_ADDR_RESOLVED:
                if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING))
                        break;
                sc->status = SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED;
                wake_up(&sc->status_wait);
                break;

        case RDMA_CM_EVENT_ROUTE_RESOLVED:
                if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING))
                        break;
                sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED;
                wake_up(&sc->status_wait);
                break;

        case RDMA_CM_EVENT_ADDR_ERROR:
                log_rdma_event(ERR, "connecting failed event=%s\n", event_name);
                sc->status = SMBDIRECT_SOCKET_RESOLVE_ADDR_FAILED;
                smbd_disconnect_rdma_work(&sc->disconnect_work);
                break;

        case RDMA_CM_EVENT_ROUTE_ERROR:
                log_rdma_event(ERR, "connecting failed event=%s\n", event_name);
                sc->status = SMBDIRECT_SOCKET_RESOLVE_ROUTE_FAILED;
                smbd_disconnect_rdma_work(&sc->disconnect_work);
                break;

        case RDMA_CM_EVENT_ESTABLISHED:
                log_rdma_event(INFO, "connected event=%s\n", event_name);

                /*
                 * Here we work around an inconsistency between
                 * iWarp and other devices (at least rxe and irdma using RoCEv2)
                 */
                if (rdma_protocol_iwarp(id->device, id->port_num)) {
                        /*
                         * iWarp devices report the peer's values
                         * with the perspective of the peer here.
                         * Tested with siw and irdma (in iwarp mode)
                         * We need to change to our perspective here,
                         * so we need to switch the values.
                         */
                        peer_initiator_depth = event->param.conn.responder_resources;
                        peer_responder_resources = event->param.conn.initiator_depth;
                } else {
                        /*
                         * Non iWarp devices report the peer's values
                         * already changed to our perspective here.
                         * Tested with rxe and irdma (in roce mode).
                         */
                        peer_initiator_depth = event->param.conn.initiator_depth;
                        peer_responder_resources = event->param.conn.responder_resources;
                }
                if (rdma_protocol_iwarp(id->device, id->port_num) &&
                    event->param.conn.private_data_len == 8) {
                        /*
                         * Legacy clients with only iWarp MPA v1 support
                         * need a private blob in order to negotiate
                         * the IRD/ORD values.
                         */
                        const __be32 *ird_ord_hdr = event->param.conn.private_data;
                        u32 ird32 = be32_to_cpu(ird_ord_hdr[0]);
                        u32 ord32 = be32_to_cpu(ird_ord_hdr[1]);

                        /*
                         * cifs.ko sends the legacy IRD/ORD negotiation
                         * event if iWarp MPA v2 was used.
                         *
                         * Here we check that the values match and only
                         * mark the client as legacy if they don't match.
                         */
                        if ((u32)event->param.conn.initiator_depth != ird32 ||
                            (u32)event->param.conn.responder_resources != ord32) {
                                /*
                                 * There are broken clients (old cifs.ko)
                                 * using little endian and also
                                 * struct rdma_conn_param only uses u8
                                 * for initiator_depth and responder_resources,
                                 * so we truncate the value to U8_MAX.
                                 *
                                 * smb_direct_accept_client() will then
                                 * do the real negotiation in order to
                                 * select the minimum between client and
                                 * server.
                                 */
                                ird32 = min_t(u32, ird32, U8_MAX);
                                ord32 = min_t(u32, ord32, U8_MAX);

                                sc->rdma.legacy_iwarp = true;
                                peer_initiator_depth = (u8)ird32;
                                peer_responder_resources = (u8)ord32;
                        }
                }

                /*
                 * negotiate the value by using the minimum
                 * between client and server if the client provided
                 * non 0 values.
                 */
                if (peer_initiator_depth != 0)
                        sp->initiator_depth =
                                        min_t(u8, sp->initiator_depth,
                                              peer_initiator_depth);
                if (peer_responder_resources != 0)
                        sp->responder_resources =
                                        min_t(u8, sp->responder_resources,
                                              peer_responder_resources);

                if (SMBDIRECT_CHECK_STATUS_DISCONNECT(sc, SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING))
                        break;
                sc->status = SMBDIRECT_SOCKET_NEGOTIATE_NEEDED;
                wake_up(&sc->status_wait);
                break;

        case RDMA_CM_EVENT_CONNECT_ERROR:
        case RDMA_CM_EVENT_UNREACHABLE:
        case RDMA_CM_EVENT_REJECTED:
                log_rdma_event(ERR, "connecting failed event=%s\n", event_name);
                sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_FAILED;
                smbd_disconnect_rdma_work(&sc->disconnect_work);
                break;

        case RDMA_CM_EVENT_DEVICE_REMOVAL:
        case RDMA_CM_EVENT_DISCONNECTED:
                /* This happens when we fail the negotiation */
                if (sc->status == SMBDIRECT_SOCKET_NEGOTIATE_FAILED) {
                        log_rdma_event(ERR, "event=%s during negotiation\n", event_name);
                }

                sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
                smbd_disconnect_rdma_work(&sc->disconnect_work);
                break;

        default:
                log_rdma_event(ERR, "unexpected event=%s status=%d\n",
                               event_name, event->status);
                break;
        }

        return 0;
}

/* Upcall from RDMA QP */
static void
smbd_qp_async_error_upcall(struct ib_event *event, void *context)
{
        struct smbdirect_socket *sc = context;

        log_rdma_event(ERR, "%s on device %s socket %p\n",
                ib_event_msg(event->event), event->device->name, sc);

        switch (event->event) {
        case IB_EVENT_CQ_ERR:
        case IB_EVENT_QP_FATAL:
                smbd_disconnect_rdma_connection(sc);
                break;

        default:
                break;
        }
}

static inline void *smbdirect_send_io_payload(struct smbdirect_send_io *request)
{
        return (void *)request->packet;
}

static inline void *smbdirect_recv_io_payload(struct smbdirect_recv_io *response)
{
        return (void *)response->packet;
}

static struct smbdirect_send_io *smbd_alloc_send_io(struct smbdirect_socket *sc)
{
        struct smbdirect_send_io *msg;

        msg = mempool_alloc(sc->send_io.mem.pool, GFP_KERNEL);
        if (!msg)
                return ERR_PTR(-ENOMEM);
        msg->socket = sc;
        INIT_LIST_HEAD(&msg->sibling_list);
        msg->num_sge = 0;

        return msg;
}

static void smbd_free_send_io(struct smbdirect_send_io *msg)
{
        struct smbdirect_socket *sc = msg->socket;
        size_t i;

        /*
         * The list needs to be empty!
         * The caller should take care of it.
         */
        WARN_ON_ONCE(!list_empty(&msg->sibling_list));

        /*
         * Note we call ib_dma_unmap_page(), even if some sges are mapped using
         * ib_dma_map_single().
         *
         * The difference between _single() and _page() only matters for the
         * ib_dma_map_*() case.
         *
         * For the ib_dma_unmap_*() case it does not matter as both take the
         * dma_addr_t and dma_unmap_single_attrs() is just an alias to
         * dma_unmap_page_attrs().
         */
        for (i = 0; i < msg->num_sge; i++)
                ib_dma_unmap_page(sc->ib.dev,
                                  msg->sge[i].addr,
                                  msg->sge[i].length,
                                  DMA_TO_DEVICE);

        mempool_free(msg, sc->send_io.mem.pool);
}

/* Called when a RDMA send is done */
static void send_done(struct ib_cq *cq, struct ib_wc *wc)
{
        struct smbdirect_send_io *request =
                container_of(wc->wr_cqe, struct smbdirect_send_io, cqe);
        struct smbdirect_socket *sc = request->socket;
        struct smbdirect_send_io *sibling, *next;
        int lcredits = 0;

        log_rdma_send(INFO, "smbdirect_send_io 0x%p completed wc->status=%s\n",
                request, ib_wc_status_msg(wc->status));

        if (unlikely(!(request->wr.send_flags & IB_SEND_SIGNALED))) {
                /*
                 * This happens when smbdirect_send_io is a sibling
                 * before the final message, it is signaled on
                 * error anyway, so we need to skip
                 * smbdirect_connection_free_send_io here,
                 * otherwise is will destroy the memory
                 * of the siblings too, which will cause
                 * use after free problems for the others
                 * triggered from ib_drain_qp().
                 */
                if (wc->status != IB_WC_SUCCESS)
                        goto skip_free;

                /*
                 * This should not happen!
                 * But we better just close the
                 * connection...
                 */
                log_rdma_send(ERR,
                        "unexpected send completion wc->status=%s (%d) wc->opcode=%d\n",
                        ib_wc_status_msg(wc->status), wc->status, wc->opcode);
                smbd_disconnect_rdma_connection(sc);
                return;
        }

        /*
         * Free possible siblings and then the main send_io
         */
        list_for_each_entry_safe(sibling, next, &request->sibling_list, sibling_list) {
                list_del_init(&sibling->sibling_list);
                smbd_free_send_io(sibling);
                lcredits += 1;
        }
        /* Note this frees wc->wr_cqe, but not wc */
        smbd_free_send_io(request);
        lcredits += 1;

        if (wc->status != IB_WC_SUCCESS || wc->opcode != IB_WC_SEND) {
skip_free:
                if (wc->status != IB_WC_WR_FLUSH_ERR)
                        log_rdma_send(ERR, "wc->status=%s wc->opcode=%d\n",
                                ib_wc_status_msg(wc->status), wc->opcode);
                smbd_disconnect_rdma_connection(sc);
                return;
        }

        atomic_add(lcredits, &sc->send_io.lcredits.count);
        wake_up(&sc->send_io.lcredits.wait_queue);

        if (atomic_dec_and_test(&sc->send_io.pending.count))
                wake_up(&sc->send_io.pending.zero_wait_queue);

        wake_up(&sc->send_io.pending.dec_wait_queue);
}

static void dump_smbdirect_negotiate_resp(struct smbdirect_negotiate_resp *resp)
{
        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",
                       resp->min_version, resp->max_version,
                       resp->negotiated_version, resp->credits_requested,
                       resp->credits_granted, resp->status,
                       resp->max_readwrite_size, resp->preferred_send_size,
                       resp->max_receive_size, resp->max_fragmented_size);
}

/*
 * Process a negotiation response message, according to [MS-SMBD]3.1.5.7
 * response, packet_length: the negotiation response message
 * return value: true if negotiation is a success, false if failed
 */
static bool process_negotiation_response(
                struct smbdirect_recv_io *response, int packet_length)
{
        struct smbdirect_socket *sc = response->socket;
        struct smbdirect_socket_parameters *sp = &sc->parameters;
        struct smbdirect_negotiate_resp *packet = smbdirect_recv_io_payload(response);

        if (packet_length < sizeof(struct smbdirect_negotiate_resp)) {
                log_rdma_event(ERR,
                        "error: packet_length=%d\n", packet_length);
                return false;
        }

        if (le16_to_cpu(packet->negotiated_version) != SMBDIRECT_V1) {
                log_rdma_event(ERR, "error: negotiated_version=%x\n",
                        le16_to_cpu(packet->negotiated_version));
                return false;
        }

        if (packet->credits_requested == 0) {
                log_rdma_event(ERR, "error: credits_requested==0\n");
                return false;
        }
        sc->recv_io.credits.target = le16_to_cpu(packet->credits_requested);
        sc->recv_io.credits.target = min_t(u16, sc->recv_io.credits.target, sp->recv_credit_max);

        if (packet->credits_granted == 0) {
                log_rdma_event(ERR, "error: credits_granted==0\n");
                return false;
        }
        atomic_set(&sc->send_io.lcredits.count, sp->send_credit_target);
        atomic_set(&sc->send_io.credits.count, le16_to_cpu(packet->credits_granted));

        if (le32_to_cpu(packet->preferred_send_size) > sp->max_recv_size) {
                log_rdma_event(ERR, "error: preferred_send_size=%d\n",
                        le32_to_cpu(packet->preferred_send_size));
                return false;
        }
        sp->max_recv_size = le32_to_cpu(packet->preferred_send_size);

        if (le32_to_cpu(packet->max_receive_size) < SMBD_MIN_RECEIVE_SIZE) {
                log_rdma_event(ERR, "error: max_receive_size=%d\n",
                        le32_to_cpu(packet->max_receive_size));
                return false;
        }
        sp->max_send_size = min_t(u32, sp->max_send_size,
                                  le32_to_cpu(packet->max_receive_size));

        if (le32_to_cpu(packet->max_fragmented_size) <
                        SMBD_MIN_FRAGMENTED_SIZE) {
                log_rdma_event(ERR, "error: max_fragmented_size=%d\n",
                        le32_to_cpu(packet->max_fragmented_size));
                return false;
        }
        sp->max_fragmented_send_size =
                le32_to_cpu(packet->max_fragmented_size);


        sp->max_read_write_size = min_t(u32,
                        le32_to_cpu(packet->max_readwrite_size),
                        sp->max_frmr_depth * PAGE_SIZE);
        sp->max_frmr_depth = sp->max_read_write_size / PAGE_SIZE;

        atomic_set(&sc->send_io.bcredits.count, 1);
        sc->recv_io.expected = SMBDIRECT_EXPECT_DATA_TRANSFER;
        return true;
}

static void smbd_post_send_credits(struct work_struct *work)
{
        int rc;
        struct smbdirect_recv_io *response;
        struct smbdirect_socket *sc =
                container_of(work, struct smbdirect_socket, recv_io.posted.refill_work);
        int posted = 0;

        if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
                return;
        }

        if (sc->recv_io.credits.target >
                atomic_read(&sc->recv_io.credits.count)) {
                while (true) {
                        response = get_receive_buffer(sc);
                        if (!response)
                                break;

                        response->first_segment = false;
                        rc = smbd_post_recv(sc, response);
                        if (rc) {
                                log_rdma_recv(ERR,
                                        "post_recv failed rc=%d\n", rc);
                                put_receive_buffer(sc, response);
                                break;
                        }

                        atomic_inc(&sc->recv_io.posted.count);
                        posted += 1;
                }
        }

        atomic_add(posted, &sc->recv_io.credits.available);

        /*
         * If the last send credit is waiting for credits
         * it can grant we need to wake it up
         */
        if (posted &&
            atomic_read(&sc->send_io.bcredits.count) == 0 &&
            atomic_read(&sc->send_io.credits.count) == 0)
                wake_up(&sc->send_io.credits.wait_queue);

        /* Promptly send an immediate packet as defined in [MS-SMBD] 3.1.1.1 */
        if (atomic_read(&sc->recv_io.credits.count) <
                sc->recv_io.credits.target - 1) {
                log_keep_alive(INFO, "schedule send of an empty message\n");
                queue_work(sc->workqueue, &sc->idle.immediate_work);
        }
}

/* Called from softirq, when recv is done */
static void recv_done(struct ib_cq *cq, struct ib_wc *wc)
{
        struct smbdirect_data_transfer *data_transfer;
        struct smbdirect_recv_io *response =
                container_of(wc->wr_cqe, struct smbdirect_recv_io, cqe);
        struct smbdirect_socket *sc = response->socket;
        struct smbdirect_socket_parameters *sp = &sc->parameters;
        int current_recv_credits;
        u16 old_recv_credit_target;
        u32 data_offset = 0;
        u32 data_length = 0;
        u32 remaining_data_length = 0;
        bool negotiate_done = false;

        log_rdma_recv(INFO,
                      "response=0x%p type=%d wc status=%s wc opcode %d byte_len=%d pkey_index=%u\n",
                      response, sc->recv_io.expected,
                      ib_wc_status_msg(wc->status), wc->opcode,
                      wc->byte_len, wc->pkey_index);

        if (wc->status != IB_WC_SUCCESS || wc->opcode != IB_WC_RECV) {
                if (wc->status != IB_WC_WR_FLUSH_ERR)
                        log_rdma_recv(ERR, "wc->status=%s opcode=%d\n",
                                ib_wc_status_msg(wc->status), wc->opcode);
                goto error;
        }

        ib_dma_sync_single_for_cpu(
                wc->qp->device,
                response->sge.addr,
                response->sge.length,
                DMA_FROM_DEVICE);

        /*
         * Reset timer to the keepalive interval in
         * order to trigger our next keepalive message.
         */
        sc->idle.keepalive = SMBDIRECT_KEEPALIVE_NONE;
        mod_delayed_work(sc->workqueue, &sc->idle.timer_work,
                         msecs_to_jiffies(sp->keepalive_interval_msec));

        switch (sc->recv_io.expected) {
        /* SMBD negotiation response */
        case SMBDIRECT_EXPECT_NEGOTIATE_REP:
                dump_smbdirect_negotiate_resp(smbdirect_recv_io_payload(response));
                sc->recv_io.reassembly.full_packet_received = true;
                negotiate_done =
                        process_negotiation_response(response, wc->byte_len);
                put_receive_buffer(sc, response);
                if (SMBDIRECT_CHECK_STATUS_WARN(sc, SMBDIRECT_SOCKET_NEGOTIATE_RUNNING))
                        negotiate_done = false;
                if (!negotiate_done) {
                        sc->status = SMBDIRECT_SOCKET_NEGOTIATE_FAILED;
                        smbd_disconnect_rdma_connection(sc);
                } else {
                        sc->status = SMBDIRECT_SOCKET_CONNECTED;
                        wake_up(&sc->status_wait);
                }

                return;

        /* SMBD data transfer packet */
        case SMBDIRECT_EXPECT_DATA_TRANSFER:
                data_transfer = smbdirect_recv_io_payload(response);

                if (wc->byte_len <
                    offsetof(struct smbdirect_data_transfer, padding))
                        goto error;

                remaining_data_length = le32_to_cpu(data_transfer->remaining_data_length);
                data_offset = le32_to_cpu(data_transfer->data_offset);
                data_length = le32_to_cpu(data_transfer->data_length);
                if (wc->byte_len < data_offset ||
                    (u64)wc->byte_len < (u64)data_offset + data_length)
                        goto error;

                if (remaining_data_length > sp->max_fragmented_recv_size ||
                    data_length > sp->max_fragmented_recv_size ||
                    (u64)remaining_data_length + (u64)data_length > (u64)sp->max_fragmented_recv_size)
                        goto error;

                if (data_length) {
                        if (sc->recv_io.reassembly.full_packet_received)
                                response->first_segment = true;

                        if (le32_to_cpu(data_transfer->remaining_data_length))
                                sc->recv_io.reassembly.full_packet_received = false;
                        else
                                sc->recv_io.reassembly.full_packet_received = true;
                }

                atomic_dec(&sc->recv_io.posted.count);
                current_recv_credits = atomic_dec_return(&sc->recv_io.credits.count);

                old_recv_credit_target = sc->recv_io.credits.target;
                sc->recv_io.credits.target =
                        le16_to_cpu(data_transfer->credits_requested);
                sc->recv_io.credits.target =
                        min_t(u16, sc->recv_io.credits.target, sp->recv_credit_max);
                sc->recv_io.credits.target =
                        max_t(u16, sc->recv_io.credits.target, 1);
                if (le16_to_cpu(data_transfer->credits_granted)) {
                        atomic_add(le16_to_cpu(data_transfer->credits_granted),
                                &sc->send_io.credits.count);
                        /*
                         * We have new send credits granted from remote peer
                         * If any sender is waiting for credits, unblock it
                         */
                        wake_up(&sc->send_io.credits.wait_queue);
                }

                log_incoming(INFO, "data flags %d data_offset %d data_length %d remaining_data_length %d\n",
                             le16_to_cpu(data_transfer->flags),
                             le32_to_cpu(data_transfer->data_offset),
                             le32_to_cpu(data_transfer->data_length),
                             le32_to_cpu(data_transfer->remaining_data_length));

                /* Send an immediate response right away if requested */
                if (le16_to_cpu(data_transfer->flags) &
                                SMBDIRECT_FLAG_RESPONSE_REQUESTED) {
                        log_keep_alive(INFO, "schedule send of immediate response\n");
                        queue_work(sc->workqueue, &sc->idle.immediate_work);
                }

                /*
                 * If this is a packet with data playload place the data in
                 * reassembly queue and wake up the reading thread
                 */
                if (data_length) {
                        if (current_recv_credits <= (sc->recv_io.credits.target / 4) ||
                            sc->recv_io.credits.target > old_recv_credit_target)
                                queue_work(sc->workqueue, &sc->recv_io.posted.refill_work);

                        enqueue_reassembly(sc, response, data_length);
                        wake_up(&sc->recv_io.reassembly.wait_queue);
                } else
                        put_receive_buffer(sc, response);

                return;

        case SMBDIRECT_EXPECT_NEGOTIATE_REQ:
                /* Only server... */
                break;
        }

        /*
         * This is an internal error!
         */
        log_rdma_recv(ERR, "unexpected response type=%d\n", sc->recv_io.expected);
        WARN_ON_ONCE(sc->recv_io.expected != SMBDIRECT_EXPECT_DATA_TRANSFER);
error:
        put_receive_buffer(sc, response);
        smbd_disconnect_rdma_connection(sc);
}

static struct rdma_cm_id *smbd_create_id(
                struct smbdirect_socket *sc,
                struct sockaddr *dstaddr, int port)
{
        struct smbdirect_socket_parameters *sp = &sc->parameters;
        struct rdma_cm_id *id;
        u8 node_type = RDMA_NODE_UNSPECIFIED;
        int rc;
        __be16 *sport;

        id = rdma_create_id(&init_net, smbd_conn_upcall, sc,
                RDMA_PS_TCP, IB_QPT_RC);
        if (IS_ERR(id)) {
                rc = PTR_ERR(id);
                log_rdma_event(ERR, "rdma_create_id() failed %i\n", rc);
                return id;
        }

        switch (port) {
        case SMBD_PORT:
                /*
                 * only allow iWarp devices
                 * for port 5445.
                 */
                node_type = RDMA_NODE_RNIC;
                break;
        case SMB_PORT:
                /*
                 * only allow InfiniBand, RoCEv1 or RoCEv2
                 * devices for port 445.
                 *
                 * (Basically don't allow iWarp devices)
                 */
                node_type = RDMA_NODE_IB_CA;
                break;
        }
        rc = rdma_restrict_node_type(id, node_type);
        if (rc) {
                log_rdma_event(ERR, "rdma_restrict_node_type(%u) failed %i\n",
                               node_type, rc);
                goto out;
        }

        if (dstaddr->sa_family == AF_INET6)
                sport = &((struct sockaddr_in6 *)dstaddr)->sin6_port;
        else
                sport = &((struct sockaddr_in *)dstaddr)->sin_port;

        *sport = htons(port);

        WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED);
        sc->status = SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING;
        rc = rdma_resolve_addr(id, NULL, (struct sockaddr *)dstaddr,
                sp->resolve_addr_timeout_msec);
        if (rc) {
                log_rdma_event(ERR, "rdma_resolve_addr() failed %i\n", rc);
                goto out;
        }
        rc = wait_event_interruptible_timeout(
                sc->status_wait,
                sc->status != SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING,
                msecs_to_jiffies(sp->resolve_addr_timeout_msec));
        /* e.g. if interrupted returns -ERESTARTSYS */
        if (rc < 0) {
                log_rdma_event(ERR, "rdma_resolve_addr timeout rc: %i\n", rc);
                goto out;
        }
        if (sc->status == SMBDIRECT_SOCKET_RESOLVE_ADDR_RUNNING) {
                rc = -ETIMEDOUT;
                log_rdma_event(ERR, "rdma_resolve_addr() completed %i\n", rc);
                goto out;
        }
        if (sc->status != SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED) {
                rc = -EHOSTUNREACH;
                log_rdma_event(ERR, "rdma_resolve_addr() completed %i\n", rc);
                goto out;
        }

        WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_RESOLVE_ROUTE_NEEDED);
        sc->status = SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING;
        rc = rdma_resolve_route(id, sp->resolve_route_timeout_msec);
        if (rc) {
                log_rdma_event(ERR, "rdma_resolve_route() failed %i\n", rc);
                goto out;
        }
        rc = wait_event_interruptible_timeout(
                sc->status_wait,
                sc->status != SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING,
                msecs_to_jiffies(sp->resolve_route_timeout_msec));
        /* e.g. if interrupted returns -ERESTARTSYS */
        if (rc < 0)  {
                log_rdma_event(ERR, "rdma_resolve_addr timeout rc: %i\n", rc);
                goto out;
        }
        if (sc->status == SMBDIRECT_SOCKET_RESOLVE_ROUTE_RUNNING) {
                rc = -ETIMEDOUT;
                log_rdma_event(ERR, "rdma_resolve_route() completed %i\n", rc);
                goto out;
        }
        if (sc->status != SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED) {
                rc = -ENETUNREACH;
                log_rdma_event(ERR, "rdma_resolve_route() completed %i\n", rc);
                goto out;
        }

        return id;

out:
        rdma_destroy_id(id);
        return ERR_PTR(rc);
}

/*
 * Test if FRWR (Fast Registration Work Requests) is supported on the device
 * This implementation requires FRWR on RDMA read/write
 * return value: true if it is supported
 */
static bool frwr_is_supported(struct ib_device_attr *attrs)
{
        if (!(attrs->device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS))
                return false;
        if (attrs->max_fast_reg_page_list_len == 0)
                return false;
        return true;
}

static int smbd_ia_open(
                struct smbdirect_socket *sc,
                struct sockaddr *dstaddr, int port)
{
        struct smbdirect_socket_parameters *sp = &sc->parameters;
        int rc;

        WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_CREATED);
        sc->status = SMBDIRECT_SOCKET_RESOLVE_ADDR_NEEDED;

        sc->rdma.cm_id = smbd_create_id(sc, dstaddr, port);
        if (IS_ERR(sc->rdma.cm_id)) {
                rc = PTR_ERR(sc->rdma.cm_id);
                goto out1;
        }
        sc->ib.dev = sc->rdma.cm_id->device;

        if (!frwr_is_supported(&sc->ib.dev->attrs)) {
                log_rdma_event(ERR, "Fast Registration Work Requests (FRWR) is not supported\n");
                log_rdma_event(ERR, "Device capability flags = %llx max_fast_reg_page_list_len = %u\n",
                               sc->ib.dev->attrs.device_cap_flags,
                               sc->ib.dev->attrs.max_fast_reg_page_list_len);
                rc = -EPROTONOSUPPORT;
                goto out2;
        }
        sp->max_frmr_depth = min_t(u32,
                sp->max_frmr_depth,
                sc->ib.dev->attrs.max_fast_reg_page_list_len);
        sc->mr_io.type = IB_MR_TYPE_MEM_REG;
        if (sc->ib.dev->attrs.kernel_cap_flags & IBK_SG_GAPS_REG)
                sc->mr_io.type = IB_MR_TYPE_SG_GAPS;

        return 0;

out2:
        rdma_destroy_id(sc->rdma.cm_id);
        sc->rdma.cm_id = NULL;

out1:
        return rc;
}

/*
 * Send a negotiation request message to the peer
 * The negotiation procedure is in [MS-SMBD] 3.1.5.2 and 3.1.5.3
 * After negotiation, the transport is connected and ready for
 * carrying upper layer SMB payload
 */
static int smbd_post_send_negotiate_req(struct smbdirect_socket *sc)
{
        struct smbdirect_socket_parameters *sp = &sc->parameters;
        int rc;
        struct smbdirect_send_io *request;
        struct smbdirect_negotiate_req *packet;

        request = smbd_alloc_send_io(sc);
        if (IS_ERR(request))
                return PTR_ERR(request);

        packet = smbdirect_send_io_payload(request);
        packet->min_version = cpu_to_le16(SMBDIRECT_V1);
        packet->max_version = cpu_to_le16(SMBDIRECT_V1);
        packet->reserved = 0;
        packet->credits_requested = cpu_to_le16(sp->send_credit_target);
        packet->preferred_send_size = cpu_to_le32(sp->max_send_size);
        packet->max_receive_size = cpu_to_le32(sp->max_recv_size);
        packet->max_fragmented_size =
                cpu_to_le32(sp->max_fragmented_recv_size);

        request->sge[0].addr = ib_dma_map_single(
                                sc->ib.dev, (void *)packet,
                                sizeof(*packet), DMA_TO_DEVICE);
        if (ib_dma_mapping_error(sc->ib.dev, request->sge[0].addr)) {
                rc = -EIO;
                goto dma_mapping_failed;
        }
        request->num_sge = 1;

        request->sge[0].length = sizeof(*packet);
        request->sge[0].lkey = sc->ib.pd->local_dma_lkey;

        rc = smbd_post_send(sc, NULL, request);
        if (!rc)
                return 0;

        if (rc == -EAGAIN)
                rc = -EIO;

dma_mapping_failed:
        smbd_free_send_io(request);
        return rc;
}

/*
 * Extend the credits to remote peer
 * This implements [MS-SMBD] 3.1.5.9
 * The idea is that we should extend credits to remote peer as quickly as
 * it's allowed, to maintain data flow. We allocate as much receive
 * buffer as possible, and extend the receive credits to remote peer
 * return value: the new credtis being granted.
 */
static int manage_credits_prior_sending(struct smbdirect_socket *sc)
{
        int missing;
        int available;
        int new_credits;

        if (atomic_read(&sc->recv_io.credits.count) >= sc->recv_io.credits.target)
                return 0;

        missing = (int)sc->recv_io.credits.target - atomic_read(&sc->recv_io.credits.count);
        available = atomic_xchg(&sc->recv_io.credits.available, 0);
        new_credits = (u16)min3(U16_MAX, missing, available);
        if (new_credits <= 0) {
                /*
                 * If credits are available, but not granted
                 * we need to re-add them again.
                 */
                if (available)
                        atomic_add(available, &sc->recv_io.credits.available);
                return 0;
        }

        if (new_credits < available) {
                /*
                 * Readd the remaining available again.
                 */
                available -= new_credits;
                atomic_add(available, &sc->recv_io.credits.available);
        }

        /*
         * Remember we granted the credits
         */
        atomic_add(new_credits, &sc->recv_io.credits.count);
        return new_credits;
}

/*
 * Check if we need to send a KEEP_ALIVE message
 * The idle connection timer triggers a KEEP_ALIVE message when expires
 * SMBDIRECT_FLAG_RESPONSE_REQUESTED is set in the message flag to have peer send
 * back a response.
 * return value:
 * 1 if SMBDIRECT_FLAG_RESPONSE_REQUESTED needs to be set
 * 0: otherwise
 */
static int manage_keep_alive_before_sending(struct smbdirect_socket *sc)
{
        struct smbdirect_socket_parameters *sp = &sc->parameters;

        if (sc->idle.keepalive == SMBDIRECT_KEEPALIVE_PENDING) {
                sc->idle.keepalive = SMBDIRECT_KEEPALIVE_SENT;
                /*
                 * Now use the keepalive timeout (instead of keepalive interval)
                 * in order to wait for a response
                 */
                mod_delayed_work(sc->workqueue, &sc->idle.timer_work,
                                 msecs_to_jiffies(sp->keepalive_timeout_msec));
                return 1;
        }
        return 0;
}

static int smbd_ib_post_send(struct smbdirect_socket *sc,
                             struct ib_send_wr *wr)
{
        int ret;

        atomic_inc(&sc->send_io.pending.count);
        ret = ib_post_send(sc->ib.qp, wr, NULL);
        if (ret) {
                pr_err("failed to post send: %d\n", ret);
                smbd_disconnect_rdma_connection(sc);
                ret = -EAGAIN;
        }
        return ret;
}

/* Post the send request */
static int smbd_post_send(struct smbdirect_socket *sc,
                          struct smbdirect_send_batch *batch,
                          struct smbdirect_send_io *request)
{
        int i;

        for (i = 0; i < request->num_sge; i++) {
                log_rdma_send(INFO,
                        "rdma_request sge[%d] addr=0x%llx length=%u\n",
                        i, request->sge[i].addr, request->sge[i].length);
                ib_dma_sync_single_for_device(
                        sc->ib.dev,
                        request->sge[i].addr,
                        request->sge[i].length,
                        DMA_TO_DEVICE);
        }

        request->cqe.done = send_done;
        request->wr.next = NULL;
        request->wr.sg_list = request->sge;
        request->wr.num_sge = request->num_sge;
        request->wr.opcode = IB_WR_SEND;

        if (batch) {
                request->wr.wr_cqe = NULL;
                request->wr.send_flags = 0;
                if (!list_empty(&batch->msg_list)) {
                        struct smbdirect_send_io *last;

                        last = list_last_entry(&batch->msg_list,
                                               struct smbdirect_send_io,
                                               sibling_list);
                        last->wr.next = &request->wr;
                }
                list_add_tail(&request->sibling_list, &batch->msg_list);
                batch->wr_cnt++;
                return 0;
        }

        request->wr.wr_cqe = &request->cqe;
        request->wr.send_flags = IB_SEND_SIGNALED;
        return smbd_ib_post_send(sc, &request->wr);
}

static void smbd_send_batch_init(struct smbdirect_send_batch *batch,
                                 bool need_invalidate_rkey,
                                 unsigned int remote_key)
{
        INIT_LIST_HEAD(&batch->msg_list);
        batch->wr_cnt = 0;
        batch->need_invalidate_rkey = need_invalidate_rkey;
        batch->remote_key = remote_key;
        batch->credit = 0;
}

static int smbd_send_batch_flush(struct smbdirect_socket *sc,
                                 struct smbdirect_send_batch *batch,
                                 bool is_last)
{
        struct smbdirect_send_io *first, *last;
        int ret = 0;

        if (list_empty(&batch->msg_list))
                goto release_credit;

        first = list_first_entry(&batch->msg_list,
                                 struct smbdirect_send_io,
                                 sibling_list);
        last = list_last_entry(&batch->msg_list,
                               struct smbdirect_send_io,
                               sibling_list);

        if (batch->need_invalidate_rkey) {
                first->wr.opcode = IB_WR_SEND_WITH_INV;
                first->wr.ex.invalidate_rkey = batch->remote_key;
                batch->need_invalidate_rkey = false;
                batch->remote_key = 0;
        }

        last->wr.send_flags = IB_SEND_SIGNALED;
        last->wr.wr_cqe = &last->cqe;

        /*
         * Remove last from batch->msg_list
         * and splice the rest of batch->msg_list
         * to last->sibling_list.
         *
         * batch->msg_list is a valid empty list
         * at the end.
         */
        list_del_init(&last->sibling_list);
        list_splice_tail_init(&batch->msg_list, &last->sibling_list);
        batch->wr_cnt = 0;

        ret = smbd_ib_post_send(sc, &first->wr);
        if (ret) {
                struct smbdirect_send_io *sibling, *next;

                list_for_each_entry_safe(sibling, next, &last->sibling_list, sibling_list) {
                        list_del_init(&sibling->sibling_list);
                        smbd_free_send_io(sibling);
                }
                smbd_free_send_io(last);
        }

release_credit:
        if (is_last && !ret && batch->credit) {
                atomic_add(batch->credit, &sc->send_io.bcredits.count);
                batch->credit = 0;
                wake_up(&sc->send_io.bcredits.wait_queue);
        }

        return ret;
}

static int wait_for_credits(struct smbdirect_socket *sc,
                            wait_queue_head_t *waitq, atomic_t *total_credits,
                            int needed)
{
        int ret;

        do {
                if (atomic_sub_return(needed, total_credits) >= 0)
                        return 0;

                atomic_add(needed, total_credits);
                ret = wait_event_interruptible(*waitq,
                                               atomic_read(total_credits) >= needed ||
                                               sc->status != SMBDIRECT_SOCKET_CONNECTED);

                if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
                        return -ENOTCONN;
                else if (ret < 0)
                        return ret;
        } while (true);
}

static int wait_for_send_bcredit(struct smbdirect_socket *sc,
                                 struct smbdirect_send_batch *batch)
{
        int ret;

        if (batch->credit)
                return 0;

        ret = wait_for_credits(sc,
                               &sc->send_io.bcredits.wait_queue,
                               &sc->send_io.bcredits.count,
                               1);
        if (ret)
                return ret;

        batch->credit = 1;
        return 0;
}

static int wait_for_send_lcredit(struct smbdirect_socket *sc,
                                 struct smbdirect_send_batch *batch)
{
        if (batch && (atomic_read(&sc->send_io.lcredits.count) <= 1)) {
                int ret;

                ret = smbd_send_batch_flush(sc, batch, false);
                if (ret)
                        return ret;
        }

        return wait_for_credits(sc,
                                &sc->send_io.lcredits.wait_queue,
                                &sc->send_io.lcredits.count,
                                1);
}

static int wait_for_send_credits(struct smbdirect_socket *sc,
                                 struct smbdirect_send_batch *batch)
{
        if (batch &&
            (batch->wr_cnt >= 16 || atomic_read(&sc->send_io.credits.count) <= 1)) {
                int ret;

                ret = smbd_send_batch_flush(sc, batch, false);
                if (ret)
                        return ret;
        }

        return wait_for_credits(sc,
                                &sc->send_io.credits.wait_queue,
                                &sc->send_io.credits.count,
                                1);
}

static int smbd_post_send_iter(struct smbdirect_socket *sc,
                               struct smbdirect_send_batch *batch,
                               struct iov_iter *iter,
                               int *_remaining_data_length)
{
        struct smbdirect_socket_parameters *sp = &sc->parameters;
        int rc;
        int header_length;
        int data_length;
        struct smbdirect_send_io *request;
        struct smbdirect_data_transfer *packet;
        int new_credits = 0;
        struct smbdirect_send_batch _batch;

        if (!batch) {
                smbd_send_batch_init(&_batch, false, 0);
                batch = &_batch;
        }

        rc = wait_for_send_bcredit(sc, batch);
        if (rc) {
                log_outgoing(ERR, "disconnected not sending on wait_bcredit\n");
                rc = -EAGAIN;
                goto err_wait_bcredit;
        }

        rc = wait_for_send_lcredit(sc, batch);
        if (rc) {
                log_outgoing(ERR, "disconnected not sending on wait_lcredit\n");
                rc = -EAGAIN;
                goto err_wait_lcredit;
        }

        rc = wait_for_send_credits(sc, batch);
        if (rc) {
                log_outgoing(ERR, "disconnected not sending on wait_credit\n");
                rc = -EAGAIN;
                goto err_wait_credit;
        }

        new_credits = manage_credits_prior_sending(sc);
        if (new_credits == 0 &&
            atomic_read(&sc->send_io.credits.count) == 0 &&
            atomic_read(&sc->recv_io.credits.count) == 0) {
                queue_work(sc->workqueue, &sc->recv_io.posted.refill_work);
                rc = wait_event_interruptible(sc->send_io.credits.wait_queue,
                                              atomic_read(&sc->send_io.credits.count) >= 1 ||
                                              atomic_read(&sc->recv_io.credits.available) >= 1 ||
                                              sc->status != SMBDIRECT_SOCKET_CONNECTED);
                if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
                        rc = -ENOTCONN;
                if (rc < 0) {
                        log_outgoing(ERR, "disconnected not sending on last credit\n");
                        rc = -EAGAIN;
                        goto err_wait_credit;
                }

                new_credits = manage_credits_prior_sending(sc);
        }

        request = smbd_alloc_send_io(sc);
        if (IS_ERR(request)) {
                rc = PTR_ERR(request);
                goto err_alloc;
        }

        memset(request->sge, 0, sizeof(request->sge));

        /* Map the packet to DMA */
        header_length = sizeof(struct smbdirect_data_transfer);
        /* If this is a packet without payload, don't send padding */
        if (!iter)
                header_length = offsetof(struct smbdirect_data_transfer, padding);

        packet = smbdirect_send_io_payload(request);
        request->sge[0].addr = ib_dma_map_single(sc->ib.dev,
                                                 (void *)packet,
                                                 header_length,
                                                 DMA_TO_DEVICE);
        if (ib_dma_mapping_error(sc->ib.dev, request->sge[0].addr)) {
                rc = -EIO;
                goto err_dma;
        }

        request->sge[0].length = header_length;
        request->sge[0].lkey = sc->ib.pd->local_dma_lkey;
        request->num_sge = 1;

        /* Fill in the data payload to find out how much data we can add */
        if (iter) {
                struct smb_extract_to_rdma extract = {
                        .nr_sge         = request->num_sge,
                        .max_sge        = SMBDIRECT_SEND_IO_MAX_SGE,
                        .sge            = request->sge,
                        .device         = sc->ib.dev,
                        .local_dma_lkey = sc->ib.pd->local_dma_lkey,
                        .direction      = DMA_TO_DEVICE,
                };
                size_t payload_len = umin(*_remaining_data_length,
                                          sp->max_send_size - sizeof(*packet));

                rc = smb_extract_iter_to_rdma(iter, payload_len,
                                              &extract);
                if (rc < 0)
                        goto err_dma;
                data_length = rc;
                request->num_sge = extract.nr_sge;
                *_remaining_data_length -= data_length;
        } else {
                data_length = 0;
        }

        /* Fill in the packet header */
        packet->credits_requested = cpu_to_le16(sp->send_credit_target);
        packet->credits_granted = cpu_to_le16(new_credits);

        packet->flags = 0;
        if (manage_keep_alive_before_sending(sc))
                packet->flags |= cpu_to_le16(SMBDIRECT_FLAG_RESPONSE_REQUESTED);

        packet->reserved = 0;
        if (!data_length)
                packet->data_offset = 0;
        else
                packet->data_offset = cpu_to_le32(24);
        packet->data_length = cpu_to_le32(data_length);
        packet->remaining_data_length = cpu_to_le32(*_remaining_data_length);
        packet->padding = 0;

        log_outgoing(INFO, "credits_requested=%d credits_granted=%d data_offset=%d data_length=%d remaining_data_length=%d\n",
                     le16_to_cpu(packet->credits_requested),
                     le16_to_cpu(packet->credits_granted),
                     le32_to_cpu(packet->data_offset),
                     le32_to_cpu(packet->data_length),
                     le32_to_cpu(packet->remaining_data_length));

        rc = smbd_post_send(sc, batch, request);
        if (!rc) {
                if (batch != &_batch)
                        return 0;

                rc = smbd_send_batch_flush(sc, batch, true);
                if (!rc)
                        return 0;
        }

err_dma:
        smbd_free_send_io(request);

err_alloc:
        atomic_inc(&sc->send_io.credits.count);
        wake_up(&sc->send_io.credits.wait_queue);

err_wait_credit:
        atomic_inc(&sc->send_io.lcredits.count);
        wake_up(&sc->send_io.lcredits.wait_queue);

err_wait_lcredit:
        atomic_add(batch->credit, &sc->send_io.bcredits.count);
        batch->credit = 0;
        wake_up(&sc->send_io.bcredits.wait_queue);

err_wait_bcredit:
        return rc;
}

/*
 * Send an empty message
 * Empty message is used to extend credits to peer to for keep live
 * while there is no upper layer payload to send at the time
 */
static int smbd_post_send_empty(struct smbdirect_socket *sc)
{
        int remaining_data_length = 0;

        sc->statistics.send_empty++;
        return smbd_post_send_iter(sc, NULL, NULL, &remaining_data_length);
}

static int smbd_post_send_full_iter(struct smbdirect_socket *sc,
                                    struct smbdirect_send_batch *batch,
                                    struct iov_iter *iter,
                                    int *_remaining_data_length)
{
        int rc = 0;

        /*
         * smbd_post_send_iter() respects the
         * negotiated max_send_size, so we need to
         * loop until the full iter is posted
         */

        while (iov_iter_count(iter) > 0) {
                rc = smbd_post_send_iter(sc, batch, iter, _remaining_data_length);
                if (rc < 0)
                        break;
        }

        return rc;
}

/*
 * Post a receive request to the transport
 * The remote peer can only send data when a receive request is posted
 * The interaction is controlled by send/receive credit system
 */
static int smbd_post_recv(
                struct smbdirect_socket *sc, struct smbdirect_recv_io *response)
{
        struct smbdirect_socket_parameters *sp = &sc->parameters;
        struct ib_recv_wr recv_wr;
        int rc = -EIO;

        response->sge.addr = ib_dma_map_single(
                                sc->ib.dev, response->packet,
                                sp->max_recv_size, DMA_FROM_DEVICE);
        if (ib_dma_mapping_error(sc->ib.dev, response->sge.addr))
                return rc;

        response->sge.length = sp->max_recv_size;
        response->sge.lkey = sc->ib.pd->local_dma_lkey;

        response->cqe.done = recv_done;

        recv_wr.wr_cqe = &response->cqe;
        recv_wr.next = NULL;
        recv_wr.sg_list = &response->sge;
        recv_wr.num_sge = 1;

        rc = ib_post_recv(sc->ib.qp, &recv_wr, NULL);
        if (rc) {
                ib_dma_unmap_single(sc->ib.dev, response->sge.addr,
                                    response->sge.length, DMA_FROM_DEVICE);
                response->sge.length = 0;
                smbd_disconnect_rdma_connection(sc);
                log_rdma_recv(ERR, "ib_post_recv failed rc=%d\n", rc);
        }

        return rc;
}

/* Perform SMBD negotiate according to [MS-SMBD] 3.1.5.2 */
static int smbd_negotiate(struct smbdirect_socket *sc)
{
        struct smbdirect_socket_parameters *sp = &sc->parameters;
        int rc;
        struct smbdirect_recv_io *response = get_receive_buffer(sc);

        WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_NEGOTIATE_NEEDED);
        sc->status = SMBDIRECT_SOCKET_NEGOTIATE_RUNNING;

        sc->recv_io.expected = SMBDIRECT_EXPECT_NEGOTIATE_REP;
        rc = smbd_post_recv(sc, response);
        log_rdma_event(INFO, "smbd_post_recv rc=%d iov.addr=0x%llx iov.length=%u iov.lkey=0x%x\n",
                       rc, response->sge.addr,
                       response->sge.length, response->sge.lkey);
        if (rc) {
                put_receive_buffer(sc, response);
                return rc;
        }

        rc = smbd_post_send_negotiate_req(sc);
        if (rc)
                return rc;

        rc = wait_event_interruptible_timeout(
                sc->status_wait,
                sc->status != SMBDIRECT_SOCKET_NEGOTIATE_RUNNING,
                msecs_to_jiffies(sp->negotiate_timeout_msec));
        log_rdma_event(INFO, "wait_event_interruptible_timeout rc=%d\n", rc);

        if (sc->status == SMBDIRECT_SOCKET_CONNECTED)
                return 0;

        if (rc == 0)
                rc = -ETIMEDOUT;
        else if (rc == -ERESTARTSYS)
                rc = -EINTR;
        else
                rc = -ENOTCONN;

        return rc;
}

/*
 * Implement Connection.FragmentReassemblyBuffer defined in [MS-SMBD] 3.1.1.1
 * This is a queue for reassembling upper layer payload and present to upper
 * layer. All the inncoming payload go to the reassembly queue, regardless of
 * if reassembly is required. The uuper layer code reads from the queue for all
 * incoming payloads.
 * Put a received packet to the reassembly queue
 * response: the packet received
 * data_length: the size of payload in this packet
 */
static void enqueue_reassembly(
        struct smbdirect_socket *sc,
        struct smbdirect_recv_io *response,
        int data_length)
{
        unsigned long flags;

        spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
        list_add_tail(&response->list, &sc->recv_io.reassembly.list);
        sc->recv_io.reassembly.queue_length++;
        /*
         * Make sure reassembly_data_length is updated after list and
         * reassembly_queue_length are updated. On the dequeue side
         * reassembly_data_length is checked without a lock to determine
         * if reassembly_queue_length and list is up to date
         */
        virt_wmb();
        sc->recv_io.reassembly.data_length += data_length;
        spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
        sc->statistics.enqueue_reassembly_queue++;
}

/*
 * Get the first entry at the front of reassembly queue
 * Caller is responsible for locking
 * return value: the first entry if any, NULL if queue is empty
 */
static struct smbdirect_recv_io *_get_first_reassembly(struct smbdirect_socket *sc)
{
        struct smbdirect_recv_io *ret = NULL;

        if (!list_empty(&sc->recv_io.reassembly.list)) {
                ret = list_first_entry(
                        &sc->recv_io.reassembly.list,
                        struct smbdirect_recv_io, list);
        }
        return ret;
}

/*
 * Get a receive buffer
 * For each remote send, we need to post a receive. The receive buffers are
 * pre-allocated in advance.
 * return value: the receive buffer, NULL if none is available
 */
static struct smbdirect_recv_io *get_receive_buffer(struct smbdirect_socket *sc)
{
        struct smbdirect_recv_io *ret = NULL;
        unsigned long flags;

        spin_lock_irqsave(&sc->recv_io.free.lock, flags);
        if (!list_empty(&sc->recv_io.free.list)) {
                ret = list_first_entry(
                        &sc->recv_io.free.list,
                        struct smbdirect_recv_io, list);
                list_del(&ret->list);
                sc->statistics.get_receive_buffer++;
        }
        spin_unlock_irqrestore(&sc->recv_io.free.lock, flags);

        return ret;
}

/*
 * Return a receive buffer
 * Upon returning of a receive buffer, we can post new receive and extend
 * more receive credits to remote peer. This is done immediately after a
 * receive buffer is returned.
 */
static void put_receive_buffer(
        struct smbdirect_socket *sc, struct smbdirect_recv_io *response)
{
        unsigned long flags;

        if (likely(response->sge.length != 0)) {
                ib_dma_unmap_single(sc->ib.dev,
                                    response->sge.addr,
                                    response->sge.length,
                                    DMA_FROM_DEVICE);
                response->sge.length = 0;
        }

        spin_lock_irqsave(&sc->recv_io.free.lock, flags);
        list_add_tail(&response->list, &sc->recv_io.free.list);
        sc->statistics.put_receive_buffer++;
        spin_unlock_irqrestore(&sc->recv_io.free.lock, flags);

        queue_work(sc->workqueue, &sc->recv_io.posted.refill_work);
}

/* Preallocate all receive buffer on transport establishment */
static int allocate_receive_buffers(struct smbdirect_socket *sc, int num_buf)
{
        struct smbdirect_recv_io *response;
        int i;

        for (i = 0; i < num_buf; i++) {
                response = mempool_alloc(sc->recv_io.mem.pool, GFP_KERNEL);
                if (!response)
                        goto allocate_failed;

                response->socket = sc;
                response->sge.length = 0;
                list_add_tail(&response->list, &sc->recv_io.free.list);
        }

        return 0;

allocate_failed:
        while (!list_empty(&sc->recv_io.free.list)) {
                response = list_first_entry(
                                &sc->recv_io.free.list,
                                struct smbdirect_recv_io, list);
                list_del(&response->list);

                mempool_free(response, sc->recv_io.mem.pool);
        }
        return -ENOMEM;
}

static void destroy_receive_buffers(struct smbdirect_socket *sc)
{
        struct smbdirect_recv_io *response;

        while ((response = get_receive_buffer(sc)))
                mempool_free(response, sc->recv_io.mem.pool);
}

static void send_immediate_empty_message(struct work_struct *work)
{
        struct smbdirect_socket *sc =
                container_of(work, struct smbdirect_socket, idle.immediate_work);

        if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
                return;

        log_keep_alive(INFO, "send an empty message\n");
        smbd_post_send_empty(sc);
}

/* Implement idle connection timer [MS-SMBD] 3.1.6.2 */
static void idle_connection_timer(struct work_struct *work)
{
        struct smbdirect_socket *sc =
                container_of(work, struct smbdirect_socket, idle.timer_work.work);
        struct smbdirect_socket_parameters *sp = &sc->parameters;

        if (sc->idle.keepalive != SMBDIRECT_KEEPALIVE_NONE) {
                log_keep_alive(ERR,
                        "error status sc->idle.keepalive=%d\n",
                        sc->idle.keepalive);
                smbd_disconnect_rdma_connection(sc);
                return;
        }

        if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
                return;

        /*
         * Now use the keepalive timeout (instead of keepalive interval)
         * in order to wait for a response
         */
        sc->idle.keepalive = SMBDIRECT_KEEPALIVE_PENDING;
        mod_delayed_work(sc->workqueue, &sc->idle.timer_work,
                         msecs_to_jiffies(sp->keepalive_timeout_msec));
        log_keep_alive(INFO, "schedule send of empty idle message\n");
        queue_work(sc->workqueue, &sc->idle.immediate_work);
}

/*
 * Destroy the transport and related RDMA and memory resources
 * Need to go through all the pending counters and make sure on one is using
 * the transport while it is destroyed
 */
void smbd_destroy(struct TCP_Server_Info *server)
{
        struct smbd_connection *info = server->smbd_conn;
        struct smbdirect_socket *sc;
        struct smbdirect_recv_io *response;
        unsigned long flags;

        if (!info) {
                log_rdma_event(INFO, "rdma session already destroyed\n");
                return;
        }
        sc = &info->socket;

        log_rdma_event(INFO, "cancelling and disable disconnect_work\n");
        disable_work_sync(&sc->disconnect_work);

        log_rdma_event(INFO, "destroying rdma session\n");
        if (sc->status < SMBDIRECT_SOCKET_DISCONNECTING)
                smbd_disconnect_rdma_work(&sc->disconnect_work);
        if (sc->status < SMBDIRECT_SOCKET_DISCONNECTED) {
                log_rdma_event(INFO, "wait for transport being disconnected\n");
                wait_event(sc->status_wait, sc->status == SMBDIRECT_SOCKET_DISCONNECTED);
                log_rdma_event(INFO, "waited for transport being disconnected\n");
        }

        /*
         * Wake up all waiters in all wait queues
         * in order to notice the broken connection.
         *
         * Most likely this was already called via
         * smbd_disconnect_rdma_work(), but call it again...
         */
        smbd_disconnect_wake_up_all(sc);

        log_rdma_event(INFO, "cancelling recv_io.posted.refill_work\n");
        disable_work_sync(&sc->recv_io.posted.refill_work);

        log_rdma_event(INFO, "destroying qp\n");
        ib_drain_qp(sc->ib.qp);
        rdma_destroy_qp(sc->rdma.cm_id);
        sc->ib.qp = NULL;

        log_rdma_event(INFO, "cancelling idle timer\n");
        disable_delayed_work_sync(&sc->idle.timer_work);
        log_rdma_event(INFO, "cancelling send immediate work\n");
        disable_work_sync(&sc->idle.immediate_work);

        /* It's not possible for upper layer to get to reassembly */
        log_rdma_event(INFO, "drain the reassembly queue\n");
        do {
                spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
                response = _get_first_reassembly(sc);
                if (response) {
                        list_del(&response->list);
                        spin_unlock_irqrestore(
                                &sc->recv_io.reassembly.lock, flags);
                        put_receive_buffer(sc, response);
                } else
                        spin_unlock_irqrestore(
                                &sc->recv_io.reassembly.lock, flags);
        } while (response);
        sc->recv_io.reassembly.data_length = 0;

        log_rdma_event(INFO, "free receive buffers\n");
        destroy_receive_buffers(sc);

        log_rdma_event(INFO, "freeing mr list\n");
        destroy_mr_list(sc);

        ib_free_cq(sc->ib.send_cq);
        ib_free_cq(sc->ib.recv_cq);
        ib_dealloc_pd(sc->ib.pd);
        rdma_destroy_id(sc->rdma.cm_id);

        /* free mempools */
        mempool_destroy(sc->send_io.mem.pool);
        kmem_cache_destroy(sc->send_io.mem.cache);

        mempool_destroy(sc->recv_io.mem.pool);
        kmem_cache_destroy(sc->recv_io.mem.cache);

        sc->status = SMBDIRECT_SOCKET_DESTROYED;

        destroy_workqueue(sc->workqueue);
        log_rdma_event(INFO,  "rdma session destroyed\n");
        kfree(info);
        server->smbd_conn = NULL;
}

/*
 * Reconnect this SMBD connection, called from upper layer
 * return value: 0 on success, or actual error code
 */
int smbd_reconnect(struct TCP_Server_Info *server)
{
        log_rdma_event(INFO, "reconnecting rdma session\n");

        if (!server->smbd_conn) {
                log_rdma_event(INFO, "rdma session already destroyed\n");
                goto create_conn;
        }

        /*
         * This is possible if transport is disconnected and we haven't received
         * notification from RDMA, but upper layer has detected timeout
         */
        if (server->smbd_conn->socket.status == SMBDIRECT_SOCKET_CONNECTED) {
                log_rdma_event(INFO, "disconnecting transport\n");
                smbd_destroy(server);
        }

create_conn:
        log_rdma_event(INFO, "creating rdma session\n");
        server->smbd_conn = smbd_get_connection(
                server, (struct sockaddr *) &server->dstaddr);

        if (server->smbd_conn) {
                cifs_dbg(VFS, "RDMA transport re-established\n");
                trace_smb3_smbd_connect_done(server->hostname, server->conn_id, &server->dstaddr);
                return 0;
        }
        trace_smb3_smbd_connect_err(server->hostname, server->conn_id, &server->dstaddr);
        return -ENOENT;
}

static void destroy_caches(struct smbdirect_socket *sc)
{
        destroy_receive_buffers(sc);
        mempool_destroy(sc->recv_io.mem.pool);
        kmem_cache_destroy(sc->recv_io.mem.cache);
        mempool_destroy(sc->send_io.mem.pool);
        kmem_cache_destroy(sc->send_io.mem.cache);
}

#define MAX_NAME_LEN    80
static int allocate_caches(struct smbdirect_socket *sc)
{
        struct smbdirect_socket_parameters *sp = &sc->parameters;
        char name[MAX_NAME_LEN];
        int rc;

        if (WARN_ON_ONCE(sp->max_recv_size < sizeof(struct smbdirect_data_transfer)))
                return -ENOMEM;

        scnprintf(name, MAX_NAME_LEN, "smbdirect_send_io_%p", sc);
        sc->send_io.mem.cache =
                kmem_cache_create(
                        name,
                        sizeof(struct smbdirect_send_io) +
                                sizeof(struct smbdirect_data_transfer),
                        0, SLAB_HWCACHE_ALIGN, NULL);
        if (!sc->send_io.mem.cache)
                return -ENOMEM;

        sc->send_io.mem.pool =
                mempool_create(sp->send_credit_target, mempool_alloc_slab,
                        mempool_free_slab, sc->send_io.mem.cache);
        if (!sc->send_io.mem.pool)
                goto out1;

        scnprintf(name, MAX_NAME_LEN, "smbdirect_recv_io_%p", sc);

        struct kmem_cache_args response_args = {
                .align          = __alignof__(struct smbdirect_recv_io),
                .useroffset     = (offsetof(struct smbdirect_recv_io, packet) +
                                   sizeof(struct smbdirect_data_transfer)),
                .usersize       = sp->max_recv_size - sizeof(struct smbdirect_data_transfer),
        };
        sc->recv_io.mem.cache =
                kmem_cache_create(name,
                                  sizeof(struct smbdirect_recv_io) + sp->max_recv_size,
                                  &response_args, SLAB_HWCACHE_ALIGN);
        if (!sc->recv_io.mem.cache)
                goto out2;

        sc->recv_io.mem.pool =
                mempool_create(sp->recv_credit_max, mempool_alloc_slab,
                       mempool_free_slab, sc->recv_io.mem.cache);
        if (!sc->recv_io.mem.pool)
                goto out3;

        rc = allocate_receive_buffers(sc, sp->recv_credit_max);
        if (rc) {
                log_rdma_event(ERR, "failed to allocate receive buffers\n");
                goto out4;
        }

        return 0;

out4:
        mempool_destroy(sc->recv_io.mem.pool);
out3:
        kmem_cache_destroy(sc->recv_io.mem.cache);
out2:
        mempool_destroy(sc->send_io.mem.pool);
out1:
        kmem_cache_destroy(sc->send_io.mem.cache);
        return -ENOMEM;
}

/* Create a SMBD connection, called by upper layer */
static struct smbd_connection *_smbd_get_connection(
        struct TCP_Server_Info *server, struct sockaddr *dstaddr, int port)
{
        int rc;
        struct smbd_connection *info;
        struct smbdirect_socket *sc;
        struct smbdirect_socket_parameters *sp;
        struct rdma_conn_param conn_param;
        struct ib_qp_cap qp_cap;
        struct ib_qp_init_attr qp_attr;
        struct sockaddr_in *addr_in = (struct sockaddr_in *) dstaddr;
        struct ib_port_immutable port_immutable;
        __be32 ird_ord_hdr[2];
        char wq_name[80];
        struct workqueue_struct *workqueue;

        info = kzalloc_obj(struct smbd_connection);
        if (!info)
                return NULL;
        sc = &info->socket;
        scnprintf(wq_name, ARRAY_SIZE(wq_name), "smbd_%p", sc);
        workqueue = create_workqueue(wq_name);
        if (!workqueue)
                goto create_wq_failed;
        smbdirect_socket_init(sc);
        sc->workqueue = workqueue;
        sp = &sc->parameters;

        INIT_WORK(&sc->disconnect_work, smbd_disconnect_rdma_work);

        sp->resolve_addr_timeout_msec = RDMA_RESOLVE_TIMEOUT;
        sp->resolve_route_timeout_msec = RDMA_RESOLVE_TIMEOUT;
        sp->rdma_connect_timeout_msec = RDMA_RESOLVE_TIMEOUT;
        sp->negotiate_timeout_msec = SMBD_NEGOTIATE_TIMEOUT * 1000;
        sp->initiator_depth = 1;
        sp->responder_resources = SMBD_CM_RESPONDER_RESOURCES;
        sp->recv_credit_max = smbd_receive_credit_max;
        sp->send_credit_target = smbd_send_credit_target;
        sp->max_send_size = smbd_max_send_size;
        sp->max_fragmented_recv_size = smbd_max_fragmented_recv_size;
        sp->max_recv_size = smbd_max_receive_size;
        sp->max_frmr_depth = smbd_max_frmr_depth;
        sp->keepalive_interval_msec = smbd_keep_alive_interval * 1000;
        sp->keepalive_timeout_msec = KEEPALIVE_RECV_TIMEOUT * 1000;

        rc = smbd_ia_open(sc, dstaddr, port);
        if (rc) {
                log_rdma_event(INFO, "smbd_ia_open rc=%d\n", rc);
                goto create_id_failed;
        }

        if (sp->send_credit_target > sc->ib.dev->attrs.max_cqe ||
            sp->send_credit_target > sc->ib.dev->attrs.max_qp_wr) {
                log_rdma_event(ERR, "consider lowering send_credit_target = %d. Possible CQE overrun, device reporting max_cqe %d max_qp_wr %d\n",
                               sp->send_credit_target,
                               sc->ib.dev->attrs.max_cqe,
                               sc->ib.dev->attrs.max_qp_wr);
                goto config_failed;
        }

        if (sp->recv_credit_max > sc->ib.dev->attrs.max_cqe ||
            sp->recv_credit_max > sc->ib.dev->attrs.max_qp_wr) {
                log_rdma_event(ERR, "consider lowering receive_credit_max = %d. Possible CQE overrun, device reporting max_cqe %d max_qp_wr %d\n",
                               sp->recv_credit_max,
                               sc->ib.dev->attrs.max_cqe,
                               sc->ib.dev->attrs.max_qp_wr);
                goto config_failed;
        }

        if (sc->ib.dev->attrs.max_send_sge < SMBDIRECT_SEND_IO_MAX_SGE ||
            sc->ib.dev->attrs.max_recv_sge < SMBDIRECT_RECV_IO_MAX_SGE) {
                log_rdma_event(ERR,
                        "device %.*s max_send_sge/max_recv_sge = %d/%d too small\n",
                        IB_DEVICE_NAME_MAX,
                        sc->ib.dev->name,
                        sc->ib.dev->attrs.max_send_sge,
                        sc->ib.dev->attrs.max_recv_sge);
                goto config_failed;
        }

        sp->responder_resources =
                min_t(u8, sp->responder_resources,
                      sc->ib.dev->attrs.max_qp_rd_atom);
        log_rdma_mr(INFO, "responder_resources=%d\n",
                sp->responder_resources);

        /*
         * We use allocate sp->responder_resources * 2 MRs
         * and each MR needs WRs for REG and INV, so
         * we use '* 4'.
         *
         * +1 for ib_drain_qp()
         */
        memset(&qp_cap, 0, sizeof(qp_cap));
        qp_cap.max_send_wr = sp->send_credit_target + sp->responder_resources * 4 + 1;
        qp_cap.max_recv_wr = sp->recv_credit_max + 1;
        qp_cap.max_send_sge = SMBDIRECT_SEND_IO_MAX_SGE;
        qp_cap.max_recv_sge = SMBDIRECT_RECV_IO_MAX_SGE;

        sc->ib.pd = ib_alloc_pd(sc->ib.dev, 0);
        if (IS_ERR(sc->ib.pd)) {
                rc = PTR_ERR(sc->ib.pd);
                sc->ib.pd = NULL;
                log_rdma_event(ERR, "ib_alloc_pd() returned %d\n", rc);
                goto alloc_pd_failed;
        }

        sc->ib.send_cq =
                ib_alloc_cq_any(sc->ib.dev, sc,
                                qp_cap.max_send_wr, IB_POLL_SOFTIRQ);
        if (IS_ERR(sc->ib.send_cq)) {
                sc->ib.send_cq = NULL;
                goto alloc_cq_failed;
        }

        sc->ib.recv_cq =
                ib_alloc_cq_any(sc->ib.dev, sc,
                                qp_cap.max_recv_wr, IB_POLL_SOFTIRQ);
        if (IS_ERR(sc->ib.recv_cq)) {
                sc->ib.recv_cq = NULL;
                goto alloc_cq_failed;
        }

        memset(&qp_attr, 0, sizeof(qp_attr));
        qp_attr.event_handler = smbd_qp_async_error_upcall;
        qp_attr.qp_context = sc;
        qp_attr.cap = qp_cap;
        qp_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
        qp_attr.qp_type = IB_QPT_RC;
        qp_attr.send_cq = sc->ib.send_cq;
        qp_attr.recv_cq = sc->ib.recv_cq;
        qp_attr.port_num = ~0;

        rc = rdma_create_qp(sc->rdma.cm_id, sc->ib.pd, &qp_attr);
        if (rc) {
                log_rdma_event(ERR, "rdma_create_qp failed %i\n", rc);
                goto create_qp_failed;
        }
        sc->ib.qp = sc->rdma.cm_id->qp;

        memset(&conn_param, 0, sizeof(conn_param));
        conn_param.initiator_depth = sp->initiator_depth;
        conn_param.responder_resources = sp->responder_resources;

        /* Need to send IRD/ORD in private data for iWARP */
        sc->ib.dev->ops.get_port_immutable(
                sc->ib.dev, sc->rdma.cm_id->port_num, &port_immutable);
        if (port_immutable.core_cap_flags & RDMA_CORE_PORT_IWARP) {
                ird_ord_hdr[0] = cpu_to_be32(conn_param.responder_resources);
                ird_ord_hdr[1] = cpu_to_be32(conn_param.initiator_depth);
                conn_param.private_data = ird_ord_hdr;
                conn_param.private_data_len = sizeof(ird_ord_hdr);
        } else {
                conn_param.private_data = NULL;
                conn_param.private_data_len = 0;
        }

        conn_param.retry_count = SMBD_CM_RETRY;
        conn_param.rnr_retry_count = SMBD_CM_RNR_RETRY;
        conn_param.flow_control = 0;

        log_rdma_event(INFO, "connecting to IP %pI4 port %d\n",
                &addr_in->sin_addr, port);

        WARN_ON_ONCE(sc->status != SMBDIRECT_SOCKET_RDMA_CONNECT_NEEDED);
        sc->status = SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING;
        rc = rdma_connect(sc->rdma.cm_id, &conn_param);
        if (rc) {
                log_rdma_event(ERR, "rdma_connect() failed with %i\n", rc);
                goto rdma_connect_failed;
        }

        wait_event_interruptible_timeout(
                sc->status_wait,
                sc->status != SMBDIRECT_SOCKET_RDMA_CONNECT_RUNNING,
                msecs_to_jiffies(sp->rdma_connect_timeout_msec));

        if (sc->status != SMBDIRECT_SOCKET_NEGOTIATE_NEEDED) {
                log_rdma_event(ERR, "rdma_connect failed port=%d\n", port);
                goto rdma_connect_failed;
        }

        log_rdma_event(INFO, "rdma_connect connected\n");

        rc = allocate_caches(sc);
        if (rc) {
                log_rdma_event(ERR, "cache allocation failed\n");
                goto allocate_cache_failed;
        }

        INIT_WORK(&sc->idle.immediate_work, send_immediate_empty_message);
        INIT_DELAYED_WORK(&sc->idle.timer_work, idle_connection_timer);
        /*
         * start with the negotiate timeout and SMBDIRECT_KEEPALIVE_PENDING
         * so that the timer will cause a disconnect.
         */
        sc->idle.keepalive = SMBDIRECT_KEEPALIVE_PENDING;
        mod_delayed_work(sc->workqueue, &sc->idle.timer_work,
                         msecs_to_jiffies(sp->negotiate_timeout_msec));

        INIT_WORK(&sc->recv_io.posted.refill_work, smbd_post_send_credits);

        rc = smbd_negotiate(sc);
        if (rc) {
                log_rdma_event(ERR, "smbd_negotiate rc=%d\n", rc);
                goto negotiation_failed;
        }

        rc = allocate_mr_list(sc);
        if (rc) {
                log_rdma_mr(ERR, "memory registration allocation failed\n");
                goto allocate_mr_failed;
        }

        return info;

allocate_mr_failed:
        /* At this point, need to a full transport shutdown */
        server->smbd_conn = info;
        smbd_destroy(server);
        return NULL;

negotiation_failed:
        disable_delayed_work_sync(&sc->idle.timer_work);
        destroy_caches(sc);
        sc->status = SMBDIRECT_SOCKET_NEGOTIATE_FAILED;
        rdma_disconnect(sc->rdma.cm_id);
        wait_event(sc->status_wait,
                sc->status == SMBDIRECT_SOCKET_DISCONNECTED);

allocate_cache_failed:
rdma_connect_failed:
        rdma_destroy_qp(sc->rdma.cm_id);

create_qp_failed:
alloc_cq_failed:
        if (sc->ib.send_cq)
                ib_free_cq(sc->ib.send_cq);
        if (sc->ib.recv_cq)
                ib_free_cq(sc->ib.recv_cq);

        ib_dealloc_pd(sc->ib.pd);

alloc_pd_failed:
config_failed:
        rdma_destroy_id(sc->rdma.cm_id);

create_id_failed:
        destroy_workqueue(sc->workqueue);
create_wq_failed:
        kfree(info);
        return NULL;
}

struct smbd_connection *smbd_get_connection(
        struct TCP_Server_Info *server, struct sockaddr *dstaddr)
{
        struct smbd_connection *ret;
        const struct smbdirect_socket_parameters *sp;
        int port = SMBD_PORT;

try_again:
        ret = _smbd_get_connection(server, dstaddr, port);

        /* Try SMB_PORT if SMBD_PORT doesn't work */
        if (!ret && port == SMBD_PORT) {
                port = SMB_PORT;
                goto try_again;
        }
        if (!ret)
                return NULL;

        sp = &ret->socket.parameters;

        server->rdma_readwrite_threshold =
                rdma_readwrite_threshold > sp->max_fragmented_send_size ?
                sp->max_fragmented_send_size :
                rdma_readwrite_threshold;

        return ret;
}

/*
 * Receive data from the transport's receive reassembly queue
 * All the incoming data packets are placed in reassembly queue
 * iter: the buffer to read data into
 * size: the length of data to read
 * return value: actual data read
 *
 * Note: this implementation copies the data from reassembly queue to receive
 * buffers used by upper layer. This is not the optimal code path. A better way
 * to do it is to not have upper layer allocate its receive buffers but rather
 * borrow the buffer from reassembly queue, and return it after data is
 * consumed. But this will require more changes to upper layer code, and also
 * need to consider packet boundaries while they still being reassembled.
 */
int smbd_recv(struct smbd_connection *info, struct msghdr *msg)
{
        struct smbdirect_socket *sc = &info->socket;
        struct smbdirect_recv_io *response;
        struct smbdirect_data_transfer *data_transfer;
        size_t size = iov_iter_count(&msg->msg_iter);
        int to_copy, to_read, data_read, offset;
        u32 data_length, remaining_data_length, data_offset;
        int rc;

        if (WARN_ON_ONCE(iov_iter_rw(&msg->msg_iter) == WRITE))
                return -EINVAL; /* It's a bug in upper layer to get there */

again:
        /*
         * No need to hold the reassembly queue lock all the time as we are
         * the only one reading from the front of the queue. The transport
         * may add more entries to the back of the queue at the same time
         */
        log_read(INFO, "size=%zd sc->recv_io.reassembly.data_length=%d\n", size,
                sc->recv_io.reassembly.data_length);
        if (sc->recv_io.reassembly.data_length >= size) {
                int queue_length;
                int queue_removed = 0;
                unsigned long flags;

                /*
                 * Need to make sure reassembly_data_length is read before
                 * reading reassembly_queue_length and calling
                 * _get_first_reassembly. This call is lock free
                 * as we never read at the end of the queue which are being
                 * updated in SOFTIRQ as more data is received
                 */
                virt_rmb();
                queue_length = sc->recv_io.reassembly.queue_length;
                data_read = 0;
                to_read = size;
                offset = sc->recv_io.reassembly.first_entry_offset;
                while (data_read < size) {
                        response = _get_first_reassembly(sc);
                        data_transfer = smbdirect_recv_io_payload(response);
                        data_length = le32_to_cpu(data_transfer->data_length);
                        remaining_data_length =
                                le32_to_cpu(
                                        data_transfer->remaining_data_length);
                        data_offset = le32_to_cpu(data_transfer->data_offset);

                        /*
                         * The upper layer expects RFC1002 length at the
                         * beginning of the payload. Return it to indicate
                         * the total length of the packet. This minimize the
                         * change to upper layer packet processing logic. This
                         * will be eventually remove when an intermediate
                         * transport layer is added
                         */
                        if (response->first_segment && size == 4) {
                                unsigned int rfc1002_len =
                                        data_length + remaining_data_length;
                                __be32 rfc1002_hdr = cpu_to_be32(rfc1002_len);
                                if (copy_to_iter(&rfc1002_hdr, sizeof(rfc1002_hdr),
                                                 &msg->msg_iter) != sizeof(rfc1002_hdr))
                                        return -EFAULT;
                                data_read = 4;
                                response->first_segment = false;
                                log_read(INFO, "returning rfc1002 length %d\n",
                                        rfc1002_len);
                                goto read_rfc1002_done;
                        }

                        to_copy = min_t(int, data_length - offset, to_read);
                        if (copy_to_iter((char *)data_transfer + data_offset + offset,
                                         to_copy, &msg->msg_iter) != to_copy)
                                return -EFAULT;

                        /* move on to the next buffer? */
                        if (to_copy == data_length - offset) {
                                queue_length--;
                                /*
                                 * No need to lock if we are not at the
                                 * end of the queue
                                 */
                                if (queue_length)
                                        list_del(&response->list);
                                else {
                                        spin_lock_irqsave(
                                                &sc->recv_io.reassembly.lock, flags);
                                        list_del(&response->list);
                                        spin_unlock_irqrestore(
                                                &sc->recv_io.reassembly.lock, flags);
                                }
                                queue_removed++;
                                sc->statistics.dequeue_reassembly_queue++;
                                put_receive_buffer(sc, response);
                                offset = 0;
                                log_read(INFO, "put_receive_buffer offset=0\n");
                        } else
                                offset += to_copy;

                        to_read -= to_copy;
                        data_read += to_copy;

                        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",
                                 to_copy, data_length - offset,
                                 to_read, data_read, offset);
                }

                spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
                sc->recv_io.reassembly.data_length -= data_read;
                sc->recv_io.reassembly.queue_length -= queue_removed;
                spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);

                sc->recv_io.reassembly.first_entry_offset = offset;
                log_read(INFO, "returning to thread data_read=%d reassembly_data_length=%d first_entry_offset=%d\n",
                         data_read, sc->recv_io.reassembly.data_length,
                         sc->recv_io.reassembly.first_entry_offset);
read_rfc1002_done:
                return data_read;
        }

        log_read(INFO, "wait_event on more data\n");
        rc = wait_event_interruptible(
                sc->recv_io.reassembly.wait_queue,
                sc->recv_io.reassembly.data_length >= size ||
                        sc->status != SMBDIRECT_SOCKET_CONNECTED);
        /* Don't return any data if interrupted */
        if (rc)
                return rc;

        if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
                log_read(ERR, "disconnected\n");
                return -ECONNABORTED;
        }

        goto again;
}

/*
 * Send data to transport
 * Each rqst is transported as a SMBDirect payload
 * rqst: the data to write
 * return value: 0 if successfully write, otherwise error code
 */
int smbd_send(struct TCP_Server_Info *server,
        int num_rqst, struct smb_rqst *rqst_array)
{
        struct smbd_connection *info = server->smbd_conn;
        struct smbdirect_socket *sc = &info->socket;
        struct smbdirect_socket_parameters *sp = &sc->parameters;
        struct smb_rqst *rqst;
        struct iov_iter iter;
        struct smbdirect_send_batch batch;
        unsigned int remaining_data_length, klen;
        int rc, i, rqst_idx;
        int error = 0;

        if (sc->status != SMBDIRECT_SOCKET_CONNECTED)
                return -EAGAIN;

        /*
         * Add in the page array if there is one. The caller needs to set
         * rq_tailsz to PAGE_SIZE when the buffer has multiple pages and
         * ends at page boundary
         */
        remaining_data_length = 0;
        for (i = 0; i < num_rqst; i++)
                remaining_data_length += smb_rqst_len(server, &rqst_array[i]);

        if (unlikely(remaining_data_length > sp->max_fragmented_send_size)) {
                /* assertion: payload never exceeds negotiated maximum */
                log_write(ERR, "payload size %d > max size %d\n",
                        remaining_data_length, sp->max_fragmented_send_size);
                return -EINVAL;
        }

        log_write(INFO, "num_rqst=%d total length=%u\n",
                        num_rqst, remaining_data_length);

        rqst_idx = 0;
        smbd_send_batch_init(&batch, false, 0);
        do {
                rqst = &rqst_array[rqst_idx];

                cifs_dbg(FYI, "Sending smb (RDMA): idx=%d smb_len=%lu\n",
                         rqst_idx, smb_rqst_len(server, rqst));
                for (i = 0; i < rqst->rq_nvec; i++)
                        dump_smb(rqst->rq_iov[i].iov_base, rqst->rq_iov[i].iov_len);

                log_write(INFO, "RDMA-WR[%u] nvec=%d len=%u iter=%zu rqlen=%lu\n",
                          rqst_idx, rqst->rq_nvec, remaining_data_length,
                          iov_iter_count(&rqst->rq_iter), smb_rqst_len(server, rqst));

                /* Send the metadata pages. */
                klen = 0;
                for (i = 0; i < rqst->rq_nvec; i++)
                        klen += rqst->rq_iov[i].iov_len;
                iov_iter_kvec(&iter, ITER_SOURCE, rqst->rq_iov, rqst->rq_nvec, klen);

                rc = smbd_post_send_full_iter(sc, &batch, &iter, &remaining_data_length);
                if (rc < 0) {
                        error = rc;
                        break;
                }

                if (iov_iter_count(&rqst->rq_iter) > 0) {
                        /* And then the data pages if there are any */
                        rc = smbd_post_send_full_iter(sc, &batch, &rqst->rq_iter,
                                                      &remaining_data_length);
                        if (rc < 0) {
                                error = rc;
                                break;
                        }
                }

        } while (++rqst_idx < num_rqst);

        rc = smbd_send_batch_flush(sc, &batch, true);
        if (unlikely(!rc && error))
                rc = error;

        /*
         * As an optimization, we don't wait for individual I/O to finish
         * before sending the next one.
         * Send them all and wait for pending send count to get to 0
         * that means all the I/Os have been out and we are good to return
         */

        wait_event(sc->send_io.pending.zero_wait_queue,
                atomic_read(&sc->send_io.pending.count) == 0 ||
                sc->status != SMBDIRECT_SOCKET_CONNECTED);

        if (sc->status != SMBDIRECT_SOCKET_CONNECTED && rc == 0)
                rc = -EAGAIN;

        return rc;
}

static void register_mr_done(struct ib_cq *cq, struct ib_wc *wc)
{
        struct smbdirect_mr_io *mr =
                container_of(wc->wr_cqe, struct smbdirect_mr_io, cqe);
        struct smbdirect_socket *sc = mr->socket;

        if (wc->status) {
                log_rdma_mr(ERR, "status=%d\n", wc->status);
                smbd_disconnect_rdma_connection(sc);
        }
}

/*
 * The work queue function that recovers MRs
 * We need to call ib_dereg_mr() and ib_alloc_mr() before this MR can be used
 * again. Both calls are slow, so finish them in a workqueue. This will not
 * block I/O path.
 * There is one workqueue that recovers MRs, there is no need to lock as the
 * I/O requests calling smbd_register_mr will never update the links in the
 * mr_list.
 */
static void smbd_mr_recovery_work(struct work_struct *work)
{
        struct smbdirect_socket *sc =
                container_of(work, struct smbdirect_socket, mr_io.recovery_work);
        struct smbdirect_socket_parameters *sp = &sc->parameters;
        struct smbdirect_mr_io *smbdirect_mr;
        int rc;

        list_for_each_entry(smbdirect_mr, &sc->mr_io.all.list, list) {
                if (smbdirect_mr->state == SMBDIRECT_MR_ERROR) {

                        /* recover this MR entry */
                        rc = ib_dereg_mr(smbdirect_mr->mr);
                        if (rc) {
                                log_rdma_mr(ERR,
                                        "ib_dereg_mr failed rc=%x\n",
                                        rc);
                                smbd_disconnect_rdma_connection(sc);
                                continue;
                        }

                        smbdirect_mr->mr = ib_alloc_mr(
                                sc->ib.pd, sc->mr_io.type,
                                sp->max_frmr_depth);
                        if (IS_ERR(smbdirect_mr->mr)) {
                                log_rdma_mr(ERR, "ib_alloc_mr failed mr_type=%x max_frmr_depth=%x\n",
                                            sc->mr_io.type,
                                            sp->max_frmr_depth);
                                smbd_disconnect_rdma_connection(sc);
                                continue;
                        }
                } else
                        /* This MR is being used, don't recover it */
                        continue;

                smbdirect_mr->state = SMBDIRECT_MR_READY;

                /* smbdirect_mr->state is updated by this function
                 * and is read and updated by I/O issuing CPUs trying
                 * to get a MR, the call to atomic_inc_return
                 * implicates a memory barrier and guarantees this
                 * value is updated before waking up any calls to
                 * get_mr() from the I/O issuing CPUs
                 */
                if (atomic_inc_return(&sc->mr_io.ready.count) == 1)
                        wake_up(&sc->mr_io.ready.wait_queue);
        }
}

static void smbd_mr_disable_locked(struct smbdirect_mr_io *mr)
{
        struct smbdirect_socket *sc = mr->socket;

        lockdep_assert_held(&mr->mutex);

        if (mr->state == SMBDIRECT_MR_DISABLED)
                return;

        if (mr->mr)
                ib_dereg_mr(mr->mr);
        if (mr->sgt.nents)
                ib_dma_unmap_sg(sc->ib.dev, mr->sgt.sgl, mr->sgt.nents, mr->dir);
        kfree(mr->sgt.sgl);

        mr->mr = NULL;
        mr->sgt.sgl = NULL;
        mr->sgt.nents = 0;

        mr->state = SMBDIRECT_MR_DISABLED;
}

static void smbd_mr_free_locked(struct kref *kref)
{
        struct smbdirect_mr_io *mr =
                container_of(kref, struct smbdirect_mr_io, kref);

        lockdep_assert_held(&mr->mutex);

        /*
         * smbd_mr_disable_locked() should already be called!
         */
        if (WARN_ON_ONCE(mr->state != SMBDIRECT_MR_DISABLED))
                smbd_mr_disable_locked(mr);

        mutex_unlock(&mr->mutex);
        mutex_destroy(&mr->mutex);
        kfree(mr);
}

static void destroy_mr_list(struct smbdirect_socket *sc)
{
        struct smbdirect_mr_io *mr, *tmp;
        LIST_HEAD(all_list);
        unsigned long flags;

        disable_work_sync(&sc->mr_io.recovery_work);

        spin_lock_irqsave(&sc->mr_io.all.lock, flags);
        list_splice_tail_init(&sc->mr_io.all.list, &all_list);
        spin_unlock_irqrestore(&sc->mr_io.all.lock, flags);

        list_for_each_entry_safe(mr, tmp, &all_list, list) {
                mutex_lock(&mr->mutex);

                smbd_mr_disable_locked(mr);
                list_del(&mr->list);
                mr->socket = NULL;

                /*
                 * No kref_put_mutex() as it's already locked.
                 *
                 * If smbd_mr_free_locked() is called
                 * and the mutex is unlocked and mr is gone,
                 * in that case kref_put() returned 1.
                 *
                 * If kref_put() returned 0 we know that
                 * smbd_mr_free_locked() didn't
                 * run. Not by us nor by anyone else, as we
                 * still hold the mutex, so we need to unlock.
                 *
                 * If the mr is still registered it will
                 * be dangling (detached from the connection
                 * waiting for smbd_deregister_mr() to be
                 * called in order to free the memory.
                 */
                if (!kref_put(&mr->kref, smbd_mr_free_locked))
                        mutex_unlock(&mr->mutex);
        }
}

/*
 * Allocate MRs used for RDMA read/write
 * The number of MRs will not exceed hardware capability in responder_resources
 * All MRs are kept in mr_list. The MR can be recovered after it's used
 * Recovery is done in smbd_mr_recovery_work. The content of list entry changes
 * as MRs are used and recovered for I/O, but the list links will not change
 */
static int allocate_mr_list(struct smbdirect_socket *sc)
{
        struct smbdirect_socket_parameters *sp = &sc->parameters;
        struct smbdirect_mr_io *mr;
        int ret;
        u32 i;

        if (sp->responder_resources == 0) {
                log_rdma_mr(ERR, "responder_resources negotiated as 0\n");
                return -EINVAL;
        }

        /* Allocate more MRs (2x) than hardware responder_resources */
        for (i = 0; i < sp->responder_resources * 2; i++) {
                mr = kzalloc_obj(*mr);
                if (!mr) {
                        ret = -ENOMEM;
                        goto kzalloc_mr_failed;
                }

                kref_init(&mr->kref);
                mutex_init(&mr->mutex);

                mr->mr = ib_alloc_mr(sc->ib.pd,
                                     sc->mr_io.type,
                                     sp->max_frmr_depth);
                if (IS_ERR(mr->mr)) {
                        ret = PTR_ERR(mr->mr);
                        log_rdma_mr(ERR, "ib_alloc_mr failed mr_type=%x max_frmr_depth=%x\n",
                                    sc->mr_io.type, sp->max_frmr_depth);
                        goto ib_alloc_mr_failed;
                }

                mr->sgt.sgl = kzalloc_objs(struct scatterlist,
                                           sp->max_frmr_depth);
                if (!mr->sgt.sgl) {
                        ret = -ENOMEM;
                        log_rdma_mr(ERR, "failed to allocate sgl\n");
                        goto kcalloc_sgl_failed;
                }
                mr->state = SMBDIRECT_MR_READY;
                mr->socket = sc;

                list_add_tail(&mr->list, &sc->mr_io.all.list);
                atomic_inc(&sc->mr_io.ready.count);
        }

        INIT_WORK(&sc->mr_io.recovery_work, smbd_mr_recovery_work);

        return 0;

kcalloc_sgl_failed:
        ib_dereg_mr(mr->mr);
ib_alloc_mr_failed:
        mutex_destroy(&mr->mutex);
        kfree(mr);
kzalloc_mr_failed:
        destroy_mr_list(sc);
        return ret;
}

/*
 * Get a MR from mr_list. This function waits until there is at least one
 * MR available in the list. It may access the list while the
 * smbd_mr_recovery_work is recovering the MR list. This doesn't need a lock
 * as they never modify the same places. However, there may be several CPUs
 * issuing I/O trying to get MR at the same time, mr_list_lock is used to
 * protect this situation.
 */
static struct smbdirect_mr_io *get_mr(struct smbdirect_socket *sc)
{
        struct smbdirect_mr_io *ret;
        unsigned long flags;
        int rc;
again:
        rc = wait_event_interruptible(sc->mr_io.ready.wait_queue,
                atomic_read(&sc->mr_io.ready.count) ||
                sc->status != SMBDIRECT_SOCKET_CONNECTED);
        if (rc) {
                log_rdma_mr(ERR, "wait_event_interruptible rc=%x\n", rc);
                return NULL;
        }

        if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
                log_rdma_mr(ERR, "sc->status=%x\n", sc->status);
                return NULL;
        }

        spin_lock_irqsave(&sc->mr_io.all.lock, flags);
        list_for_each_entry(ret, &sc->mr_io.all.list, list) {
                if (ret->state == SMBDIRECT_MR_READY) {
                        ret->state = SMBDIRECT_MR_REGISTERED;
                        kref_get(&ret->kref);
                        spin_unlock_irqrestore(&sc->mr_io.all.lock, flags);
                        atomic_dec(&sc->mr_io.ready.count);
                        atomic_inc(&sc->mr_io.used.count);
                        return ret;
                }
        }

        spin_unlock_irqrestore(&sc->mr_io.all.lock, flags);
        /*
         * It is possible that we could fail to get MR because other processes may
         * try to acquire a MR at the same time. If this is the case, retry it.
         */
        goto again;
}

/*
 * Transcribe the pages from an iterator into an MR scatterlist.
 */
static int smbd_iter_to_mr(struct iov_iter *iter,
                           struct sg_table *sgt,
                           unsigned int max_sg)
{
        int ret;

        memset(sgt->sgl, 0, max_sg * sizeof(struct scatterlist));

        ret = extract_iter_to_sg(iter, iov_iter_count(iter), sgt, max_sg, 0);
        WARN_ON(ret < 0);
        if (sgt->nents > 0)
                sg_mark_end(&sgt->sgl[sgt->nents - 1]);
        return ret;
}

/*
 * Register memory for RDMA read/write
 * iter: the buffer to register memory with
 * writing: true if this is a RDMA write (SMB read), false for RDMA read
 * need_invalidate: true if this MR needs to be locally invalidated after I/O
 * return value: the MR registered, NULL if failed.
 */
struct smbdirect_mr_io *smbd_register_mr(struct smbd_connection *info,
                                 struct iov_iter *iter,
                                 bool writing, bool need_invalidate)
{
        struct smbdirect_socket *sc = &info->socket;
        struct smbdirect_socket_parameters *sp = &sc->parameters;
        struct smbdirect_mr_io *mr;
        int rc, num_pages;
        struct ib_reg_wr *reg_wr;

        num_pages = iov_iter_npages(iter, sp->max_frmr_depth + 1);
        if (num_pages > sp->max_frmr_depth) {
                log_rdma_mr(ERR, "num_pages=%d max_frmr_depth=%d\n",
                        num_pages, sp->max_frmr_depth);
                WARN_ON_ONCE(1);
                return NULL;
        }

        mr = get_mr(sc);
        if (!mr) {
                log_rdma_mr(ERR, "get_mr returning NULL\n");
                return NULL;
        }

        mutex_lock(&mr->mutex);

        mr->dir = writing ? DMA_FROM_DEVICE : DMA_TO_DEVICE;
        mr->need_invalidate = need_invalidate;
        mr->sgt.nents = 0;
        mr->sgt.orig_nents = 0;

        log_rdma_mr(INFO, "num_pages=0x%x count=0x%zx depth=%u\n",
                    num_pages, iov_iter_count(iter), sp->max_frmr_depth);
        smbd_iter_to_mr(iter, &mr->sgt, sp->max_frmr_depth);

        rc = ib_dma_map_sg(sc->ib.dev, mr->sgt.sgl, mr->sgt.nents, mr->dir);
        if (!rc) {
                log_rdma_mr(ERR, "ib_dma_map_sg num_pages=%x dir=%x rc=%x\n",
                            num_pages, mr->dir, rc);
                goto dma_map_error;
        }

        rc = ib_map_mr_sg(mr->mr, mr->sgt.sgl, mr->sgt.nents, NULL, PAGE_SIZE);
        if (rc != mr->sgt.nents) {
                log_rdma_mr(ERR,
                            "ib_map_mr_sg failed rc = %d nents = %x\n",
                            rc, mr->sgt.nents);
                goto map_mr_error;
        }

        ib_update_fast_reg_key(mr->mr, ib_inc_rkey(mr->mr->rkey));
        reg_wr = &mr->wr;
        reg_wr->wr.opcode = IB_WR_REG_MR;
        mr->cqe.done = register_mr_done;
        reg_wr->wr.wr_cqe = &mr->cqe;
        reg_wr->wr.num_sge = 0;
        reg_wr->wr.send_flags = IB_SEND_SIGNALED;
        reg_wr->mr = mr->mr;
        reg_wr->key = mr->mr->rkey;
        reg_wr->access = writing ?
                        IB_ACCESS_REMOTE_WRITE | IB_ACCESS_LOCAL_WRITE :
                        IB_ACCESS_REMOTE_READ;

        /*
         * There is no need for waiting for complemtion on ib_post_send
         * on IB_WR_REG_MR. Hardware enforces a barrier and order of execution
         * on the next ib_post_send when we actually send I/O to remote peer
         */
        rc = ib_post_send(sc->ib.qp, &reg_wr->wr, NULL);
        if (!rc) {
                /*
                 * get_mr() gave us a reference
                 * via kref_get(&mr->kref), we keep that and let
                 * the caller use smbd_deregister_mr()
                 * to remove it again.
                 */
                mutex_unlock(&mr->mutex);
                return mr;
        }

        log_rdma_mr(ERR, "ib_post_send failed rc=%x reg_wr->key=%x\n",
                rc, reg_wr->key);

        /* If all failed, attempt to recover this MR by setting it SMBDIRECT_MR_ERROR*/
map_mr_error:
        ib_dma_unmap_sg(sc->ib.dev, mr->sgt.sgl, mr->sgt.nents, mr->dir);

dma_map_error:
        mr->sgt.nents = 0;
        mr->state = SMBDIRECT_MR_ERROR;
        if (atomic_dec_and_test(&sc->mr_io.used.count))
                wake_up(&sc->mr_io.cleanup.wait_queue);

        smbd_disconnect_rdma_connection(sc);

        /*
         * get_mr() gave us a reference
         * via kref_get(&mr->kref), we need to remove it again
         * on error.
         *
         * No kref_put_mutex() as it's already locked.
         *
         * If smbd_mr_free_locked() is called
         * and the mutex is unlocked and mr is gone,
         * in that case kref_put() returned 1.
         *
         * If kref_put() returned 0 we know that
         * smbd_mr_free_locked() didn't
         * run. Not by us nor by anyone else, as we
         * still hold the mutex, so we need to unlock.
         */
        if (!kref_put(&mr->kref, smbd_mr_free_locked))
                mutex_unlock(&mr->mutex);

        return NULL;
}

static void local_inv_done(struct ib_cq *cq, struct ib_wc *wc)
{
        struct smbdirect_mr_io *smbdirect_mr;
        struct ib_cqe *cqe;

        cqe = wc->wr_cqe;
        smbdirect_mr = container_of(cqe, struct smbdirect_mr_io, cqe);
        smbdirect_mr->state = SMBDIRECT_MR_INVALIDATED;
        if (wc->status != IB_WC_SUCCESS) {
                log_rdma_mr(ERR, "invalidate failed status=%x\n", wc->status);
                smbdirect_mr->state = SMBDIRECT_MR_ERROR;
        }
        complete(&smbdirect_mr->invalidate_done);
}

/*
 * Deregister a MR after I/O is done
 * This function may wait if remote invalidation is not used
 * and we have to locally invalidate the buffer to prevent data is being
 * modified by remote peer after upper layer consumes it
 */
void smbd_deregister_mr(struct smbdirect_mr_io *mr)
{
        struct smbdirect_socket *sc = mr->socket;

        mutex_lock(&mr->mutex);
        if (mr->state == SMBDIRECT_MR_DISABLED)
                goto put_kref;

        if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
                smbd_mr_disable_locked(mr);
                goto put_kref;
        }

        if (mr->need_invalidate) {
                struct ib_send_wr *wr = &mr->inv_wr;
                int rc;

                /* Need to finish local invalidation before returning */
                wr->opcode = IB_WR_LOCAL_INV;
                mr->cqe.done = local_inv_done;
                wr->wr_cqe = &mr->cqe;
                wr->num_sge = 0;
                wr->ex.invalidate_rkey = mr->mr->rkey;
                wr->send_flags = IB_SEND_SIGNALED;

                init_completion(&mr->invalidate_done);
                rc = ib_post_send(sc->ib.qp, wr, NULL);
                if (rc) {
                        log_rdma_mr(ERR, "ib_post_send failed rc=%x\n", rc);
                        smbd_mr_disable_locked(mr);
                        smbd_disconnect_rdma_connection(sc);
                        goto done;
                }
                wait_for_completion(&mr->invalidate_done);
                mr->need_invalidate = false;
        } else
                /*
                 * For remote invalidation, just set it to SMBDIRECT_MR_INVALIDATED
                 * and defer to mr_recovery_work to recover the MR for next use
                 */
                mr->state = SMBDIRECT_MR_INVALIDATED;

        if (mr->sgt.nents) {
                ib_dma_unmap_sg(sc->ib.dev, mr->sgt.sgl, mr->sgt.nents, mr->dir);
                mr->sgt.nents = 0;
        }

        if (mr->state == SMBDIRECT_MR_INVALIDATED) {
                mr->state = SMBDIRECT_MR_READY;
                if (atomic_inc_return(&sc->mr_io.ready.count) == 1)
                        wake_up(&sc->mr_io.ready.wait_queue);
        } else
                /*
                 * Schedule the work to do MR recovery for future I/Os MR
                 * recovery is slow and don't want it to block current I/O
                 */
                queue_work(sc->workqueue, &sc->mr_io.recovery_work);

done:
        if (atomic_dec_and_test(&sc->mr_io.used.count))
                wake_up(&sc->mr_io.cleanup.wait_queue);

put_kref:
        /*
         * No kref_put_mutex() as it's already locked.
         *
         * If smbd_mr_free_locked() is called
         * and the mutex is unlocked and mr is gone,
         * in that case kref_put() returned 1.
         *
         * If kref_put() returned 0 we know that
         * smbd_mr_free_locked() didn't
         * run. Not by us nor by anyone else, as we
         * still hold the mutex, so we need to unlock
         * and keep the mr in SMBDIRECT_MR_READY or
         * SMBDIRECT_MR_ERROR state.
         */
        if (!kref_put(&mr->kref, smbd_mr_free_locked))
                mutex_unlock(&mr->mutex);
}

static bool smb_set_sge(struct smb_extract_to_rdma *rdma,
                        struct page *lowest_page, size_t off, size_t len)
{
        struct ib_sge *sge = &rdma->sge[rdma->nr_sge];
        u64 addr;

        addr = ib_dma_map_page(rdma->device, lowest_page,
                               off, len, rdma->direction);
        if (ib_dma_mapping_error(rdma->device, addr))
                return false;

        sge->addr   = addr;
        sge->length = len;
        sge->lkey   = rdma->local_dma_lkey;
        rdma->nr_sge++;
        return true;
}

/*
 * Extract page fragments from a BVEC-class iterator and add them to an RDMA
 * element list.  The pages are not pinned.
 */
static ssize_t smb_extract_bvec_to_rdma(struct iov_iter *iter,
                                        struct smb_extract_to_rdma *rdma,
                                        ssize_t maxsize)
{
        const struct bio_vec *bv = iter->bvec;
        unsigned long start = iter->iov_offset;
        unsigned int i;
        ssize_t ret = 0;

        for (i = 0; i < iter->nr_segs; i++) {
                size_t off, len;

                len = bv[i].bv_len;
                if (start >= len) {
                        start -= len;
                        continue;
                }

                len = min_t(size_t, maxsize, len - start);
                off = bv[i].bv_offset + start;

                if (!smb_set_sge(rdma, bv[i].bv_page, off, len))
                        return -EIO;

                ret += len;
                maxsize -= len;
                if (rdma->nr_sge >= rdma->max_sge || maxsize <= 0)
                        break;
                start = 0;
        }

        if (ret > 0)
                iov_iter_advance(iter, ret);
        return ret;
}

/*
 * Extract fragments from a KVEC-class iterator and add them to an RDMA list.
 * This can deal with vmalloc'd buffers as well as kmalloc'd or static buffers.
 * The pages are not pinned.
 */
static ssize_t smb_extract_kvec_to_rdma(struct iov_iter *iter,
                                        struct smb_extract_to_rdma *rdma,
                                        ssize_t maxsize)
{
        const struct kvec *kv = iter->kvec;
        unsigned long start = iter->iov_offset;
        unsigned int i;
        ssize_t ret = 0;

        for (i = 0; i < iter->nr_segs; i++) {
                struct page *page;
                unsigned long kaddr;
                size_t off, len, seg;

                len = kv[i].iov_len;
                if (start >= len) {
                        start -= len;
                        continue;
                }

                kaddr = (unsigned long)kv[i].iov_base + start;
                off = kaddr & ~PAGE_MASK;
                len = min_t(size_t, maxsize, len - start);
                kaddr &= PAGE_MASK;

                maxsize -= len;
                do {
                        seg = min_t(size_t, len, PAGE_SIZE - off);

                        if (is_vmalloc_or_module_addr((void *)kaddr))
                                page = vmalloc_to_page((void *)kaddr);
                        else
                                page = virt_to_page((void *)kaddr);

                        if (!smb_set_sge(rdma, page, off, seg))
                                return -EIO;

                        ret += seg;
                        len -= seg;
                        kaddr += PAGE_SIZE;
                        off = 0;
                } while (len > 0 && rdma->nr_sge < rdma->max_sge);

                if (rdma->nr_sge >= rdma->max_sge || maxsize <= 0)
                        break;
                start = 0;
        }

        if (ret > 0)
                iov_iter_advance(iter, ret);
        return ret;
}

/*
 * Extract folio fragments from a FOLIOQ-class iterator and add them to an RDMA
 * list.  The folios are not pinned.
 */
static ssize_t smb_extract_folioq_to_rdma(struct iov_iter *iter,
                                          struct smb_extract_to_rdma *rdma,
                                          ssize_t maxsize)
{
        const struct folio_queue *folioq = iter->folioq;
        unsigned int slot = iter->folioq_slot;
        ssize_t ret = 0;
        size_t offset = iter->iov_offset;

        BUG_ON(!folioq);

        if (slot >= folioq_nr_slots(folioq)) {
                folioq = folioq->next;
                if (WARN_ON_ONCE(!folioq))
                        return -EIO;
                slot = 0;
        }

        do {
                struct folio *folio = folioq_folio(folioq, slot);
                size_t fsize = folioq_folio_size(folioq, slot);

                if (offset < fsize) {
                        size_t part = umin(maxsize, fsize - offset);

                        if (!smb_set_sge(rdma, folio_page(folio, 0), offset, part))
                                return -EIO;

                        offset += part;
                        ret += part;
                        maxsize -= part;
                }

                if (offset >= fsize) {
                        offset = 0;
                        slot++;
                        if (slot >= folioq_nr_slots(folioq)) {
                                if (!folioq->next) {
                                        WARN_ON_ONCE(ret < iter->count);
                                        break;
                                }
                                folioq = folioq->next;
                                slot = 0;
                        }
                }
        } while (rdma->nr_sge < rdma->max_sge && maxsize > 0);

        iter->folioq = folioq;
        iter->folioq_slot = slot;
        iter->iov_offset = offset;
        iter->count -= ret;
        return ret;
}

/*
 * Extract page fragments from up to the given amount of the source iterator
 * and build up an RDMA list that refers to all of those bits.  The RDMA list
 * is appended to, up to the maximum number of elements set in the parameter
 * block.
 *
 * The extracted page fragments are not pinned or ref'd in any way; if an
 * IOVEC/UBUF-type iterator is to be used, it should be converted to a
 * BVEC-type iterator and the pages pinned, ref'd or otherwise held in some
 * way.
 */
static ssize_t smb_extract_iter_to_rdma(struct iov_iter *iter, size_t len,
                                        struct smb_extract_to_rdma *rdma)
{
        ssize_t ret;
        int before = rdma->nr_sge;

        switch (iov_iter_type(iter)) {
        case ITER_BVEC:
                ret = smb_extract_bvec_to_rdma(iter, rdma, len);
                break;
        case ITER_KVEC:
                ret = smb_extract_kvec_to_rdma(iter, rdma, len);
                break;
        case ITER_FOLIOQ:
                ret = smb_extract_folioq_to_rdma(iter, rdma, len);
                break;
        default:
                WARN_ON_ONCE(1);
                return -EIO;
        }

        if (ret < 0) {
                while (rdma->nr_sge > before) {
                        struct ib_sge *sge = &rdma->sge[rdma->nr_sge--];

                        ib_dma_unmap_single(rdma->device, sge->addr, sge->length,
                                            rdma->direction);
                        sge->addr = 0;
                }
        }

        return ret;
}