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

#include "internal.h"

bool smbdirect_frwr_is_supported(const struct ib_device_attr *attrs)
{
	/*
	 * 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
	 */

	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;
}
EXPORT_SYMBOL_GPL(smbdirect_frwr_is_supported);

static void smbdirect_socket_cleanup_work(struct work_struct *work);

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

	/*
	 * This should be replaced before any real work
	 * starts! So it should never be called!
	 */

	if (event->event == RDMA_CM_EVENT_DEVICE_REMOVAL)
		ret = -ENETDOWN;
	if (IS_ERR(SMBDIRECT_DEBUG_ERR_PTR(event->status)))
		ret = event->status;
	pr_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));
	WARN_ONCE(1, "%s should not be called!\n", __func__);
	sc->rdma.cm_id = NULL;
	return -ESTALE;
}

int smbdirect_socket_init_new(struct net *net, struct smbdirect_socket *sc)
{
	struct rdma_cm_id *id;
	int ret;

	smbdirect_socket_init(sc);

	id = rdma_create_id(net,
			    smbdirect_socket_rdma_event_handler,
			    sc,
			    RDMA_PS_TCP,
			    IB_QPT_RC);
	if (IS_ERR(id)) {
		pr_err("%s: rdma_create_id() failed %1pe\n", __func__, id);
		return PTR_ERR(id);
	}

	ret = rdma_set_afonly(id, 1);
	if (ret) {
		rdma_destroy_id(id);
		pr_err("%s: rdma_set_afonly() failed %1pe\n",
		       __func__, SMBDIRECT_DEBUG_ERR_PTR(ret));
		return ret;
	}

	sc->rdma.cm_id = id;

	INIT_WORK(&sc->disconnect_work, smbdirect_socket_cleanup_work);

	return 0;
}

int smbdirect_socket_create_kern(struct net *net, struct smbdirect_socket **_sc)
{
	struct smbdirect_socket *sc;
	int ret;

	ret = -ENOMEM;
	sc = kzalloc_obj(*sc);
	if (!sc)
		goto alloc_failed;

	ret = smbdirect_socket_init_new(net, sc);
	if (ret)
		goto init_failed;

	kref_init(&sc->refs.destroy);

	*_sc = sc;
	return 0;

init_failed:
	kfree(sc);
alloc_failed:
	return ret;
}
EXPORT_SYMBOL_GPL(smbdirect_socket_create_kern);

int smbdirect_socket_init_accepting(struct rdma_cm_id *id, struct smbdirect_socket *sc)
{
	smbdirect_socket_init(sc);

	sc->rdma.cm_id = id;
	sc->rdma.cm_id->context = sc;
	sc->rdma.cm_id->event_handler = smbdirect_socket_rdma_event_handler;

	sc->ib.dev = sc->rdma.cm_id->device;

	INIT_WORK(&sc->disconnect_work, smbdirect_socket_cleanup_work);

	return 0;
}

int smbdirect_socket_create_accepting(struct rdma_cm_id *id, struct smbdirect_socket **_sc)
{
	struct smbdirect_socket *sc;
	int ret;

	ret = -ENOMEM;
	sc = kzalloc_obj(*sc);
	if (!sc)
		goto alloc_failed;

	ret = smbdirect_socket_init_accepting(id, sc);
	if (ret)
		goto init_failed;

	kref_init(&sc->refs.destroy);

	*_sc = sc;
	return 0;

init_failed:
	kfree(sc);
alloc_failed:
	return ret;
}
EXPORT_SYMBOL_GPL(smbdirect_socket_create_accepting);

int smbdirect_socket_set_initial_parameters(struct smbdirect_socket *sc,
					    const struct smbdirect_socket_parameters *sp)
{
	/*
	 * This is only allowed before connect or accept
	 */
	WARN_ONCE(sc->status != SMBDIRECT_SOCKET_CREATED,
		  "status=%s first_error=%1pe",
		  smbdirect_socket_status_string(sc->status),
		  SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
	if (sc->status != SMBDIRECT_SOCKET_CREATED)
		return -EINVAL;

	if (sp->flags & ~SMBDIRECT_FLAG_PORT_RANGE_MASK)
		return -EINVAL;

	if (sp->initiator_depth > U8_MAX)
		return -EINVAL;
	if (sp->responder_resources > U8_MAX)
		return -EINVAL;

	if (sp->flags & SMBDIRECT_FLAG_PORT_RANGE_ONLY_IB &&
	    sp->flags & SMBDIRECT_FLAG_PORT_RANGE_ONLY_IW)
		return -EINVAL;
	else if (sp->flags & SMBDIRECT_FLAG_PORT_RANGE_ONLY_IB)
		rdma_restrict_node_type(sc->rdma.cm_id, RDMA_NODE_IB_CA);
	else if (sp->flags & SMBDIRECT_FLAG_PORT_RANGE_ONLY_IW)
		rdma_restrict_node_type(sc->rdma.cm_id, RDMA_NODE_RNIC);

	/*
	 * Make a copy of the callers parameters
	 * from here we only work on the copy
	 *
	 * TODO: do we want consistency checking?
	 */
	sc->parameters = *sp;

	return 0;
}
EXPORT_SYMBOL_GPL(smbdirect_socket_set_initial_parameters);

const struct smbdirect_socket_parameters *
smbdirect_socket_get_current_parameters(struct smbdirect_socket *sc)
{
	return &sc->parameters;
}
EXPORT_SYMBOL_GPL(smbdirect_socket_get_current_parameters);

int smbdirect_socket_set_kernel_settings(struct smbdirect_socket *sc,
					 enum ib_poll_context poll_ctx,
					 gfp_t gfp_mask)
{
	/*
	 * This is only allowed before connect or accept
	 */
	WARN_ONCE(sc->status != SMBDIRECT_SOCKET_CREATED,
		  "status=%s first_error=%1pe",
		  smbdirect_socket_status_string(sc->status),
		  SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
	if (sc->status != SMBDIRECT_SOCKET_CREATED)
		return -EINVAL;

	sc->ib.poll_ctx = poll_ctx;

	sc->send_io.mem.gfp_mask = gfp_mask;
	sc->recv_io.mem.gfp_mask = gfp_mask;
	sc->rw_io.mem.gfp_mask = gfp_mask;

	return 0;
}
EXPORT_SYMBOL_GPL(smbdirect_socket_set_kernel_settings);

void smbdirect_socket_set_logging(struct smbdirect_socket *sc,
				  void *private_ptr,
				  bool (*needed)(struct smbdirect_socket *sc,
						 void *private_ptr,
						 unsigned int lvl,
						 unsigned int cls),
				  void (*vaprintf)(struct smbdirect_socket *sc,
						   const char *func,
						   unsigned int line,
						   void *private_ptr,
						   unsigned int lvl,
						   unsigned int cls,
						   struct va_format *vaf))
{
	sc->logging.private_ptr = private_ptr;
	sc->logging.needed = needed;
	sc->logging.vaprintf = vaprintf;
}
EXPORT_SYMBOL_GPL(smbdirect_socket_set_logging);

static void smbdirect_socket_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->listen.wait_queue);
	wake_up_all(&sc->send_io.bcredits.wait_queue);
	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.zero_wait_queue);
	wake_up_all(&sc->recv_io.reassembly.wait_queue);
	wake_up_all(&sc->rw_io.credits.wait_queue);
	wake_up_all(&sc->mr_io.ready.wait_queue);
}

void __smbdirect_socket_schedule_cleanup(struct smbdirect_socket *sc,
					 const char *macro_name,
					 unsigned int lvl,
					 const char *func,
					 unsigned int line,
					 int error,
					 enum smbdirect_socket_status *force_status)
{
	struct smbdirect_socket *psc, *tsc;
	unsigned long flags;
	bool was_first = false;

	if (!sc->first_error) {
		___smbdirect_log_generic(sc, func, line,
			lvl,
			SMBDIRECT_LOG_RDMA_EVENT,
			"%s(%1pe%s%s) called from %s in line=%u status=%s\n",
			macro_name,
			SMBDIRECT_DEBUG_ERR_PTR(error),
			force_status ? ", " : "",
			force_status ? smbdirect_socket_status_string(*force_status) : "",
			func, line,
			smbdirect_socket_status_string(sc->status));
		if (error)
			sc->first_error = error;
		else
			sc->first_error = -ECONNABORTED;
		was_first = true;
	}

	/*
	 * 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->connect.work);
	disable_work(&sc->recv_io.posted.refill_work);
	disable_work(&sc->idle.immediate_work);
	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_NONE;
	disable_delayed_work(&sc->idle.timer_work);

	/*
	 * In case we were a listener we need to
	 * disconnect all pending and ready sockets
	 *
	 * First we move ready sockets to pending again.
	 *
	 * Only a socket that was a listener (listen.backlog != -1) owns a
	 * populated listen.ready/pending list.  Guarding on that also keeps
	 * lockdep quiet: without it, the listener holds sc->listen.lock while
	 * the loop recurses into each child psc, which takes psc->listen.lock.
	 * Those are always different instances of the same lock class -- a
	 * child never listens, so the nesting is strictly listener -> child
	 * and cannot really deadlock -- but lockdep only sees the class and
	 * reports "possible recursive locking".  A child has empty listen
	 * lists and nothing to do here, so skipping it loses nothing, and a
	 * pending child stays on its listener's list for the free path
	 * (smbdirect_socket_destroy) to reap.
	 */
	if (sc->listen.backlog != -1) { /* was a listener */
		spin_lock_irqsave(&sc->listen.lock, flags);
		list_splice_init(&sc->listen.ready, &sc->listen.pending);
		list_for_each_entry_safe(psc, tsc, &sc->listen.pending, accept.list)
			smbdirect_socket_schedule_cleanup(psc, sc->first_error);
		spin_unlock_irqrestore(&sc->listen.lock, flags);
	}

	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:
	case SMBDIRECT_SOCKET_LISTENING:
		sc->status = SMBDIRECT_SOCKET_DISCONNECTED;
		break;

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

	if (force_status && (was_first || *force_status > sc->status))
		sc->status = *force_status;

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

	queue_work(sc->workqueues.cleanup, &sc->disconnect_work);
}

static void smbdirect_socket_cleanup_work(struct work_struct *work)
{
	struct smbdirect_socket *sc =
		container_of(work, struct smbdirect_socket, disconnect_work);
	struct smbdirect_socket *psc, *tsc;
	unsigned long flags;

	/*
	 * This should not never be called in an interrupt!
	 */
	WARN_ON_ONCE(in_interrupt());

	if (!sc->first_error) {
		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_ERR,
			"%s called with first_error==0\n",
			smbdirect_socket_status_string(sc->status));

		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->connect.work);
	disable_work(&sc->recv_io.posted.refill_work);
	disable_work(&sc->idle.immediate_work);
	sc->idle.keepalive = SMBDIRECT_KEEPALIVE_NONE;
	disable_delayed_work(&sc->idle.timer_work);

	/*
	 * In case we were a listener we need to
	 * disconnect all pending and ready sockets
	 *
	 * First we move ready sockets to pending again.
	 *
	 * Guarded on listen.backlog != -1 for the same reason as in
	 * __smbdirect_socket_schedule_cleanup(): only a listener owns a
	 * populated listen list, and skipping the block for a child avoids
	 * nesting psc->listen.lock under a listener's listen.lock (different
	 * instances of one class -- harmless, but lockdep cannot tell).
	 */
	if (sc->listen.backlog != -1) { /* was a listener */
		spin_lock_irqsave(&sc->listen.lock, flags);
		list_splice_init(&sc->listen.ready, &sc->listen.pending);
		list_for_each_entry_safe(psc, tsc, &sc->listen.pending, accept.list)
			smbdirect_socket_schedule_cleanup(psc, sc->first_error);
		spin_unlock_irqrestore(&sc->listen.lock, flags);
	}

	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;
		/*
		 * Make sure we hold the callback lock
		 * im order to coordinate with the
		 * rdma_event handlers, typically
		 * smbdirect_connection_rdma_event_handler(),
		 * and smbdirect_socket_destroy().
		 *
		 * So that the order of ib_drain_qp()
		 * and rdma_disconnect() is controlled
		 * by the mutex.
		 */
		rdma_lock_handler(sc->rdma.cm_id);
		rdma_disconnect(sc->rdma.cm_id);
		rdma_unlock_handler(sc->rdma.cm_id);
		break;

	case SMBDIRECT_SOCKET_CREATED:
	case SMBDIRECT_SOCKET_LISTENING:
	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_{accept,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.
	 */
	smbdirect_socket_wake_up_all(sc);
}

static void smbdirect_socket_destroy(struct smbdirect_socket *sc)
{
	struct smbdirect_socket *psc, *tsc;
	size_t psockets;
	struct smbdirect_recv_io *recv_io;
	struct smbdirect_recv_io *recv_tmp;
	LIST_HEAD(all_list);
	LIST_HEAD(pending_list);
	unsigned long flags;

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

	/*
	 * This should not never be called in an interrupt!
	 */
	WARN_ON_ONCE(in_interrupt());

	if (sc->status == SMBDIRECT_SOCKET_DESTROYED)
		return;

	WARN_ONCE(sc->status != SMBDIRECT_SOCKET_DISCONNECTED,
		  "status=%s first_error=%1pe",
		  smbdirect_socket_status_string(sc->status),
		  SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));

	/*
	 * The listener should clear this before we reach this
	 */
	WARN_ONCE(sc->accept.listener,
		  "status=%s first_error=%1pe",
		  smbdirect_socket_status_string(sc->status),
		  SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));

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

	disable_work_sync(&sc->disconnect_work);
	disable_work_sync(&sc->connect.work);
	disable_work_sync(&sc->recv_io.posted.refill_work);
	disable_work_sync(&sc->idle.immediate_work);
	disable_delayed_work_sync(&sc->idle.timer_work);

	if (sc->rdma.cm_id)
		rdma_lock_handler(sc->rdma.cm_id);

	if (sc->ib.qp) {
		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
			"drain qp\n");
		ib_drain_qp(sc->ib.qp);
	}

	/*
	 * In case we were a listener we need to
	 * disconnect all pending and ready sockets
	 *
	 * We move ready sockets to pending again.
	 *
	 * Capture them here -- rdma_lock_handler(sc->rdma.cm_id) is held above,
	 * so a concurrent CM CONNECT_REQUEST cannot add more; sc->listen.lock
	 * below only protects the list splice itself -- but DEFER releasing
	 * them until the listener's cm_id is destroyed:
	 *
	 * - smbdirect_socket_release() -> smbdirect_socket_destroy() takes the
	 *   child's own rdma_lock_handler() lock (&id_priv->handler_mutex).
	 *   The listener's and the child's cm_id are always different
	 *   instances, so the nesting cannot really deadlock, but lockdep only
	 *   sees one lock class and reports "possible recursive locking".
	 *
	 * - rdma_destroy_id() of a child before the listener's own
	 *   rdma_destroy_id() below lets _cma_cancel_listens() walk the freed
	 *   child id_priv (KASAN slab-use-after-free in __mutex_lock()).
	 *
	 * The children are independent sockets whose teardown does not need
	 * the listener's handler lock.
	 */
	spin_lock_irqsave(&sc->listen.lock, flags);
	list_splice_tail_init(&sc->listen.ready, &pending_list);
	list_splice_tail_init(&sc->listen.pending, &pending_list);
	spin_unlock_irqrestore(&sc->listen.lock, flags);

	/* It's not possible for upper layer to get to reassembly */
	if (sc->listen.backlog == -1) /* was not a listener */
		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
			"drain the reassembly queue\n");
	spin_lock_irqsave(&sc->recv_io.reassembly.lock, flags);
	list_splice_tail_init(&sc->recv_io.reassembly.list, &all_list);
	spin_unlock_irqrestore(&sc->recv_io.reassembly.lock, flags);
	list_for_each_entry_safe(recv_io, recv_tmp, &all_list, list)
		smbdirect_connection_put_recv_io(recv_io);
	sc->recv_io.reassembly.data_length = 0;

	if (sc->listen.backlog == -1) /* was not a listener */
		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
			"freeing mr list\n");
	smbdirect_connection_destroy_mr_list(sc);

	if (sc->listen.backlog == -1) /* was not a listener */
		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
			"destroying qp\n");
	smbdirect_connection_destroy_qp(sc);
	if (sc->rdma.cm_id) {
		rdma_unlock_handler(sc->rdma.cm_id);
		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
			"destroying cm_id\n");
		rdma_destroy_id(sc->rdma.cm_id);
		sc->rdma.cm_id = NULL;
	}

	/*
	 * The listener's rdma_lock_handler() lock is dropped and its cm_id is
	 * destroyed, so it is safe to release the child sockets captured
	 * above: each release recurses into smbdirect_socket_destroy() and
	 * takes that child's own handler_mutex without nesting it under the
	 * listener's, and _cma_cancel_listens() can no longer reach them.
	 */
	psockets = list_count_nodes(&pending_list);
	if (sc->listen.backlog != -1) /* was a listener */
		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
			"release %zu pending sockets\n", psockets);
	list_for_each_entry_safe(psc, tsc, &pending_list, accept.list) {
		list_del_init(&psc->accept.list);
		psc->accept.listener = NULL;
		smbdirect_socket_release(psc);
	}
	if (sc->listen.backlog != -1) /* was a listener */
		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
			"released %zu pending sockets\n", psockets);

	if (sc->listen.backlog == -1) /* was not a listener */
		smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
			"destroying mem pools\n");
	smbdirect_connection_destroy_mem_pools(sc);

	sc->status = SMBDIRECT_SOCKET_DESTROYED;

	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
		"rdma session destroyed\n");
}

void smbdirect_socket_destroy_sync(struct smbdirect_socket *sc)
{
	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
		"status=%s first_error=%1pe",
		smbdirect_socket_status_string(sc->status),
		SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));

	/*
	 * This should not never be called in an interrupt!
	 */
	WARN_ON_ONCE(in_interrupt());

	/*
	 * First we try to disable the work
	 * without disable_work_sync() in a
	 * non blocking way, if it's already
	 * running it will be handles by
	 * disable_work_sync() below.
	 *
	 * Here we just want to make sure queue_work() in
	 * smbdirect_socket_schedule_cleanup_lvl()
	 * is a no-op.
	 */
	disable_work(&sc->disconnect_work);

	if (!sc->first_error)
		/*
		 * SMBDIRECT_LOG_INFO is enough here
		 * as this is the typical case where
		 * we terminate the connection ourself.
		 */
		smbdirect_socket_schedule_cleanup_lvl(sc,
						      SMBDIRECT_LOG_INFO,
						      -ESHUTDOWN);

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

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

	/*
	 * Once we reached SMBDIRECT_SOCKET_DISCONNECTED,
	 * we should call smbdirect_socket_destroy()
	 */
	smbdirect_socket_destroy(sc);
	smbdirect_log_rdma_event(sc, SMBDIRECT_LOG_INFO,
		"status=%s first_error=%1pe",
		smbdirect_socket_status_string(sc->status),
		SMBDIRECT_DEBUG_ERR_PTR(sc->first_error));
}

int smbdirect_socket_bind(struct smbdirect_socket *sc, struct sockaddr *addr)
{
	int ret;

	if (sc->status != SMBDIRECT_SOCKET_CREATED)
		return -EINVAL;

	ret = rdma_bind_addr(sc->rdma.cm_id, addr);
	if (ret)
		return ret;

	return 0;
}
EXPORT_SYMBOL_GPL(smbdirect_socket_bind);

void smbdirect_socket_shutdown(struct smbdirect_socket *sc)
{
	smbdirect_socket_schedule_cleanup_lvl(sc, SMBDIRECT_LOG_INFO, -ESHUTDOWN);
}
EXPORT_SYMBOL_GPL(smbdirect_socket_shutdown);

static void smbdirect_socket_release_disconnect(struct kref *kref)
{
	struct smbdirect_socket *sc =
		container_of(kref, struct smbdirect_socket, refs.disconnect);

	/*
	 * For now do a sync disconnect/destroy
	 */
	smbdirect_socket_destroy_sync(sc);
}

static void smbdirect_socket_release_destroy(struct kref *kref)
{
	struct smbdirect_socket *sc =
		container_of(kref, struct smbdirect_socket, refs.destroy);

	/*
	 * Do a sync disconnect/destroy...
	 * hopefully a no-op, as it should be already
	 * in DESTROYED state, before we free the memory.
	 */
	smbdirect_socket_destroy_sync(sc);
	kfree(sc);
}

void smbdirect_socket_release(struct smbdirect_socket *sc)
{
	/*
	 * We expect only 1 disconnect reference
	 * and if it is already 0, it's a use after free!
	 */
	WARN_ON_ONCE(kref_read(&sc->refs.disconnect) != 1);
	WARN_ON(!kref_put(&sc->refs.disconnect, smbdirect_socket_release_disconnect));

	/*
	 * This may not trigger smbdirect_socket_release_destroy(),
	 * if struct smbdirect_socket is embedded in another structure
	 * indicated by REFCOUNT_MAX.
	 */
	kref_put(&sc->refs.destroy, smbdirect_socket_release_destroy);
}
EXPORT_SYMBOL_GPL(smbdirect_socket_release);

int smbdirect_socket_wait_for_credits(struct smbdirect_socket *sc,
				      enum smbdirect_socket_status expected_status,
				      int unexpected_errno,
				      wait_queue_head_t *waitq,
				      atomic_t *total_credits,
				      int needed)
{
	int ret;

	if (WARN_ON_ONCE(needed < 0))
		return -EINVAL;

	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 != expected_status);

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