// SPDX-License-Identifier: GPL-2.0-or-later
/*
 *   Copyright (C) 2017, Microsoft Corporation.
 *   Copyright (c) 2025, Stefan Metzmacher
 */

#include "internal.h"
#include <linux/folio_queue.h>

struct smbdirect_map_sges {
        struct ib_sge *sge;
        size_t num_sge;
        size_t max_sge;
        struct ib_device *device;
        u32 local_dma_lkey;
        enum dma_data_direction direction;
};

static ssize_t smbdirect_map_sges_from_iter(struct iov_iter *iter, size_t len,
                                            struct smbdirect_map_sges *state);

static void smbdirect_connection_recv_io_refill_work(struct work_struct *work);
static void smbdirect_connection_send_immediate_work(struct work_struct *work);

static void smbdirect_connection_qp_event_handler(struct ib_event *event, void *context)
{
        struct smbdirect_socket *sc = context;

        smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
                "%s on device %.*s socket %p (cm_id=%p) status %s first_error %1pe\n",
                ib_event_msg(event->event),
                IB_DEVICE_NAME_MAX,
                event->device->name,
                sc, sc->rdma.cm_id,
                smbdirect_socket_status_string(sc->status),
                SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));

        switch (event->event) {
        case IB_EVENT_CQ_ERR:
        case IB_EVENT_QP_FATAL:
                smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
                break;

        default:
                break;
        }
}

static int smbdirect_connection_rdma_event_handler(struct rdma_cm_id *id,
                                                   struct rdma_cm_event *event)
{
        struct smbdirect_socket *sc = id->context;
        int ret = -ECONNRESET;

        if (event->event == RDMA_CM_EVENT_DEVICE_REMOVAL)
                ret = -ENETDOWN;
        if (IS_ERR(SMBDIRECT_DEBUG_ERR_PTR(event->status)))
                ret = event->status;

        /*
         * cma_cm_event_handler() has
         * lockdep_assert_held(&id_priv->handler_mutex);
         *
         * Mutexes are not allowed in interrupts,
         * and we rely on not being in an interrupt here.
         */
        WARN_ON_ONCE(in_interrupt());

        if (event->event != sc->rdma.expected_event) {
                smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
                        "%s (first_error=%1pe, expected=%s) => event=%s status=%d => ret=%1pe\n",
                        smbdirect_socket_status_string(sc->status),
                        SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
                        rdma_event_msg(sc->rdma.expected_event),
                        rdma_event_msg(event->event),
                        event->status,
                        SMBDIRECT_DEBUG_ERR_PTR(ret));

                /*
                 * If we get RDMA_CM_EVENT_DEVICE_REMOVAL,
                 * we should change to SMBDIRECT_SOCKET_DISCONNECTED,
                 * so that rdma_disconnect() is avoided later via
                 * smbdirect_socket_schedule_cleanup[_status]() =>
                 * smbdirect_socket_cleanup_work().
                 *
                 * As otherwise we'd set SMBDIRECT_SOCKET_DISCONNECTING,
                 * but never ever get RDMA_CM_EVENT_DISCONNECTED and
                 * never reach SMBDIRECT_SOCKET_DISCONNECTED.
                 */
                if (event->event == RDMA_CM_EVENT_DEVICE_REMOVAL)
                        smbdirect_socket_schedule_cleanup_status(sc,
                                                                 SMBDIRECT_LOG_ERR,
                                                                 ret,
                                                                 SMBDIRECT_SOCKET_DISCONNECTED);
                else
                        smbdirect_socket_schedule_cleanup(sc, ret);
                if (sc->ib.qp)
                        ib_drain_qp(sc->ib.qp);
                return 0;
        }

        smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
                "%s (first_error=%1pe) event=%s\n",
                smbdirect_socket_status_string(sc->status),
                SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
                rdma_event_msg(event->event));

        switch (event->event) {
        case RDMA_CM_EVENT_DISCONNECTED:
                /*
                 * We need to change to SMBDIRECT_SOCKET_DISCONNECTED,
                 * so that rdma_disconnect() is avoided later via
                 * smbdirect_socket_schedule_cleanup_status() =>
                 * smbdirect_socket_cleanup_work().
                 *
                 * As otherwise we'd set SMBDIRECT_SOCKET_DISCONNECTING,
                 * but never ever get RDMA_CM_EVENT_DISCONNECTED and
                 * never reach SMBDIRECT_SOCKET_DISCONNECTED.
                 *
                 * This is also a normal disconnect so
                 * SMBDIRECT_LOG_INFO should be good enough
                 * and avoids spamming the default logs.
                 */
                smbdirect_socket_schedule_cleanup_status(sc,
                                                         SMBDIRECT_LOG_INFO,
                                                         ret,
                                                         SMBDIRECT_SOCKET_DISCONNECTED);
                if (sc->ib.qp)
                        ib_drain_qp(sc->ib.qp);
                return 0;

        default:
                break;
        }

        /*
         * This is an internal error, should be handled above via
         * event->event != sc->rdma.expected_event already.
         */
        WARN_ON_ONCE(sc->rdma.expected_event != RDMA_CM_EVENT_DISCONNECTED);
        smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
        return 0;
}

void smbdirect_connection_rdma_established(struct smbdirect_socket *sc)
{
        smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
                "rdma established: device: %.*s local: %pISpsfc remote: %pISpsfc\n",
                IB_DEVICE_NAME_MAX,
                sc->ib.dev->name,
                &sc->rdma.cm_id->route.addr.src_addr,
                &sc->rdma.cm_id->route.addr.dst_addr);

        sc->rdma.cm_id->event_handler = smbdirect_connection_rdma_event_handler;
        sc->rdma.expected_event = RDMA_CM_EVENT_DISCONNECTED;
}

void smbdirect_connection_negotiation_done(struct smbdirect_socket *sc)
{
        if (unlikely(sc->first_error))
                return;

        if (sc->status == SMBDIRECT_SOCKET_CONNECTED)
                /*
                 * This is the accept case where
                 * smbdirect_socket_accept() already sets
                 * SMBDIRECT_SOCKET_CONNECTED
                 */
                goto done;

        if (sc->status != SMBDIRECT_SOCKET_NEGOTIATE_RUNNING) {
                /*
                 * Something went wrong...
                 */
                smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
                        "status=%s first_error=%1pe local: %pISpsfc remote: %pISpsfc\n",
                        smbdirect_socket_status_string(sc->status),
                        SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
                        &sc->rdma.cm_id->route.addr.src_addr,
                        &sc->rdma.cm_id->route.addr.dst_addr);
                return;
        }

        /*
         * We are done, so we can wake up the waiter.
         */
        WARN_ONCE(sc->status == SMBDIRECT_SOCKET_CONNECTED,
                  "status=%s first_error=%1pe",
                  smbdirect_socket_status_string(sc->status),
                  SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
        sc->status = SMBDIRECT_SOCKET_CONNECTED;

        /*
         * We need to setup the refill and send immediate work
         * in order to get a working connection.
         */
done:
        INIT_WORK(&sc->recv_io.posted.refill_work, smbdirect_connection_recv_io_refill_work);
        INIT_WORK(&sc->idle.immediate_work, smbdirect_connection_send_immediate_work);

        smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
                "negotiated: local: %pISpsfc remote: %pISpsfc\n",
                &sc->rdma.cm_id->route.addr.src_addr,
                &sc->rdma.cm_id->route.addr.dst_addr);

        wake_up(&sc->status_wait);
}

static u32 smbdirect_rdma_rw_send_wrs(struct ib_device *dev,
                                      const struct ib_qp_init_attr *attr)
{
        /*
         * This could be split out of rdma_rw_init_qp()
         * and be a helper function next to rdma_rw_mr_factor()
         *
         * We can't check unlikely(rdma_rw_force_mr) here,
         * but that is most likely 0 anyway.
         */
        u32 factor;

        WARN_ON_ONCE(attr->port_num == 0);

        /*
         * Each context needs at least one RDMA READ or WRITE WR.
         *
         * For some hardware we might need more, eventually we should ask the
         * HCA driver for a multiplier here.
         */
        factor = 1;

        /*
         * If the device needs MRs to perform RDMA READ or WRITE operations,
         * we'll need two additional MRs for the registrations and the
         * invalidation.
         */
        if (rdma_protocol_iwarp(dev, attr->port_num) || dev->attrs.max_sgl_rd)
                factor += 2;    /* inv + reg */

        return factor * attr->cap.max_rdma_ctxs;
}

int smbdirect_connection_create_qp(struct smbdirect_socket *sc)
{
        const struct smbdirect_socket_parameters *sp = &sc->parameters;
        struct ib_qp_init_attr qp_attr;
        struct ib_qp_cap qp_cap;
        u32 rdma_send_wr;
        u32 max_send_wr;
        int ret;

        /*
         * Note that {rdma,ib}_create_qp() will call
         * rdma_rw_init_qp() if max_rdma_ctxs is not 0.
         * It will adjust max_send_wr to the required
         * number of additional WRs for the RDMA RW operations.
         * It will cap max_send_wr to the device limit.
         *
         * 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;
        qp_cap.max_inline_data = 0;
        qp_cap.max_rdma_ctxs = sc->rw_io.credits.max;

        /*
         * Find out the number of max_send_wr
         * after rdma_rw_init_qp() adjusted it.
         *
         * We only do it on a temporary variable,
         * as rdma_create_qp() will trigger
         * rdma_rw_init_qp() again.
         */
        memset(&qp_attr, 0, sizeof(qp_attr));
        qp_attr.cap = qp_cap;
        qp_attr.port_num = sc->rdma.cm_id->port_num;
        rdma_send_wr = smbdirect_rdma_rw_send_wrs(sc->ib.dev, &qp_attr);
        max_send_wr = qp_cap.max_send_wr + rdma_send_wr;

        if (qp_cap.max_send_wr > sc->ib.dev->attrs.max_cqe ||
            qp_cap.max_send_wr > sc->ib.dev->attrs.max_qp_wr) {
                pr_err("Possible CQE overrun: max_send_wr %d\n",
                       qp_cap.max_send_wr);
                pr_err("device %.*s reporting max_cqe %d max_qp_wr %d\n",
                       IB_DEVICE_NAME_MAX,
                       sc->ib.dev->name,
                       sc->ib.dev->attrs.max_cqe,
                       sc->ib.dev->attrs.max_qp_wr);
                pr_err("consider lowering send_credit_target = %d\n",
                       sp->send_credit_target);
                return -EINVAL;
        }

        if (qp_cap.max_rdma_ctxs &&
            (max_send_wr >= sc->ib.dev->attrs.max_cqe ||
             max_send_wr >= sc->ib.dev->attrs.max_qp_wr)) {
                pr_err("Possible CQE overrun: rdma_send_wr %d + max_send_wr %d = %d\n",
                       rdma_send_wr, qp_cap.max_send_wr, max_send_wr);
                pr_err("device %.*s reporting max_cqe %d max_qp_wr %d\n",
                       IB_DEVICE_NAME_MAX,
                       sc->ib.dev->name,
                       sc->ib.dev->attrs.max_cqe,
                       sc->ib.dev->attrs.max_qp_wr);
                pr_err("consider lowering send_credit_target = %d, max_rdma_ctxs = %d\n",
                       sp->send_credit_target, qp_cap.max_rdma_ctxs);
                return -EINVAL;
        }

        if (qp_cap.max_recv_wr > sc->ib.dev->attrs.max_cqe ||
            qp_cap.max_recv_wr > sc->ib.dev->attrs.max_qp_wr) {
                pr_err("Possible CQE overrun: max_recv_wr %d\n",
                       qp_cap.max_recv_wr);
                pr_err("device %.*s reporting max_cqe %d max_qp_wr %d\n",
                       IB_DEVICE_NAME_MAX,
                       sc->ib.dev->name,
                       sc->ib.dev->attrs.max_cqe,
                       sc->ib.dev->attrs.max_qp_wr);
                pr_err("consider lowering receive_credit_max = %d\n",
                       sp->recv_credit_max);
                return -EINVAL;
        }

        if (qp_cap.max_send_sge > sc->ib.dev->attrs.max_send_sge ||
            qp_cap.max_recv_sge > sc->ib.dev->attrs.max_recv_sge) {
                pr_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);
                return -EINVAL;
        }

        sc->ib.pd = ib_alloc_pd(sc->ib.dev, 0);
        if (IS_ERR(sc->ib.pd)) {
                pr_err("Can't create RDMA PD: %1pe\n", sc->ib.pd);
                ret = PTR_ERR(sc->ib.pd);
                sc->ib.pd = NULL;
                return ret;
        }

        sc->ib.send_cq = ib_alloc_cq_any(sc->ib.dev, sc,
                                         max_send_wr,
                                         sc->ib.poll_ctx);
        if (IS_ERR(sc->ib.send_cq)) {
                pr_err("Can't create RDMA send CQ: %1pe\n", sc->ib.send_cq);
                ret = PTR_ERR(sc->ib.send_cq);
                sc->ib.send_cq = NULL;
                goto err;
        }

        sc->ib.recv_cq = ib_alloc_cq_any(sc->ib.dev, sc,
                                         qp_cap.max_recv_wr,
                                         sc->ib.poll_ctx);
        if (IS_ERR(sc->ib.recv_cq)) {
                pr_err("Can't create RDMA recv CQ: %1pe\n", sc->ib.recv_cq);
                ret = PTR_ERR(sc->ib.recv_cq);
                sc->ib.recv_cq = NULL;
                goto err;
        }

        /*
         * We reset completely here!
         * As the above use was just temporary
         * to calc max_send_wr and rdma_send_wr.
         *
         * rdma_create_qp() will trigger rdma_rw_init_qp()
         * again if max_rdma_ctxs is not 0.
         */
        memset(&qp_attr, 0, sizeof(qp_attr));
        qp_attr.event_handler = smbdirect_connection_qp_event_handler;
        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;

        ret = rdma_create_qp(sc->rdma.cm_id, sc->ib.pd, &qp_attr);
        if (ret) {
                pr_err("Can't create RDMA QP: %1pe\n",
                       SMBDIRECT_DEBUG_ERR_PTR(ret));
                goto err;
        }
        sc->ib.qp = sc->rdma.cm_id->qp;

        return 0;
err:
        smbdirect_connection_destroy_qp(sc);
        return ret;
}

void smbdirect_connection_destroy_qp(struct smbdirect_socket *sc)
{
        if (sc->ib.qp) {
                ib_drain_qp(sc->ib.qp);
                sc->ib.qp = NULL;
                rdma_destroy_qp(sc->rdma.cm_id);
        }
        if (sc->ib.recv_cq) {
                ib_destroy_cq(sc->ib.recv_cq);
                sc->ib.recv_cq = NULL;
        }
        if (sc->ib.send_cq) {
                ib_destroy_cq(sc->ib.send_cq);
                sc->ib.send_cq = NULL;
        }
        if (sc->ib.pd) {
                ib_dealloc_pd(sc->ib.pd);
                sc->ib.pd = NULL;
        }
}

int smbdirect_connection_create_mem_pools(struct smbdirect_socket *sc)
{
        const struct smbdirect_socket_parameters *sp = &sc->parameters;
        char name[80];
        size_t i;

        /*
         * We use sizeof(struct smbdirect_negotiate_resp) for the
         * payload size as it is larger as
         * sizeof(struct smbdirect_data_transfer).
         *
         * This will fit client and server usage for now.
         */
        snprintf(name, sizeof(name), "smbdirect_send_io_cache_%p", sc);
        struct kmem_cache_args send_io_args = {
                .align          = __alignof__(struct smbdirect_send_io),
        };
        sc->send_io.mem.cache = kmem_cache_create(name,
                                                  sizeof(struct smbdirect_send_io) +
                                                  sizeof(struct smbdirect_negotiate_resp),
                                                  &send_io_args,
                                                  SLAB_HWCACHE_ALIGN);
        if (!sc->send_io.mem.cache)
                goto err;

        sc->send_io.mem.pool = mempool_create_slab_pool(sp->send_credit_target,
                                                        sc->send_io.mem.cache);
        if (!sc->send_io.mem.pool)
                goto err;

        /*
         * A payload size of sp->max_recv_size should fit
         * any message.
         *
         * For smbdirect_data_transfer messages the whole
         * buffer might be exposed to userspace
         * (currently on the client side...)
         * The documentation says data_offset = 0 would be
         * strange but valid.
         */
        snprintf(name, sizeof(name), "smbdirect_recv_io_cache_%p", sc);
        struct kmem_cache_args recv_io_args = {
                .align          = __alignof__(struct smbdirect_recv_io),
                .useroffset     = sizeof(struct smbdirect_recv_io),
                .usersize       = sp->max_recv_size,
        };
        sc->recv_io.mem.cache = kmem_cache_create(name,
                                                  sizeof(struct smbdirect_recv_io) +
                                                  sp->max_recv_size,
                                                  &recv_io_args,
                                                  SLAB_HWCACHE_ALIGN);
        if (!sc->recv_io.mem.cache)
                goto err;

        sc->recv_io.mem.pool = mempool_create_slab_pool(sp->recv_credit_max,
                                                        sc->recv_io.mem.cache);
        if (!sc->recv_io.mem.pool)
                goto err;

        for (i = 0; i < sp->recv_credit_max; i++) {
                struct smbdirect_recv_io *recv_io;

                recv_io = mempool_alloc(sc->recv_io.mem.pool,
                                        sc->recv_io.mem.gfp_mask);
                if (!recv_io)
                        goto err;
                recv_io->socket = sc;
                recv_io->sge.length = 0;
                list_add_tail(&recv_io->list, &sc->recv_io.free.list);
        }

        return 0;
err:
        smbdirect_connection_destroy_mem_pools(sc);
        return -ENOMEM;
}

void smbdirect_connection_destroy_mem_pools(struct smbdirect_socket *sc)
{
        struct smbdirect_recv_io *recv_io, *next_io;

        list_for_each_entry_safe(recv_io, next_io, &sc->recv_io.free.list, list) {
                list_del(&recv_io->list);
                mempool_free(recv_io, sc->recv_io.mem.pool);
        }

        /*
         * Note mempool_destroy() and kmem_cache_destroy()
         * work fine with a NULL pointer
         */

        mempool_destroy(sc->recv_io.mem.pool);
        sc->recv_io.mem.pool = NULL;

        kmem_cache_destroy(sc->recv_io.mem.cache);
        sc->recv_io.mem.cache = NULL;

        mempool_destroy(sc->send_io.mem.pool);
        sc->send_io.mem.pool = NULL;

        kmem_cache_destroy(sc->send_io.mem.cache);
        sc->send_io.mem.cache = NULL;
}

struct smbdirect_send_io *smbdirect_connection_alloc_send_io(struct smbdirect_socket *sc)
{
        struct smbdirect_send_io *msg;

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

        return msg;
}

void smbdirect_connection_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);
}

struct smbdirect_recv_io *smbdirect_connection_get_recv_io(struct smbdirect_socket *sc)
{
        struct smbdirect_recv_io *msg = NULL;
        unsigned long flags;

        spin_lock_irqsave(&sc->recv_io.free.lock, flags);
        if (likely(!sc->first_error))
                msg = list_first_entry_or_null(&sc->recv_io.free.list,
                                               struct smbdirect_recv_io,
                                               list);
        if (likely(msg)) {
                list_del(&msg->list);
                sc->statistics.get_receive_buffer++;
        }
        spin_unlock_irqrestore(&sc->recv_io.free.lock, flags);

        return msg;
}

void smbdirect_connection_put_recv_io(struct smbdirect_recv_io *msg)
{
        struct smbdirect_socket *sc = msg->socket;
        unsigned long flags;

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

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

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

void smbdirect_connection_reassembly_append_recv_io(struct smbdirect_socket *sc,
                                                    struct smbdirect_recv_io *msg,
                                                    u32 data_length)
{
        unsigned long flags;

        spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
        list_add_tail(&msg->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++;
}

struct smbdirect_recv_io *
smbdirect_connection_reassembly_first_recv_io(struct smbdirect_socket *sc)
{
        struct smbdirect_recv_io *msg;

        msg = list_first_entry_or_null(&sc->recv_io.reassembly.list,
                                       struct smbdirect_recv_io,
                                       list);

        return msg;
}

void smbdirect_connection_negotiate_rdma_resources(struct smbdirect_socket *sc,
                                                   u8 peer_initiator_depth,
                                                   u8 peer_responder_resources,
                                                   const struct rdma_conn_param *param)
{
        struct smbdirect_socket_parameters *sp = &sc->parameters;

        if (rdma_protocol_iwarp(sc->ib.dev, sc->rdma.cm_id->port_num) &&
            param->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 = param->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)param->initiator_depth != ird32 ||
                    (u32)param->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);
}

bool smbdirect_connection_is_connected(struct smbdirect_socket *sc)
{
        if (unlikely(!sc || sc->first_error || sc->status != SMBDIRECT_SOCKET_CONNECTED))
                return false;
        return true;
}
EXPORT_SYMBOL_GPL(smbdirect_connection_is_connected);

int smbdirect_connection_wait_for_connected(struct smbdirect_socket *sc)
{
        const struct smbdirect_socket_parameters *sp = &sc->parameters;
        union {
                struct sockaddr sa;
                struct sockaddr_storage ss;
        } src_addr, dst_addr;
        const struct sockaddr *src = NULL;
        const struct sockaddr *dst = NULL;
        char _devname[IB_DEVICE_NAME_MAX] = { 0, };
        const char *devname = NULL;
        int ret;

        if (sc->rdma.cm_id) {
                src_addr.ss = sc->rdma.cm_id->route.addr.src_addr;
                if (src_addr.sa.sa_family != AF_UNSPEC)
                        src = &src_addr.sa;
                dst_addr.ss = sc->rdma.cm_id->route.addr.dst_addr;
                if (dst_addr.sa.sa_family != AF_UNSPEC)
                        dst = &dst_addr.sa;

                if (sc->ib.dev) {
                        memcpy(_devname, sc->ib.dev->name, IB_DEVICE_NAME_MAX);
                        devname = _devname;
                }
        }

        smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
                "waiting for connection: device: %.*s local: %pISpsfc remote: %pISpsfc\n",
                IB_DEVICE_NAME_MAX, devname, src, dst);

        ret = wait_event_interruptible_timeout(sc->status_wait,
                                               sc->status == SMBDIRECT_SOCKET_CONNECTED ||
                                               sc->first_error,
                                               msecs_to_jiffies(sp->negotiate_timeout_msec));
        if (sc->rdma.cm_id) {
                /*
                 * Maybe src and dev are updated in the meantime.
                 */
                src_addr.ss = sc->rdma.cm_id->route.addr.src_addr;
                if (src_addr.sa.sa_family != AF_UNSPEC)
                        src = &src_addr.sa;
                dst_addr.ss = sc->rdma.cm_id->route.addr.dst_addr;
                if (dst_addr.sa.sa_family != AF_UNSPEC)
                        dst = &dst_addr.sa;

                if (sc->ib.dev) {
                        memcpy(_devname, sc->ib.dev->name, IB_DEVICE_NAME_MAX);
                        devname = _devname;
                }
        }
        if (ret == 0)
                ret = -ETIMEDOUT;
        if (ret < 0)
                smbdirect_socket_schedule_cleanup(sc, ret);
        if (sc->first_error) {
                int lvl = SMBDIRECT_LOG_ERR;

                ret = sc->first_error;
                if (ret == -ENODEV)
                        lvl = SMBDIRECT_LOG_INFO;

                smbdirect_log_rdma_event(sc, lvl,
                        "connection failed %1pe device: %.*s local: %pISpsfc remote: %pISpsfc\n",
                        SMBDIRECT_DEBUG_ERR_PTR(ret),
                        IB_DEVICE_NAME_MAX, devname, src, dst);
                return ret;
        }

        return 0;
}
EXPORT_SYMBOL_GPL(smbdirect_connection_wait_for_connected);

void smbdirect_connection_idle_timer_work(struct work_struct *work)
{
        struct smbdirect_socket *sc =
                container_of(work, struct smbdirect_socket, idle.timer_work.work);
        const struct smbdirect_socket_parameters *sp = &sc->parameters;

        if (sc->idle.keepalive != SMBDIRECT_KEEPALIVE_NONE) {
                smbdirect_log_keep_alive(sc, SMBDIRECT_LOG_ERR,
                        "%s => timeout sc->idle.keepalive=%s\n",
                        smbdirect_socket_status_string(sc->status),
                        sc->idle.keepalive == SMBDIRECT_KEEPALIVE_SENT ?
                        "SENT" : "PENDING");
                smbdirect_socket_schedule_cleanup(sc, -ETIMEDOUT);
                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->workqueues.idle, &sc->idle.timer_work,
                         msecs_to_jiffies(sp->keepalive_timeout_msec));
        smbdirect_log_keep_alive(sc, SMBDIRECT_LOG_INFO,
                "schedule send of empty idle message\n");
        queue_work(sc->workqueues.immediate, &sc->idle.immediate_work);
}

u16 smbdirect_connection_grant_recv_credits(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 = min3((int)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;
}

static bool smbdirect_connection_request_keep_alive(struct smbdirect_socket *sc)
{
        const 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->workqueues.idle, &sc->idle.timer_work,
                                 msecs_to_jiffies(sp->keepalive_timeout_msec));
                return true;
        }

        return false;
}

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

        if (unlikely(sc->first_error))
                return sc->first_error;

        atomic_inc(&sc->send_io.pending.count);
        ret = ib_post_send(sc->ib.qp, wr, NULL);
        if (ret) {
                atomic_dec(&sc->send_io.pending.count);
                smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_ERR,
                        "ib_post_send() failed %1pe\n",
                        SMBDIRECT_DEBUG_ERR_PTR(ret));
                smbdirect_socket_schedule_cleanup(sc, ret);
        }

        return ret;
}

static void smbdirect_connection_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;
}

int smbdirect_connection_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 send_ctx->msg_list
         * and splice the rest of send_ctx->msg_list
         * to last->sibling_list.
         *
         * send_ctx->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 = smbdirect_connection_post_send_wr(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);
                        smbdirect_connection_free_send_io(sibling);
                }
                smbdirect_connection_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;
}
EXPORT_SYMBOL_GPL(smbdirect_connection_send_batch_flush);

struct smbdirect_send_batch *
smbdirect_init_send_batch_storage(struct smbdirect_send_batch_storage *storage,
                                  bool need_invalidate_rkey,
                                  unsigned int remote_key)
{
        struct smbdirect_send_batch *batch = (struct smbdirect_send_batch *)storage;

        memset(storage, 0, sizeof(*storage));
        BUILD_BUG_ON(sizeof(*batch) > sizeof(*storage));

        smbdirect_connection_send_batch_init(batch,
                                             need_invalidate_rkey,
                                             remote_key);

        return batch;
}
EXPORT_SYMBOL_GPL(smbdirect_init_send_batch_storage);

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

        if (batch->credit)
                return 0;

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

        batch->credit = 1;
        return 0;
}

static int smbdirect_connection_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 = smbdirect_connection_send_batch_flush(sc, batch, false);
                if (ret)
                        return ret;
        }

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

static int smbdirect_connection_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 = smbdirect_connection_send_batch_flush(sc, batch, false);
                if (ret)
                        return ret;
        }

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

static void smbdirect_connection_send_io_done(struct ib_cq *cq, struct ib_wc *wc);

static int smbdirect_connection_post_send_io(struct smbdirect_socket *sc,
                                             struct smbdirect_send_batch *batch,
                                             struct smbdirect_send_io *msg)
{
        int i;

        for (i = 0; i < msg->num_sge; i++)
                ib_dma_sync_single_for_device(sc->ib.dev,
                                              msg->sge[i].addr, msg->sge[i].length,
                                              DMA_TO_DEVICE);

        msg->cqe.done = smbdirect_connection_send_io_done;
        msg->wr.wr_cqe = &msg->cqe;
        msg->wr.opcode = IB_WR_SEND;
        msg->wr.sg_list = &msg->sge[0];
        msg->wr.num_sge = msg->num_sge;
        msg->wr.next = NULL;

        if (batch) {
                msg->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 = &msg->wr;
                }
                list_add_tail(&msg->sibling_list, &batch->msg_list);
                batch->wr_cnt++;
                return 0;
        }

        msg->wr.send_flags = IB_SEND_SIGNALED;
        return smbdirect_connection_post_send_wr(sc, &msg->wr);
}

int smbdirect_connection_send_single_iter(struct smbdirect_socket *sc,
                                          struct smbdirect_send_batch *batch,
                                          struct iov_iter *iter,
                                          unsigned int flags,
                                          u32 remaining_data_length)
{
        const struct smbdirect_socket_parameters *sp = &sc->parameters;
        struct smbdirect_send_batch _batch;
        struct smbdirect_send_io *msg;
        struct smbdirect_data_transfer *packet;
        size_t header_length;
        u16 new_credits = 0;
        u32 data_length = 0;
        int ret;

        if (WARN_ON_ONCE(flags))
                return -EINVAL; /* no flags support for now */

        if (iter) {
                if (WARN_ON_ONCE(iov_iter_rw(iter) != ITER_SOURCE))
                        return -EINVAL; /* It's a bug in upper layer to get there */

                header_length = sizeof(struct smbdirect_data_transfer);
                if (WARN_ON_ONCE(remaining_data_length == 0 ||
                                 iov_iter_count(iter) > remaining_data_length))
                        return -EINVAL;
        } else {
                /* If this is a packet without payload, don't send padding */
                header_length = offsetof(struct smbdirect_data_transfer, padding);
                if (WARN_ON_ONCE(remaining_data_length))
                        return -EINVAL;
        }

        if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
                smbdirect_log_write(sc, SMBDIRECT_LOG_ERR,
                        "status=%s first_error=%1pe => %1pe\n",
                        smbdirect_socket_status_string(sc->status),
                        SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
                        SMBDIRECT_DEBUG_ERR_PTR(-ENOTCONN));
                return -ENOTCONN;
        }

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

        ret = smbdirect_connection_wait_for_send_bcredit(sc, batch);
        if (ret)
                goto bcredit_failed;

        ret = smbdirect_connection_wait_for_send_lcredit(sc, batch);
        if (ret)
                goto lcredit_failed;

        ret = smbdirect_connection_wait_for_send_credits(sc, batch);
        if (ret)
                goto credit_failed;

        new_credits = smbdirect_connection_grant_recv_credits(sc);
        if (new_credits == 0 &&
            atomic_read(&sc->send_io.credits.count) == 0 &&
            atomic_read(&sc->recv_io.credits.count) == 0) {
                /*
                 * queue the refill work in order to
                 * get some new recv credits we can grant to
                 * the peer.
                 */
                queue_work(sc->workqueues.refill, &sc->recv_io.posted.refill_work);

                /*
                 * wait until either the refill work or the peer
                 * granted new credits
                 */
                ret = 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)
                        ret = -ENOTCONN;
                if (ret < 0)
                        goto credit_failed;

                new_credits = smbdirect_connection_grant_recv_credits(sc);
        }

        msg = smbdirect_connection_alloc_send_io(sc);
        if (IS_ERR(msg)) {
                ret = PTR_ERR(msg);
                goto alloc_failed;
        }

        /* Map the packet to DMA */
        msg->sge[0].addr = ib_dma_map_single(sc->ib.dev,
                                             msg->packet,
                                             header_length,
                                             DMA_TO_DEVICE);
        ret = ib_dma_mapping_error(sc->ib.dev, msg->sge[0].addr);
        if (ret)
                goto err;

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

        if (iter) {
                struct smbdirect_map_sges extract = {
                        .num_sge        = msg->num_sge,
                        .max_sge        = ARRAY_SIZE(msg->sge),
                        .sge            = msg->sge,
                        .device         = sc->ib.dev,
                        .local_dma_lkey = sc->ib.pd->local_dma_lkey,
                        .direction      = DMA_TO_DEVICE,
                };
                size_t payload_len = umin(iov_iter_count(iter),
                                          sp->max_send_size - sizeof(*packet));

                ret = smbdirect_map_sges_from_iter(iter, payload_len, &extract);
                if (ret < 0)
                        goto err;
                data_length = ret;
                remaining_data_length -= data_length;
                msg->num_sge = extract.num_sge;
        }

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

        packet->flags = 0;
        if (smbdirect_connection_request_keep_alive(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;

        smbdirect_log_outgoing(sc, SMBDIRECT_LOG_INFO,
                "DataOut: %s=%u, %s=%u, %s=0x%x, %s=%u, %s=%u, %s=%u\n",
                "CreditsRequested",
                le16_to_cpu(packet->credits_requested),
                "CreditsGranted",
                le16_to_cpu(packet->credits_granted),
                "Flags",
                le16_to_cpu(packet->flags),
                "RemainingDataLength",
                le32_to_cpu(packet->remaining_data_length),
                "DataOffset",
                le32_to_cpu(packet->data_offset),
                "DataLength",
                le32_to_cpu(packet->data_length));

        ret = smbdirect_connection_post_send_io(sc, batch, msg);
        if (ret)
                goto err;

        /*
         * From here msg is moved to send_ctx
         * and we should not free it explicitly.
         */

        if (batch == &_batch) {
                ret = smbdirect_connection_send_batch_flush(sc, batch, true);
                if (ret)
                        goto flush_failed;
        }

        return data_length;
err:
        smbdirect_connection_free_send_io(msg);
flush_failed:
alloc_failed:
        atomic_inc(&sc->send_io.credits.count);
credit_failed:
        atomic_inc(&sc->send_io.lcredits.count);
lcredit_failed:
        atomic_add(batch->credit, &sc->send_io.bcredits.count);
        batch->credit = 0;
bcredit_failed:
        return ret;
}
EXPORT_SYMBOL_GPL(smbdirect_connection_send_single_iter);

int smbdirect_connection_send_wait_zero_pending(struct smbdirect_socket *sc)
{
        /*
         * 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) {
                smbdirect_log_write(sc, SMBDIRECT_LOG_ERR,
                        "status=%s first_error=%1pe => %1pe\n",
                        smbdirect_socket_status_string(sc->status),
                        SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
                        SMBDIRECT_DEBUG_ERR_PTR(-ENOTCONN));
                return -ENOTCONN;
        }

        return 0;
}
EXPORT_SYMBOL_GPL(smbdirect_connection_send_wait_zero_pending);

int smbdirect_connection_send_iter(struct smbdirect_socket *sc,
                                   struct iov_iter *iter,
                                   unsigned int flags,
                                   bool need_invalidate,
                                   unsigned int remote_key)
{
        const struct smbdirect_socket_parameters *sp = &sc->parameters;
        struct smbdirect_send_batch batch;
        int total_count = iov_iter_count(iter);
        int ret;
        int error = 0;
        __be32 hdr;

        if (WARN_ONCE(flags, "unexpected flags=0x%x\n", flags))
                return -EINVAL; /* no flags support for now */

        if (WARN_ON_ONCE(iov_iter_rw(iter) != ITER_SOURCE))
                return -EINVAL; /* It's a bug in upper layer to get there */

        if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
                smbdirect_log_write(sc, SMBDIRECT_LOG_INFO,
                        "status=%s first_error=%1pe => %1pe\n",
                        smbdirect_socket_status_string(sc->status),
                        SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
                        SMBDIRECT_DEBUG_ERR_PTR(-ENOTCONN));
                return -ENOTCONN;
        }

        /*
         * For now we expect the iter to have the full
         * message, including a 4 byte length header.
         */
        if (iov_iter_count(iter) <= 4)
                return -EINVAL;
        if (!copy_from_iter_full(&hdr, sizeof(hdr), iter))
                return -EFAULT;
        if (iov_iter_count(iter) != be32_to_cpu(hdr))
                return -EINVAL;

        /*
         * The size must fit into the negotiated
         * fragmented send size.
         */
        if (iov_iter_count(iter) > sp->max_fragmented_send_size)
                return -EMSGSIZE;

        smbdirect_log_write(sc, SMBDIRECT_LOG_INFO,
                "Sending (RDMA): length=%zu\n",
                iov_iter_count(iter));

        smbdirect_connection_send_batch_init(&batch, need_invalidate, remote_key);
        while (iov_iter_count(iter)) {
                ret = smbdirect_connection_send_single_iter(sc,
                                                            &batch,
                                                            iter,
                                                            flags,
                                                            iov_iter_count(iter));
                if (unlikely(ret < 0)) {
                        error = ret;
                        break;
                }
        }

        ret = smbdirect_connection_send_batch_flush(sc, &batch, true);
        if (unlikely(ret && !error))
                error = ret;

        /*
         * 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
         */

        ret = smbdirect_connection_send_wait_zero_pending(sc);
        if (unlikely(ret && !error))
                error = ret;

        if (unlikely(error))
                return error;

        return total_count;
}
EXPORT_SYMBOL_GPL(smbdirect_connection_send_iter);

static void smbdirect_connection_send_io_done(struct ib_cq *cq, struct ib_wc *wc)
{
        struct smbdirect_send_io *msg =
                container_of(wc->wr_cqe, struct smbdirect_send_io, cqe);
        struct smbdirect_socket *sc = msg->socket;
        struct smbdirect_send_io *sibling, *next;
        int lcredits = 0;

        smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_INFO,
                "smbdirect_send_io completed. status='%s (%d)', opcode=%d\n",
                ib_wc_status_msg(wc->status), wc->status, wc->opcode);

        if (unlikely(!(msg->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...
                 */
                smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_ERR,
                        "unexpected send completion wc->status=%s (%d) wc->opcode=%d\n",
                        ib_wc_status_msg(wc->status), wc->status, wc->opcode);
                smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
                return;
        }

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

        if (unlikely(wc->status != IB_WC_SUCCESS || WARN_ON_ONCE(wc->opcode != IB_WC_SEND))) {
skip_free:
                if (wc->status != IB_WC_WR_FLUSH_ERR)
                        smbdirect_log_rdma_send(sc, SMBDIRECT_LOG_ERR,
                                "wc->status=%s (%d) wc->opcode=%d\n",
                                ib_wc_status_msg(wc->status), wc->status, wc->opcode);
                smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
                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);
}

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

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

        smbdirect_log_keep_alive(sc, SMBDIRECT_LOG_INFO,
                "send an empty message\n");
        sc->statistics.send_empty++;
        ret = smbdirect_connection_send_single_iter(sc, NULL, NULL, 0, 0);
        if (ret < 0) {
                smbdirect_log_write(sc, SMBDIRECT_LOG_ERR,
                        "smbdirect_connection_send_single_iter ret=%1pe\n",
                        SMBDIRECT_DEBUG_ERR_PTR(ret));
                smbdirect_socket_schedule_cleanup(sc, ret);
        }
}

int smbdirect_connection_post_recv_io(struct smbdirect_recv_io *msg)
{
        struct smbdirect_socket *sc = msg->socket;
        const struct smbdirect_socket_parameters *sp = &sc->parameters;
        struct ib_recv_wr recv_wr = {
                .wr_cqe = &msg->cqe,
                .sg_list = &msg->sge,
                .num_sge = 1,
        };
        int ret;

        if (unlikely(sc->first_error))
                return sc->first_error;

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

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

        ret = ib_post_recv(sc->ib.qp, &recv_wr, NULL);
        if (ret) {
                smbdirect_log_rdma_recv(sc, SMBDIRECT_LOG_ERR,
                        "ib_post_recv failed ret=%d (%1pe)\n",
                        ret, SMBDIRECT_DEBUG_ERR_PTR(ret));
                ib_dma_unmap_single(sc->ib.dev,
                                    msg->sge.addr,
                                    msg->sge.length,
                                    DMA_FROM_DEVICE);
                msg->sge.length = 0;
                smbdirect_socket_schedule_cleanup(sc, ret);
        }

        return ret;
}

void smbdirect_connection_recv_io_done(struct ib_cq *cq, struct ib_wc *wc)
{
        struct smbdirect_recv_io *recv_io =
                container_of(wc->wr_cqe, struct smbdirect_recv_io, cqe);
        struct smbdirect_socket *sc = recv_io->socket;
        const struct smbdirect_socket_parameters *sp = &sc->parameters;
        struct smbdirect_data_transfer *data_transfer;
        int current_recv_credits;
        u16 old_recv_credit_target;
        u16 credits_requested;
        u16 credits_granted;
        u16 flags;
        u32 data_offset;
        u32 data_length;
        u32 remaining_data_length;

        if (unlikely(wc->status != IB_WC_SUCCESS || WARN_ON_ONCE(wc->opcode != IB_WC_RECV))) {
                if (wc->status != IB_WC_WR_FLUSH_ERR)
                        smbdirect_log_rdma_recv(sc, SMBDIRECT_LOG_ERR,
                                "wc->status=%s (%d) wc->opcode=%d\n",
                                ib_wc_status_msg(wc->status), wc->status, wc->opcode);
                goto error;
        }

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

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

        ib_dma_sync_single_for_cpu(sc->ib.dev,
                                   recv_io->sge.addr,
                                   recv_io->sge.length,
                                   DMA_FROM_DEVICE);

        if (unlikely(wc->byte_len <
            offsetof(struct smbdirect_data_transfer, padding))) {
                smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
                        "wc->byte_len=%u < %zu\n",
                        wc->byte_len,
                        offsetof(struct smbdirect_data_transfer, padding));
                goto error;
        }

        data_transfer = (struct smbdirect_data_transfer *)recv_io->packet;
        credits_requested = le16_to_cpu(data_transfer->credits_requested);
        credits_granted = le16_to_cpu(data_transfer->credits_granted);
        flags = le16_to_cpu(data_transfer->flags);
        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);

        smbdirect_log_incoming(sc, SMBDIRECT_LOG_INFO,
                "DataIn: %s=%u, %s=%u, %s=0x%x, %s=%u, %s=%u, %s=%u\n",
                "CreditsRequested",
                credits_requested,
                "CreditsGranted",
                credits_granted,
                "Flags",
                flags,
                "RemainingDataLength",
                remaining_data_length,
                "DataOffset",
                data_offset,
                "DataLength",
                data_length);

        if (unlikely(credits_requested == 0)) {
                smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
                        "invalid: credits_requested == 0\n");
                goto error;
        }

        if (unlikely(data_offset % 8 != 0)) {
                smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
                        "invalid: data_offset=%u (0x%x) not aligned to 8\n",
                        data_offset, data_offset);
                goto error;
        }

        if (unlikely(wc->byte_len < data_offset ||
            (u64)wc->byte_len < (u64)data_offset + data_length)) {
                smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
                        "wc->byte_len=%u < date_offset=%u + data_length=%u\n",
                        wc->byte_len, data_offset, data_length);
                goto error;
        }

        if (unlikely(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)) {
                smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
                        "remaining_data_length=%u + data_length=%u > max_fragmented=%u\n",
                        remaining_data_length, data_length, sp->max_fragmented_recv_size);
                goto error;
        }

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

                if (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);

        /*
         * We take the value from the peer, which is checked to be higher than 0,
         * but we limit it to the max value we support in order to have
         * the main logic simpler.
         */
        old_recv_credit_target = sc->recv_io.credits.target;
        sc->recv_io.credits.target = credits_requested;
        sc->recv_io.credits.target = min_t(u16, sc->recv_io.credits.target,
                                           sp->recv_credit_max);
        if (credits_granted) {
                atomic_add(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);
        }

        /* Send an immediate response right away if requested */
        if (flags & SMBDIRECT_FLAG_RESPONSE_REQUESTED) {
                smbdirect_log_keep_alive(sc, SMBDIRECT_LOG_INFO,
                        "schedule send of immediate response\n");
                queue_work(sc->workqueues.immediate, &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->workqueues.refill, &sc->recv_io.posted.refill_work);

                smbdirect_connection_reassembly_append_recv_io(sc, recv_io, data_length);
                wake_up(&sc->recv_io.reassembly.wait_queue);
        } else
                smbdirect_connection_put_recv_io(recv_io);

        return;

error:
        /*
         * Make sure smbdirect_connection_put_recv_io() does not
         * start recv_io.posted.refill_work.
         */
        disable_work(&sc->recv_io.posted.refill_work);
        smbdirect_connection_put_recv_io(recv_io);
        smbdirect_socket_schedule_cleanup(sc, -ECONNABORTED);
}

int smbdirect_connection_recv_io_refill(struct smbdirect_socket *sc)
{
        int missing;
        int posted = 0;

        if (unlikely(sc->first_error))
                return sc->first_error;

        /*
         * Find out how much smbdirect_recv_io buffers we should post.
         *
         * Note that sc->recv_io.credits.target is the value
         * from the peer and it can in theory change over time,
         * but it is forced to be at least 1 and at max
         * sp->recv_credit_max.
         *
         * So it can happen that missing will be lower than 0,
         * which means the peer has recently lowered its desired
         * target, while be already granted a higher number of credits.
         *
         * Note 'posted' is the number of smbdirect_recv_io buffers
         * posted within this function, while sc->recv_io.posted.count
         * is the overall value of posted smbdirect_recv_io buffers.
         *
         * We try to post as much buffers as missing, but
         * this is limited if a lot of smbdirect_recv_io buffers
         * are still in the sc->recv_io.reassembly.list instead of
         * the sc->recv_io.free.list.
         *
         */
        missing = (int)sc->recv_io.credits.target - atomic_read(&sc->recv_io.posted.count);
        while (posted < missing) {
                struct smbdirect_recv_io *recv_io;
                int ret;

                /*
                 * It's ok if smbdirect_connection_get_recv_io()
                 * returns NULL, it means smbdirect_recv_io structures
                 * are still be in the reassembly.list.
                 */
                recv_io = smbdirect_connection_get_recv_io(sc);
                if (!recv_io)
                        break;

                recv_io->first_segment = false;

                ret = smbdirect_connection_post_recv_io(recv_io);
                if (ret) {
                        smbdirect_log_rdma_recv(sc, SMBDIRECT_LOG_ERR,
                                "smbdirect_connection_post_recv_io failed rc=%d (%1pe)\n",
                                ret, SMBDIRECT_DEBUG_ERR_PTR(ret));
                        smbdirect_connection_put_recv_io(recv_io);
                        return ret;
                }

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

        /* If nothing was posted we're done */
        if (posted == 0)
                return 0;

        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 (atomic_read(&sc->send_io.bcredits.count) == 0 &&
            atomic_read(&sc->send_io.credits.count) == 0)
                wake_up(&sc->send_io.credits.wait_queue);

        /*
         * If we posted at least one smbdirect_recv_io buffer,
         * we need to inform the peer about it and grant
         * additional credits.
         *
         * However there is one case where we don't want to
         * do that.
         *
         * If only a single credit was missing before
         * reaching the requested target, we should not
         * post an immediate send, as that would cause
         * endless ping pong once a keep alive exchange
         * is started.
         *
         * However if sc->recv_io.credits.target is only 1,
         * the peer has no credit left and we need to
         * grant the credit anyway.
         */
        if (missing == 1 && sc->recv_io.credits.target != 1)
                return 0;

        return posted;
}

static void smbdirect_connection_recv_io_refill_work(struct work_struct *work)
{
        struct smbdirect_socket *sc =
                container_of(work, struct smbdirect_socket, recv_io.posted.refill_work);
        int posted;

        posted = smbdirect_connection_recv_io_refill(sc);
        if (unlikely(posted < 0)) {
                smbdirect_socket_schedule_cleanup(sc, posted);
                return;
        }
        if (posted > 0) {
                smbdirect_log_keep_alive(sc, SMBDIRECT_LOG_INFO,
                        "schedule send of an empty message\n");
                queue_work(sc->workqueues.immediate, &sc->idle.immediate_work);
        }
}

int smbdirect_connection_recvmsg(struct smbdirect_socket *sc,
                                 struct msghdr *msg,
                                 unsigned int flags)
{
        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 ret;

        if (WARN_ONCE(flags, "unexpected flags=0x%x\n", flags))
                return -EINVAL; /* no flags support for now */

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

again:
        if (sc->status != SMBDIRECT_SOCKET_CONNECTED) {
                smbdirect_log_read(sc, SMBDIRECT_LOG_INFO,
                        "status=%s first_error=%1pe => %1pe\n",
                        smbdirect_socket_status_string(sc->status),
                        SMBDIRECT_DEBUG_ERR_PTR(sc->first_error),
                        SMBDIRECT_DEBUG_ERR_PTR(-ENOTCONN));
                return -ENOTCONN;
        }

        /*
         * 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
         */
        smbdirect_log_read(sc, SMBDIRECT_LOG_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
                 * smbdirect_connection_reassembly_first_recv_io. 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 = smbdirect_connection_reassembly_first_recv_io(sc);
                        data_transfer = (void *)response->packet;
                        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;
                                smbdirect_log_read(sc, SMBDIRECT_LOG_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((u8 *)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++;
                                smbdirect_connection_put_recv_io(response);
                                offset = 0;
                                smbdirect_log_read(sc, SMBDIRECT_LOG_INFO,
                                        "smbdirect_connection_put_recv_io offset=0\n");
                        } else
                                offset += to_copy;

                        to_read -= to_copy;
                        data_read += to_copy;

                        smbdirect_log_read(sc, SMBDIRECT_LOG_INFO,
                                 "memcpy %d bytes len-ofs=%u => todo=%u done=%u ofs=%u\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;
                smbdirect_log_read(sc, SMBDIRECT_LOG_INFO,
                         "returning data_read=%d reassembly_length=%d first_ofs=%u\n",
                         data_read, sc->recv_io.reassembly.data_length,
                         sc->recv_io.reassembly.first_entry_offset);
read_rfc1002_done:
                return data_read;
        }

        smbdirect_log_read(sc, SMBDIRECT_LOG_INFO,
                "wait_event on more data\n");
        ret = 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 (ret)
                return ret;

        goto again;
}
EXPORT_SYMBOL_GPL(smbdirect_connection_recvmsg);

static bool smbdirect_map_sges_single_page(struct smbdirect_map_sges *state,
                                           struct page *page, size_t off, size_t len)
{
        struct ib_sge *sge;
        u64 addr;

        if (state->num_sge >= state->max_sge)
                return false;

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

        sge = &state->sge[state->num_sge++];
        sge->addr   = addr;
        sge->length = len;
        sge->lkey   = state->local_dma_lkey;

        return true;
}

/*
 * Extract page fragments from a BVEC-class iterator and add them to an ib_sge
 * list.  The pages are not pinned.
 */
static ssize_t smbdirect_map_sges_from_bvec(struct iov_iter *iter,
                                            struct smbdirect_map_sges *state,
                                            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;
                bool ok;

                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;

                ok = smbdirect_map_sges_single_page(state,
                                                    bv[i].bv_page,
                                                    off,
                                                    len);
                if (!ok)
                        return -EIO;

                ret += len;
                maxsize -= len;
                if (state->num_sge >= state->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 ib_sge list.
 * This can deal with vmalloc'd buffers as well as kmalloc'd or static buffers.
 * The pages are not pinned.
 */
static ssize_t smbdirect_map_sges_from_kvec(struct iov_iter *iter,
                                            struct smbdirect_map_sges *state,
                                            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 {
                        bool ok;

                        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);

                        ok = smbdirect_map_sges_single_page(state, page, off, seg);
                        if (!ok)
                                return -EIO;

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

                if (state->num_sge >= state->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
 * ib_sge list.  The folios are not pinned.
 */
static ssize_t smbdirect_map_sges_from_folioq(struct iov_iter *iter,
                                              struct smbdirect_map_sges *state,
                                              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;

        if (WARN_ON_ONCE(!folioq))
                return -EIO;

        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);
                        bool ok;

                        ok = smbdirect_map_sges_single_page(state,
                                                            folio_page(folio, 0),
                                                            offset,
                                                            part);
                        if (!ok)
                                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 (state->num_sge < state->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 ib_sge list that refers to all of those bits.  The ib_sge 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 smbdirect_map_sges_from_iter(struct iov_iter *iter, size_t len,
                                            struct smbdirect_map_sges *state)
{
        ssize_t ret;
        size_t before = state->num_sge;

        if (WARN_ON_ONCE(iov_iter_rw(iter) != ITER_SOURCE))
                return -EIO;

        switch (iov_iter_type(iter)) {
        case ITER_BVEC:
                ret = smbdirect_map_sges_from_bvec(iter, state, len);
                break;
        case ITER_KVEC:
                ret = smbdirect_map_sges_from_kvec(iter, state, len);
                break;
        case ITER_FOLIOQ:
                ret = smbdirect_map_sges_from_folioq(iter, state, len);
                break;
        default:
                WARN_ONCE(1, "iov_iter_type[%u]\n", iov_iter_type(iter));
                return -EIO;
        }

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

                        ib_dma_unmap_page(state->device,
                                          sge->addr,
                                          sge->length,
                                          state->direction);
                }
        }

        return ret;
}