core/src/wlan_dp_rx_thread.c

/*
 * Copyright (c) 2014-2021 The Linux Foundation. All rights reserved.
 * Copyright (c) 2022-2023 Qualcomm Innovation Center, Inc. All rights reserved.
 *
 * Permission to use, copy, modify, and/or distribute this software for
 * any purpose with or without fee is hereby granted, provided that the
 * above copyright notice and this permission notice appear in all
 * copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL
 * WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE
 * AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL
 * DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR
 * PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER
 * TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
 * PERFORMANCE OF THIS SOFTWARE.
 */

#include <wlan_dp_rx_thread.h>
#include "dp_peer.h"
#include "dp_internal.h"
#include "dp_types.h"
#include <cdp_txrx_cmn_struct.h>
#include <cdp_txrx_peer_ops.h>
#include <cds_sched.h>
#include "dp_rx.h"
#include "wlan_dp_ucfg_api.h"
#include "wlan_dp_prealloc.h"
#include "wlan_dp_main.h"
#include "wlan_dp_public_struct.h"
#include "wlan_dp_ucfg_api.h"
#include "qdf_nbuf.h"
#include "qdf_threads.h"
#include "qdf_net_if.h"

/* Timeout in ms to wait for a DP rx thread */
#ifdef HAL_CONFIG_SLUB_DEBUG_ON
#define DP_RX_THREAD_WAIT_TIMEOUT 4000
#else
#define DP_RX_THREAD_WAIT_TIMEOUT 2000
#endif
#define DP_RX_THREAD_FLUSH_TIMEOUT 6000
#define DP_RX_THREAD_MIN_FLUSH_TIMEOUT 10

#ifdef CONFIG_SLUB_DEBUG_ON
/* number of rx pkts that thread should yield */
#define DP_RX_THREAD_YIELD_PKT_CNT 20000
#endif

#define DP_RX_TM_DEBUG 0
#if DP_RX_TM_DEBUG
/**
 * dp_rx_tm_walk_skb_list() - Walk skb list and print members
 * @nbuf_list: nbuf list to print
 *
 * Returns: None
 */
static inline void dp_rx_tm_walk_skb_list(qdf_nbuf_t nbuf_list)
{
	qdf_nbuf_t nbuf;
	int i = 0;

	nbuf = nbuf_list;
	while (nbuf) {
		dp_debug("%d nbuf:%pK nbuf->next:%pK nbuf->data:%pK", i,
			 nbuf, qdf_nbuf_next(nbuf), qdf_nbuf_data(nbuf));
		nbuf = qdf_nbuf_next(nbuf);
		i++;
	}
}
#else
static inline void dp_rx_tm_walk_skb_list(qdf_nbuf_t nbuf_list)
{ }
#endif /* DP_RX_TM_DEBUG */

#ifdef DP_RX_REFILL_CPU_PERF_AFFINE_MASK
/**
 * dp_rx_refill_thread_set_affinity - Affine Rx refill threads
 * @refill_thread: Contains over all rx refill thread info
 *
 * Return: None
 */
static void
dp_rx_refill_thread_set_affinity(struct dp_rx_refill_thread *refill_thread)
{
	unsigned int cpus;
	char new_mask_str[10];
	qdf_cpu_mask new_mask;
	int perf_cpu_cluster = hif_get_perf_cluster_bitmap();
	int package_id;

	qdf_cpumask_clear(&new_mask);
	qdf_for_each_online_cpu(cpus) {
		package_id = qdf_topology_physical_package_id(cpus);
		if (package_id >= 0 && BIT(package_id) & perf_cpu_cluster)
			qdf_cpumask_set_cpu(cpus, &new_mask);
	}

	qdf_thread_set_cpus_allowed_mask(refill_thread->task, &new_mask);

	qdf_thread_cpumap_print_to_pagebuf(false, new_mask_str, &new_mask);
	dp_debug("Refill Thread CPU mask  %s", new_mask_str);
}
#else
static void
dp_rx_refill_thread_set_affinity(struct dp_rx_refill_thread *refill_thread)
{
}
#endif
/**
 * dp_rx_tm_get_soc_handle() - get soc handle from struct dp_rx_tm_handle_cmn
 * @rx_tm_handle_cmn: rx thread manager cmn handle
 *
 * Returns: ol_txrx_soc_handle on success, NULL on failure.
 */
static inline ol_txrx_soc_handle
dp_rx_tm_get_soc_handle(struct dp_rx_tm_handle_cmn *rx_tm_handle_cmn)
{
	struct dp_txrx_handle_cmn *txrx_handle_cmn;
	ol_txrx_soc_handle soc;

	txrx_handle_cmn =
		dp_rx_thread_get_txrx_handle(rx_tm_handle_cmn);

	soc = dp_txrx_get_soc_from_ext_handle(txrx_handle_cmn);
	return soc;
}

/**
 * dp_rx_tm_thread_dump_stats() - display stats for a rx_thread
 * @rx_thread: rx_thread pointer for which the stats need to be
 *            displayed
 *
 * Returns: None
 */
static void dp_rx_tm_thread_dump_stats(struct dp_rx_thread *rx_thread)
{
	uint8_t reo_ring_num;
	uint32_t off = 0;
	char nbuf_queued_string[100];
	uint32_t total_queued = 0;
	uint32_t temp = 0;

	qdf_mem_zero(nbuf_queued_string, sizeof(nbuf_queued_string));

	for (reo_ring_num = 0; reo_ring_num < DP_RX_TM_MAX_REO_RINGS;
	     reo_ring_num++) {
		temp = rx_thread->stats.nbuf_queued[reo_ring_num];
		if (!temp)
			continue;
		total_queued += temp;
		if (off >= sizeof(nbuf_queued_string))
			continue;
		off += qdf_scnprintf(&nbuf_queued_string[off],
				     sizeof(nbuf_queued_string) - off,
				     "reo[%u]:%u ", reo_ring_num, temp);
	}

	if (!total_queued)
		return;

	dp_info("thread:%u - qlen:%u queued:(total:%u %s) dequeued:%u stack:%u gro_flushes: %u gro_flushes_by_vdev_del: %u rx_flushes: %u max_len:%u invalid(peer:%u vdev:%u rx-handle:%u others:%u enq fail:%u)",
		rx_thread->id,
		qdf_nbuf_queue_head_qlen(&rx_thread->nbuf_queue),
		total_queued,
		nbuf_queued_string,
		rx_thread->stats.nbuf_dequeued,
		rx_thread->stats.nbuf_sent_to_stack,
		rx_thread->stats.gro_flushes,
		rx_thread->stats.gro_flushes_by_vdev_del,
		rx_thread->stats.rx_flushed,
		rx_thread->stats.nbufq_max_len,
		rx_thread->stats.dropped_invalid_peer,
		rx_thread->stats.dropped_invalid_vdev,
		rx_thread->stats.dropped_invalid_os_rx_handles,
		rx_thread->stats.dropped_others,
		rx_thread->stats.dropped_enq_fail);
}

QDF_STATUS dp_rx_tm_dump_stats(struct dp_rx_tm_handle *rx_tm_hdl)
{
	int i;

	for (i = 0; i < rx_tm_hdl->num_dp_rx_threads; i++) {
		if (!rx_tm_hdl->rx_thread[i])
			continue;
		dp_rx_tm_thread_dump_stats(rx_tm_hdl->rx_thread[i]);
	}
	return QDF_STATUS_SUCCESS;
}

#ifdef FEATURE_ALLOW_PKT_DROPPING
/*
 * dp_check_and_update_pending() - Check and Set RX Pending flag
 * @tm_handle_cmn: DP thread pointer
 *
 * Returns: QDF_STATUS_SUCCESS on success or qdf error code on
 * failure
 */
static inline
QDF_STATUS dp_check_and_update_pending(struct dp_rx_tm_handle_cmn
				       *tm_handle_cmn)
{
	struct dp_txrx_handle_cmn *txrx_handle_cmn;
	struct dp_rx_tm_handle *rx_tm_hdl =
		    (struct dp_rx_tm_handle *)tm_handle_cmn;
	struct dp_soc *dp_soc;
	uint32_t rx_pending_hl_threshold;
	uint32_t rx_pending_lo_threshold;
	uint32_t nbuf_queued_total = 0;
	uint32_t nbuf_dequeued_total = 0;
	uint32_t rx_flushed_total = 0;
	uint32_t pending = 0;
	int i;

	txrx_handle_cmn =
		dp_rx_thread_get_txrx_handle(tm_handle_cmn);
	if (!txrx_handle_cmn) {
		dp_err("invalid txrx_handle_cmn!");
		QDF_BUG(0);
		return QDF_STATUS_E_FAILURE;
	}

	dp_soc = (struct dp_soc *)dp_txrx_get_soc_from_ext_handle(
					txrx_handle_cmn);
	if (!dp_soc) {
		dp_err("invalid soc!");
		QDF_BUG(0);
		return QDF_STATUS_E_FAILURE;
	}

	rx_pending_hl_threshold = wlan_cfg_rx_pending_hl_threshold(
				  dp_soc->wlan_cfg_ctx);
	rx_pending_lo_threshold = wlan_cfg_rx_pending_lo_threshold(
				  dp_soc->wlan_cfg_ctx);

	for (i = 0; i < rx_tm_hdl->num_dp_rx_threads; i++) {
		if (likely(rx_tm_hdl->rx_thread[i])) {
			nbuf_queued_total +=
			    rx_tm_hdl->rx_thread[i]->stats.nbuf_queued_total;
			nbuf_dequeued_total +=
			    rx_tm_hdl->rx_thread[i]->stats.nbuf_dequeued;
			rx_flushed_total +=
			    rx_tm_hdl->rx_thread[i]->stats.rx_flushed;
		}
	}

	if (nbuf_queued_total > (nbuf_dequeued_total + rx_flushed_total))
		pending = nbuf_queued_total - (nbuf_dequeued_total +
					       rx_flushed_total);

	if (unlikely(pending > rx_pending_hl_threshold))
		qdf_atomic_set(&rx_tm_hdl->allow_dropping, 1);
	else if (pending < rx_pending_lo_threshold)
		qdf_atomic_set(&rx_tm_hdl->allow_dropping, 0);

	return QDF_STATUS_SUCCESS;
}

#else
static inline
QDF_STATUS dp_check_and_update_pending(struct dp_rx_tm_handle_cmn
				       *tm_handle_cmn)
{
	return QDF_STATUS_SUCCESS;
}
#endif

/**
 * dp_rx_tm_thread_enqueue() - enqueue nbuf list into rx_thread
 * @rx_thread: rx_thread in which the nbuf needs to be queued
 * @nbuf_list: list of packets to be queued into the thread
 *
 * Enqueue packet into rx_thread and wake it up. The function
 * moves the next pointer of the nbuf_list into the ext list of
 * the first nbuf for storage into the thread. Only the first
 * nbuf is queued into the thread nbuf queue. The reverse is
 * done at the time of dequeue.
 *
 * Returns: QDF_STATUS_SUCCESS on success or qdf error code on
 * failure
 */
static QDF_STATUS dp_rx_tm_thread_enqueue(struct dp_rx_thread *rx_thread,
					  qdf_nbuf_t nbuf_list)
{
	qdf_nbuf_t head_ptr, next_ptr_list;
	uint32_t temp_qlen;
	uint32_t num_elements_in_nbuf;
	uint32_t nbuf_queued;
	struct dp_rx_tm_handle_cmn *tm_handle_cmn;
	uint8_t reo_ring_num = QDF_NBUF_CB_RX_CTX_ID(nbuf_list);
	qdf_wait_queue_head_t *wait_q_ptr;
	uint8_t allow_dropping;

	tm_handle_cmn = rx_thread->rtm_handle_cmn;

	if (!tm_handle_cmn) {
		dp_alert("tm_handle_cmn is null!");
		QDF_BUG(0);
		return QDF_STATUS_E_FAILURE;
	}

	wait_q_ptr = &rx_thread->wait_q;

	if (reo_ring_num >= DP_RX_TM_MAX_REO_RINGS) {
		dp_alert("incorrect ring %u", reo_ring_num);
		QDF_BUG(0);
		return QDF_STATUS_E_FAILURE;
	}

	num_elements_in_nbuf = QDF_NBUF_CB_RX_NUM_ELEMENTS_IN_LIST(nbuf_list);
	nbuf_queued = num_elements_in_nbuf;

	allow_dropping = qdf_atomic_read(
		&((struct dp_rx_tm_handle *)tm_handle_cmn)->allow_dropping);
	if (unlikely(allow_dropping)) {
		qdf_nbuf_list_free(nbuf_list);
		rx_thread->stats.dropped_enq_fail += num_elements_in_nbuf;
		nbuf_queued = 0;
		goto enq_done;
	}

	dp_rx_tm_walk_skb_list(nbuf_list);

	head_ptr = nbuf_list;

	/* Ensure head doesn't have an ext list */
	while (qdf_unlikely(head_ptr && qdf_nbuf_get_ext_list(head_ptr))) {
		QDF_NBUF_CB_RX_NUM_ELEMENTS_IN_LIST(head_ptr) = 1;
		num_elements_in_nbuf--;
		next_ptr_list = head_ptr->next;
		qdf_nbuf_set_next(head_ptr, NULL);
		/* count aggregated RX frame into enqueued stats */
		nbuf_queued += qdf_nbuf_get_gso_segs(head_ptr);
		qdf_nbuf_queue_head_enqueue_tail(&rx_thread->nbuf_queue,
						 head_ptr);
		head_ptr = next_ptr_list;
	}

	if (!head_ptr)
		goto enq_done;

	QDF_NBUF_CB_RX_NUM_ELEMENTS_IN_LIST(head_ptr) = num_elements_in_nbuf;

	next_ptr_list = head_ptr->next;

	if (next_ptr_list) {
		/* move ->next pointer to ext list */
		qdf_nbuf_append_ext_list(head_ptr, next_ptr_list, 0);
		dp_debug("appended next_ptr_list %pK to nbuf %pK ext list %pK",
			 qdf_nbuf_next(nbuf_list), nbuf_list,
			 qdf_nbuf_get_ext_list(nbuf_list));
	}
	qdf_nbuf_set_next(head_ptr, NULL);

	qdf_nbuf_queue_head_enqueue_tail(&rx_thread->nbuf_queue, head_ptr);

enq_done:
	temp_qlen = qdf_nbuf_queue_head_qlen(&rx_thread->nbuf_queue);

	rx_thread->stats.nbuf_queued[reo_ring_num] += nbuf_queued;
	rx_thread->stats.nbuf_queued_total += nbuf_queued;

	dp_check_and_update_pending(tm_handle_cmn);

	if (temp_qlen > rx_thread->stats.nbufq_max_len)
		rx_thread->stats.nbufq_max_len = temp_qlen;

	dp_debug("enqueue packet thread %pK wait queue %pK qlen %u",
		 rx_thread, wait_q_ptr,
		 qdf_nbuf_queue_head_qlen(&rx_thread->nbuf_queue));

	qdf_set_bit(RX_POST_EVENT, &rx_thread->event_flag);
	qdf_wake_up_interruptible(wait_q_ptr);

	return QDF_STATUS_SUCCESS;
}

/**
 * dp_rx_tm_thread_gro_flush_ind() - Rxthread flush ind post
 * @rx_thread: rx_thread in which the flush needs to be handled
 * @flush_code: flush code to differentiate low TPUT flush
 *
 * Return: QDF_STATUS_SUCCESS on success or qdf error code on
 * failure
 */
static QDF_STATUS
dp_rx_tm_thread_gro_flush_ind(struct dp_rx_thread *rx_thread,
			      enum dp_rx_gro_flush_code flush_code)
{
	struct dp_rx_tm_handle_cmn *tm_handle_cmn;
	qdf_wait_queue_head_t *wait_q_ptr;

	tm_handle_cmn = rx_thread->rtm_handle_cmn;
	wait_q_ptr = &rx_thread->wait_q;

	qdf_atomic_set(&rx_thread->gro_flush_ind, flush_code);

	dp_debug("Flush indication received");

	qdf_set_bit(RX_POST_EVENT, &rx_thread->event_flag);
	qdf_wake_up_interruptible(wait_q_ptr);
	return QDF_STATUS_SUCCESS;
}

/**
 * dp_rx_thread_adjust_nbuf_list() - create an nbuf list from the frag list
 * @head: nbuf list to be created
 *
 * Returns: void
 */
static void dp_rx_thread_adjust_nbuf_list(qdf_nbuf_t head)
{
	qdf_nbuf_t next_ptr_list, nbuf_list;

	nbuf_list = head;
	if (head && QDF_NBUF_CB_RX_NUM_ELEMENTS_IN_LIST(head) > 1) {
		/* move ext list to ->next pointer */
		next_ptr_list = qdf_nbuf_get_ext_list(head);
		qdf_nbuf_append_ext_list(head, NULL, 0);
		qdf_nbuf_set_next(nbuf_list, next_ptr_list);
		dp_rx_tm_walk_skb_list(nbuf_list);
	}
}

/**
 * dp_rx_tm_thread_dequeue() - dequeue nbuf list from rx_thread
 * @rx_thread: rx_thread from which the nbuf needs to be dequeued
 *
 * Returns: nbuf or nbuf_list dequeued from rx_thread
 */
static qdf_nbuf_t dp_rx_tm_thread_dequeue(struct dp_rx_thread *rx_thread)
{
	qdf_nbuf_t head;

	head = qdf_nbuf_queue_head_dequeue(&rx_thread->nbuf_queue);
	dp_rx_thread_adjust_nbuf_list(head);

	dp_debug("Dequeued %pK nbuf_list", head);
	return head;
}

#ifdef CONFIG_SLUB_DEBUG_ON
/**
 * dp_rx_thread_should_yield() - check whether rx loop should yield
 * @iter: iteration of packets received
 * @rx_thread: rx_thread which should yield
 *
 * Returns: should yield or not
 */
static inline bool dp_rx_thread_should_yield(struct dp_rx_thread *rx_thread,
					     uint32_t iter)
{
	if (iter >= DP_RX_THREAD_YIELD_PKT_CNT ||
	    qdf_test_bit(RX_VDEV_DEL_EVENT, &rx_thread->event_flag))
		return true;
	return false;
}
#else
static inline bool dp_rx_thread_should_yield(struct dp_rx_thread *rx_thread,
					     uint32_t iter)
{
	return false;
}
#endif

/**
 * dp_rx_thread_process_nbufq() - process nbuf queue of a thread
 * @rx_thread: rx_thread whose nbuf queue needs to be processed
 *
 * Returns: 0 on success, error code on failure
 */
static int dp_rx_thread_process_nbufq(struct dp_rx_thread *rx_thread)
{
	qdf_nbuf_t nbuf_list;
	uint8_t vdev_id;
	ol_txrx_rx_fp stack_fn;
	ol_osif_vdev_handle osif_vdev;
	ol_txrx_soc_handle soc;
	uint32_t num_list_elements = 0;
	uint32_t iterates = 0;

	struct dp_txrx_handle_cmn *txrx_handle_cmn;

	txrx_handle_cmn =
		dp_rx_thread_get_txrx_handle(rx_thread->rtm_handle_cmn);

	soc = dp_txrx_get_soc_from_ext_handle(txrx_handle_cmn);
	if (!soc) {
		dp_err("invalid soc!");
		QDF_BUG(0);
		return -EFAULT;
	}

	dp_debug("enter: qlen  %u",
		 qdf_nbuf_queue_head_qlen(&rx_thread->nbuf_queue));

	nbuf_list = dp_rx_tm_thread_dequeue(rx_thread);
	while (nbuf_list) {
		num_list_elements =
			QDF_NBUF_CB_RX_NUM_ELEMENTS_IN_LIST(nbuf_list);
		/* count aggregated RX frame into stats */
		num_list_elements += qdf_nbuf_get_gso_segs(nbuf_list);
		rx_thread->stats.nbuf_dequeued += num_list_elements;
		iterates += num_list_elements;

		vdev_id = QDF_NBUF_CB_RX_VDEV_ID(nbuf_list);
		cdp_get_os_rx_handles_from_vdev(soc, vdev_id, &stack_fn,
						&osif_vdev);
		dp_debug("rx_thread %pK sending packet %pK to stack",
			 rx_thread, nbuf_list);
		if (!stack_fn || !osif_vdev ||
		    QDF_STATUS_SUCCESS != stack_fn(osif_vdev, nbuf_list)) {
			rx_thread->stats.dropped_invalid_os_rx_handles +=
							num_list_elements;
			qdf_nbuf_list_free(nbuf_list);
		} else {
			rx_thread->stats.nbuf_sent_to_stack +=
							num_list_elements;
		}
		if (qdf_unlikely(dp_rx_thread_should_yield(rx_thread,
							   iterates))) {
			rx_thread->stats.rx_nbufq_loop_yield++;
			break;
		}
		nbuf_list = dp_rx_tm_thread_dequeue(rx_thread);
	}

	dp_debug("exit: qlen  %u",
		 qdf_nbuf_queue_head_qlen(&rx_thread->nbuf_queue));

	return 0;
}

/**
 * dp_rx_thread_gro_flush() - flush GRO packets for the RX thread
 * @rx_thread: rx_thread to be processed
 * @gro_flush_code: flush code to differentiating flushes
 *
 * Return: void
 */
static void dp_rx_thread_gro_flush(struct dp_rx_thread *rx_thread,
				   enum dp_rx_gro_flush_code gro_flush_code)
{
	struct wlan_dp_psoc_context *dp_ctx;

	dp_ctx =  dp_get_context();
	if (!dp_ctx) {
		dp_err("DP context is NULL");
		return;
	}
	dp_debug("flushing packets for thread %u", rx_thread->id);
	qdf_local_bh_disable();
	dp_ctx->dp_ops.dp_rx_thread_napi_gro_flush(&rx_thread->napi,
						   gro_flush_code);
	qdf_local_bh_enable();
	rx_thread->stats.gro_flushes++;
}

/**
 * dp_rx_should_flush() - Determines whether the RX thread should be flushed.
 * @rx_thread: rx_thread to be processed
 *
 * Return: enum dp_rx_gro_flush_code
 */
static inline enum dp_rx_gro_flush_code
dp_rx_should_flush(struct dp_rx_thread *rx_thread)
{
	enum dp_rx_gro_flush_code gro_flush_code;

	gro_flush_code = qdf_atomic_read(&rx_thread->gro_flush_ind);
	if (qdf_atomic_test_bit(RX_VDEV_DEL_EVENT, &rx_thread->event_flag))
		gro_flush_code = DP_RX_GRO_NORMAL_FLUSH;

	return gro_flush_code;
}

/**
 * dp_rx_thread_sub_loop() - rx thread subloop
 * @rx_thread: rx_thread to be processed
 * @shutdown: pointer to shutdown variable
 *
 * The function handles shutdown and suspend events from other
 * threads and processes nbuf queue of a rx thread. In case a
 * shutdown event is received from some other wlan thread, the
 * function sets the shutdown pointer to true and returns
 *
 * Returns: 0 on success, error code on failure
 */
static int dp_rx_thread_sub_loop(struct dp_rx_thread *rx_thread, bool *shutdown)
{
	enum dp_rx_gro_flush_code gro_flush_code;

	while (true) {
		if (qdf_atomic_test_and_clear_bit(RX_SHUTDOWN_EVENT,
						  &rx_thread->event_flag)) {
			if (qdf_atomic_test_and_clear_bit(RX_SUSPEND_EVENT,
							  &rx_thread->event_flag)) {
				qdf_event_set(&rx_thread->suspend_event);
			}
			dp_debug("shutting down (%s) id %d pid %d",
				 qdf_get_current_comm(), rx_thread->id,
				 qdf_get_current_pid());
			*shutdown = true;
			break;
		}

		dp_rx_thread_process_nbufq(rx_thread);

		gro_flush_code = dp_rx_should_flush(rx_thread);
		/* Only flush when gro_flush_code is either
		 * DP_RX_GRO_NORMAL_FLUSH or DP_RX_GRO_LOW_TPUT_FLUSH
		 */
		if (gro_flush_code != DP_RX_GRO_NOT_FLUSH) {
			dp_rx_thread_gro_flush(rx_thread, gro_flush_code);
			qdf_atomic_set(&rx_thread->gro_flush_ind, 0);
		}

		if (qdf_atomic_test_and_clear_bit(RX_VDEV_DEL_EVENT,
						  &rx_thread->event_flag)) {
			rx_thread->stats.gro_flushes_by_vdev_del++;
			qdf_event_set(&rx_thread->vdev_del_event);
			if (qdf_nbuf_queue_head_qlen(&rx_thread->nbuf_queue))
				continue;
		}

		if (qdf_atomic_test_and_clear_bit(RX_SUSPEND_EVENT,
						  &rx_thread->event_flag)) {
			dp_debug("received suspend ind (%s) id %d pid %d",
				 qdf_get_current_comm(), rx_thread->id,
				 qdf_get_current_pid());
			qdf_event_set(&rx_thread->suspend_event);
			dp_debug("waiting for resume (%s) id %d pid %d",
				 qdf_get_current_comm(), rx_thread->id,
				 qdf_get_current_pid());
			qdf_wait_single_event(&rx_thread->resume_event, 0);
		}
		break;
	}
	return 0;
}

/**
 * dp_rx_thread_loop() - main dp rx thread loop
 * @arg: pointer to dp_rx_thread structure for the rx thread
 *
 * Return: thread exit code
 */
static int dp_rx_thread_loop(void *arg)
{
	struct dp_rx_thread *rx_thread = arg;
	bool shutdown = false;
	int status;
	QDF_STATUS status_intr;
	struct dp_rx_tm_handle_cmn *tm_handle_cmn;

	if (!arg) {
		dp_err("bad Args passed");
		return 0;
	}

	tm_handle_cmn = rx_thread->rtm_handle_cmn;

	qdf_set_user_nice(qdf_get_current_task(), -1);
	qdf_set_wake_up_idle(true);

	qdf_event_set(&rx_thread->start_event);
	dp_info("starting rx_thread (%s) id %d pid %d", qdf_get_current_comm(),
		rx_thread->id, qdf_get_current_pid());
	while (!shutdown) {
		/* This implements the execution model algorithm */
		dp_debug("sleeping");
		status =
		    qdf_wait_queue_interruptible
				(rx_thread->wait_q,
				 qdf_atomic_test_bit(RX_POST_EVENT,
						     &rx_thread->event_flag) ||
				 qdf_atomic_test_bit(RX_SUSPEND_EVENT,
						     &rx_thread->event_flag) ||
				 qdf_atomic_test_bit(RX_VDEV_DEL_EVENT,
						     &rx_thread->event_flag));
		dp_debug("woken up");
		status_intr = qdf_status_from_os_return(status);
		if (status_intr == QDF_STATUS_E_RESTART) {
			QDF_DEBUG_PANIC("wait_event_interruptible returned -ERESTARTSYS");
			break;
		}
		qdf_atomic_clear_bit(RX_POST_EVENT, &rx_thread->event_flag);
		dp_rx_thread_sub_loop(rx_thread, &shutdown);
	}

	/* If we get here the scheduler thread must exit */
	dp_info("exiting (%s) id %d pid %d", qdf_get_current_comm(),
		rx_thread->id, qdf_get_current_pid());
	qdf_event_set(&rx_thread->shutdown_event);

	return 0;
}

static int dp_rx_refill_thread_sub_loop(struct dp_rx_refill_thread *rx_thread,
					bool *shutdown)
{
	while (true) {
		if (qdf_atomic_test_and_clear_bit(RX_REFILL_SHUTDOWN_EVENT,
						  &rx_thread->event_flag)) {
			if (qdf_atomic_test_and_clear_bit(RX_REFILL_SUSPEND_EVENT,
							  &rx_thread->event_flag)) {
				qdf_event_set(&rx_thread->suspend_event);
			}
			dp_debug("shutting down (%s) pid %d",
				 qdf_get_current_comm(), qdf_get_current_pid());
			*shutdown = true;
			break;
		}

		dp_rx_refill_buff_pool_enqueue((struct dp_soc *)rx_thread->soc);

		if (qdf_atomic_test_and_clear_bit(RX_REFILL_SUSPEND_EVENT,
						  &rx_thread->event_flag)) {
			dp_debug("refill thread received suspend ind (%s) pid %d",
				 qdf_get_current_comm(),
				 qdf_get_current_pid());
			qdf_event_set(&rx_thread->suspend_event);
			dp_debug("refill thread waiting for resume (%s) pid %d",
				 qdf_get_current_comm(),
				 qdf_get_current_pid());
			qdf_wait_single_event(&rx_thread->resume_event, 0);
		}
		break;
	}
	return 0;
}

static int dp_rx_refill_thread_loop(void *arg)
{
	struct dp_rx_refill_thread *rx_thread = arg;
	bool shutdown = false;
	int status;
	QDF_STATUS status_intr;

	if (!arg) {
		dp_err("bad Args passed");
		return 0;
	}

	qdf_set_user_nice(qdf_get_current_task(), -1);
	qdf_set_wake_up_idle(true);

	qdf_event_set(&rx_thread->start_event);
	dp_info("starting rx_refill_thread (%s) pid %d", qdf_get_current_comm(),
		qdf_get_current_pid());
	while (!shutdown) {
		/* This implements the execution model algorithm */
		status =
		    qdf_wait_queue_interruptible
				(rx_thread->wait_q,
				 qdf_atomic_test_bit(RX_REFILL_POST_EVENT,
						     &rx_thread->event_flag) ||
				 qdf_atomic_test_bit(RX_REFILL_SUSPEND_EVENT,
						     &rx_thread->event_flag));

		status_intr = qdf_status_from_os_return(status);
		if (status_intr == QDF_STATUS_E_RESTART) {
			QDF_DEBUG_PANIC("wait_event_interruptible returned -ERESTARTSYS");
			break;
		}
		qdf_atomic_clear_bit(RX_REFILL_POST_EVENT,
				     &rx_thread->event_flag);
		dp_rx_refill_thread_sub_loop(rx_thread, &shutdown);
	}

	/* If we get here the scheduler thread must exit */
	dp_info("exiting (%s) pid %d", qdf_get_current_comm(),
		qdf_get_current_pid());
	qdf_event_set(&rx_thread->shutdown_event);

	return 0;
}

/**
 * dp_rx_tm_thread_napi_poll() - dummy napi poll for rx_thread NAPI
 * @napi: pointer to DP rx_thread NAPI
 * @budget: NAPI BUDGET
 *
 * Return: 0 as it is not supposed to be polled at all as it is not scheduled.
 */
static int dp_rx_tm_thread_napi_poll(qdf_napi_struct *napi, int budget)
{
	QDF_DEBUG_PANIC("this napi_poll should not be polled as we don't schedule it");

	return 0;
}

/**
 * dp_rx_tm_thread_napi_init() - Initialize dummy rx_thread NAPI
 * @rx_thread: dp_rx_thread structure containing dummy napi and netdev
 *
 * Return: None
 */
static void dp_rx_tm_thread_napi_init(struct dp_rx_thread *rx_thread)
{
	/* Todo - optimize to use only one dummy netdev for all thread napis */
	qdf_net_if_create_dummy_if((struct qdf_net_if *)&rx_thread->netdev);
	qdf_netif_napi_add(&rx_thread->netdev, &rx_thread->napi,
			   dp_rx_tm_thread_napi_poll, 64);
	qdf_napi_enable(&rx_thread->napi);
}

/**
 * dp_rx_tm_thread_napi_deinit() - De-initialize dummy rx_thread NAPI
 * @rx_thread: dp_rx_thread handle containing dummy napi and netdev
 *
 * Return: None
 */
static void dp_rx_tm_thread_napi_deinit(struct dp_rx_thread *rx_thread)
{
	qdf_netif_napi_del(&rx_thread->napi);
}

/*
 * dp_rx_tm_thread_init() - Initialize dp_rx_thread structure and thread
 *
 * @rx_thread: dp_rx_thread structure to be initialized
 * @id: id of the thread to be initialized
 *
 * Return: QDF_STATUS on success, QDF error code on failure
 */
static QDF_STATUS dp_rx_tm_thread_init(struct dp_rx_thread *rx_thread,
				       uint8_t id)
{
	char thread_name[15];
	QDF_STATUS qdf_status;

	qdf_mem_zero(thread_name, sizeof(thread_name));

	if (!rx_thread) {
		dp_err("rx_thread is null!");
		return QDF_STATUS_E_FAULT;
	}
	rx_thread->id = id;
	rx_thread->event_flag = 0;
	qdf_nbuf_queue_head_init(&rx_thread->nbuf_queue);
	qdf_event_create(&rx_thread->start_event);
	qdf_event_create(&rx_thread->suspend_event);
	qdf_event_create(&rx_thread->resume_event);
	qdf_event_create(&rx_thread->shutdown_event);
	qdf_event_create(&rx_thread->vdev_del_event);
	qdf_atomic_init(&rx_thread->gro_flush_ind);
	qdf_init_waitqueue_head(&rx_thread->wait_q);
	qdf_scnprintf(thread_name, sizeof(thread_name), "dp_rx_thread_%u", id);
	dp_info("%s %u", thread_name, id);

	if (cdp_cfg_get(dp_rx_tm_get_soc_handle(rx_thread->rtm_handle_cmn),
			cfg_dp_gro_enable))
		dp_rx_tm_thread_napi_init(rx_thread);

	rx_thread->task = qdf_create_thread(dp_rx_thread_loop,
					    rx_thread, thread_name);
	if (!rx_thread->task) {
		dp_err("could not create dp_rx_thread %d", id);
		return QDF_STATUS_E_FAILURE;
	}

	qdf_wake_up_process(rx_thread->task);
	qdf_status = qdf_wait_single_event(&rx_thread->start_event, 0);

	if (!QDF_IS_STATUS_SUCCESS(qdf_status)) {
		dp_err("failed waiting for thread creation id %d", id);
		return QDF_STATUS_E_FAILURE;
	}
	return QDF_STATUS_SUCCESS;
}

/*
 * dp_rx_tm_thread_deinit() - De-Initialize dp_rx_thread structure and thread
 * @rx_thread: dp_rx_thread structure to be de-initialized
 * @id: id of the thread to be initialized
 *
 * Return: QDF_STATUS_SUCCESS
 */
static QDF_STATUS dp_rx_tm_thread_deinit(struct dp_rx_thread *rx_thread)
{
	qdf_event_destroy(&rx_thread->start_event);
	qdf_event_destroy(&rx_thread->suspend_event);
	qdf_event_destroy(&rx_thread->resume_event);
	qdf_event_destroy(&rx_thread->shutdown_event);
	qdf_event_destroy(&rx_thread->vdev_del_event);

	if (cdp_cfg_get(dp_rx_tm_get_soc_handle(rx_thread->rtm_handle_cmn),
			cfg_dp_gro_enable))
		dp_rx_tm_thread_napi_deinit(rx_thread);

	return QDF_STATUS_SUCCESS;
}

QDF_STATUS dp_rx_refill_thread_init(struct dp_rx_refill_thread *refill_thread)
{
	char refill_thread_name[20] = {0};
	QDF_STATUS qdf_status = QDF_STATUS_SUCCESS;

	qdf_scnprintf(refill_thread_name, sizeof(refill_thread_name),
		      "dp_refill_thread");
	dp_info("Initializing %s", refill_thread_name);

	refill_thread->state = DP_RX_REFILL_THREAD_INVALID;
	refill_thread->event_flag = 0;
	qdf_event_create(&refill_thread->start_event);
	qdf_event_create(&refill_thread->suspend_event);
	qdf_event_create(&refill_thread->resume_event);
	qdf_event_create(&refill_thread->shutdown_event);
	qdf_init_waitqueue_head(&refill_thread->wait_q);
	refill_thread->task = qdf_create_thread(dp_rx_refill_thread_loop,
						refill_thread,
						refill_thread_name);
	if (!refill_thread->task) {
		dp_err("could not create dp_rx_refill_thread");
		return QDF_STATUS_E_FAILURE;
	}
	qdf_wake_up_process(refill_thread->task);
	qdf_status = qdf_wait_single_event(&refill_thread->start_event,
					   0);
	if (!QDF_IS_STATUS_SUCCESS(qdf_status)) {
		dp_err("failed waiting for refill thread creation status: %d",
		       qdf_status);
		return QDF_STATUS_E_FAILURE;
	}

	dp_rx_refill_thread_set_affinity(refill_thread);

	refill_thread->state = DP_RX_REFILL_THREAD_RUNNING;
	return QDF_STATUS_SUCCESS;
}

QDF_STATUS dp_rx_refill_thread_deinit(struct dp_rx_refill_thread *refill_thread)
{
	qdf_set_bit(RX_REFILL_SHUTDOWN_EVENT,
		    &refill_thread->event_flag);
	qdf_set_bit(RX_REFILL_POST_EVENT,
		    &refill_thread->event_flag);
	qdf_wake_up_interruptible(&refill_thread->wait_q);
	qdf_wait_single_event(&refill_thread->shutdown_event, 0);

	qdf_event_destroy(&refill_thread->start_event);
	qdf_event_destroy(&refill_thread->suspend_event);
	qdf_event_destroy(&refill_thread->resume_event);
	qdf_event_destroy(&refill_thread->shutdown_event);

	refill_thread->state = DP_RX_REFILL_THREAD_INVALID;
	return QDF_STATUS_SUCCESS;
}

QDF_STATUS dp_rx_tm_init(struct dp_rx_tm_handle *rx_tm_hdl,
			 uint8_t num_dp_rx_threads)
{
	int i;
	QDF_STATUS qdf_status = QDF_STATUS_SUCCESS;

	if (num_dp_rx_threads > DP_MAX_RX_THREADS) {
		dp_err("unable to initialize %u number of threads. MAX %u",
		       num_dp_rx_threads, DP_MAX_RX_THREADS);
		return QDF_STATUS_E_INVAL;
	}

	rx_tm_hdl->num_dp_rx_threads = num_dp_rx_threads;
	rx_tm_hdl->state = DP_RX_THREADS_INVALID;

	dp_info("initializing %u threads", num_dp_rx_threads);

	/* allocate an array to contain the DP RX thread pointers */
	rx_tm_hdl->rx_thread = qdf_mem_malloc(num_dp_rx_threads *
					      sizeof(struct dp_rx_thread *));

	if (qdf_unlikely(!rx_tm_hdl->rx_thread)) {
		qdf_status = QDF_STATUS_E_NOMEM;
		goto ret;
	}

	for (i = 0; i < rx_tm_hdl->num_dp_rx_threads; i++) {
		rx_tm_hdl->rx_thread[i] =
			(struct dp_rx_thread *)
			qdf_mem_malloc(sizeof(struct dp_rx_thread));
		if (qdf_unlikely(!rx_tm_hdl->rx_thread[i])) {
			QDF_ASSERT(0);
			qdf_status = QDF_STATUS_E_NOMEM;
			goto ret;
		}
		rx_tm_hdl->rx_thread[i]->rtm_handle_cmn =
				(struct dp_rx_tm_handle_cmn *)rx_tm_hdl;
		qdf_status =
			dp_rx_tm_thread_init(rx_tm_hdl->rx_thread[i], i);
		if (!QDF_IS_STATUS_SUCCESS(qdf_status))
			break;
	}
ret:
	if (!QDF_IS_STATUS_SUCCESS(qdf_status))
		dp_rx_tm_deinit(rx_tm_hdl);
	else
		rx_tm_hdl->state = DP_RX_THREADS_RUNNING;

	return qdf_status;
}

/**
 * dp_rx_tm_suspend() - suspend DP RX threads
 * @rx_tm_hdl: dp_rx_tm_handle containing the overall thread
 *            infrastructure
 *
 * Return: Success/Failure
 */
QDF_STATUS dp_rx_tm_suspend(struct dp_rx_tm_handle *rx_tm_hdl)
{
	int i;
	QDF_STATUS qdf_status;
	struct dp_rx_thread *rx_thread;

	if (rx_tm_hdl->state == DP_RX_THREADS_SUSPENDED) {
		dp_info("already in suspend state! Ignoring.");
		return QDF_STATUS_E_INVAL;
	}

	rx_tm_hdl->state = DP_RX_THREADS_SUSPENDING;

	for (i = 0; i < rx_tm_hdl->num_dp_rx_threads; i++) {
		if (!rx_tm_hdl->rx_thread[i])
			continue;
		qdf_event_reset(&rx_tm_hdl->rx_thread[i]->resume_event);
		qdf_event_reset(&rx_tm_hdl->rx_thread[i]->suspend_event);
		qdf_set_bit(RX_SUSPEND_EVENT,
			    &rx_tm_hdl->rx_thread[i]->event_flag);
		qdf_wake_up_interruptible(&rx_tm_hdl->rx_thread[i]->wait_q);
	}

	for (i = 0; i < rx_tm_hdl->num_dp_rx_threads; i++) {
		rx_thread = rx_tm_hdl->rx_thread[i];
		if (!rx_thread)
			continue;
		dp_debug("thread %d", i);
		qdf_status = qdf_wait_single_event(&rx_thread->suspend_event,
						   DP_RX_THREAD_WAIT_TIMEOUT);
		if (QDF_IS_STATUS_SUCCESS(qdf_status))
			dp_debug("thread:%d suspended", rx_thread->id);
		else
			goto suspend_fail;
	}
	rx_tm_hdl->state = DP_RX_THREADS_SUSPENDED;

	return QDF_STATUS_SUCCESS;

suspend_fail:
	dp_err("thread:%d %s(%d) while waiting for suspend",
	       rx_thread->id,
	       qdf_status == QDF_STATUS_E_TIMEOUT ? "timeout out" : "failed",
	       qdf_status);

	dp_rx_tm_resume(rx_tm_hdl);

	return qdf_status;
}

/**
 * dp_rx_refill_thread_suspend() - Suspend DP RX refill threads
 * @refill_thread: containing the overall refill thread infrastructure
 *
 * Return: Success/Failure
 */
QDF_STATUS
dp_rx_refill_thread_suspend(struct dp_rx_refill_thread *refill_thread)
{
	QDF_STATUS qdf_status;

	if (refill_thread->state == DP_RX_REFILL_THREAD_SUSPENDED) {
		dp_info("already in suspend state! Ignoring.");
		return QDF_STATUS_E_INVAL;
	}

	refill_thread->state = DP_RX_REFILL_THREAD_SUSPENDING;

	qdf_event_reset(&refill_thread->resume_event);
	qdf_event_reset(&refill_thread->suspend_event);
	qdf_set_bit(RX_REFILL_SUSPEND_EVENT,
		    &refill_thread->event_flag);
	qdf_wake_up_interruptible(&refill_thread->wait_q);

	qdf_status = qdf_wait_single_event(&refill_thread->suspend_event,
					   DP_RX_THREAD_WAIT_TIMEOUT);
	if (QDF_IS_STATUS_SUCCESS(qdf_status))
		dp_debug("Refill thread  suspended");
	else
		goto suspend_fail;

	refill_thread->state = DP_RX_REFILL_THREAD_SUSPENDED;
	return QDF_STATUS_SUCCESS;

suspend_fail:
	dp_err("Refill thread %s(%d) while waiting for suspend",
	       qdf_status == QDF_STATUS_E_TIMEOUT ? "timeout out" : "failed",
	       qdf_status);

	dp_rx_refill_thread_resume(refill_thread);

	return qdf_status;
}

/*
 * dp_rx_tm_flush_nbuf_list() - Flush rx thread nbuf list
 * @rx_tm_hdl: dp_rx_tm_handle containing the overall thread
 *  infrastructure
 * @vdev_id: vdev id for which packets are to be flushed
 *
 * Return: None
 */
static inline void
dp_rx_tm_flush_nbuf_list(struct dp_rx_tm_handle *rx_tm_hdl, uint8_t vdev_id)
{
	qdf_nbuf_t nbuf_list, tmp_nbuf_list;
	uint32_t num_list_elements = 0;
	uint64_t lock_time, unlock_time, flush_time;
	qdf_nbuf_t nbuf_list_head = NULL, nbuf_list_next;
	struct dp_rx_thread *rx_thread;
	int i;

	for (i = 0; i < rx_tm_hdl->num_dp_rx_threads; i++) {
		rx_thread = rx_tm_hdl->rx_thread[i];
		if (!rx_thread)
			continue;

		qdf_nbuf_queue_head_lock(&rx_thread->nbuf_queue);
		lock_time = qdf_get_log_timestamp();
		QDF_NBUF_QUEUE_WALK_SAFE(&rx_thread->nbuf_queue, nbuf_list,
					 tmp_nbuf_list) {
			if (QDF_NBUF_CB_RX_VDEV_ID(nbuf_list) == vdev_id) {
				qdf_nbuf_unlink_no_lock(nbuf_list,
							&rx_thread->nbuf_queue);
				DP_RX_HEAD_APPEND(nbuf_list_head, nbuf_list);
			}
		}
		qdf_nbuf_queue_head_unlock(&rx_thread->nbuf_queue);
		unlock_time = qdf_get_log_timestamp();

		while (nbuf_list_head) {
			nbuf_list_next = qdf_nbuf_queue_next(nbuf_list_head);
			qdf_nbuf_set_next(nbuf_list_head, NULL);
			dp_rx_thread_adjust_nbuf_list(nbuf_list_head);
			num_list_elements =
			    QDF_NBUF_CB_RX_NUM_ELEMENTS_IN_LIST(nbuf_list_head);
			rx_thread->stats.rx_flushed += num_list_elements;
			qdf_nbuf_list_free(nbuf_list_head);
			nbuf_list_head = nbuf_list_next;
		}

		flush_time = qdf_get_log_timestamp();
		dp_info("Thread: %u lock held time: %llu us flush time: %llu us",
			rx_thread->id,
			qdf_log_timestamp_to_usecs(unlock_time - lock_time),
			qdf_log_timestamp_to_usecs(flush_time - unlock_time));

		qdf_event_reset(&rx_thread->vdev_del_event);
		qdf_set_bit(RX_VDEV_DEL_EVENT, &rx_thread->event_flag);
		qdf_wake_up_interruptible(&rx_thread->wait_q);
		}
}

/**
 * dp_rx_tm_wait_vdev_del_event() - wait on rx thread vdev delete event
 * @rx_tm_hdl: dp_rx_tm_handle containing the overall thread
 *  infrastructure
 *
 * Return: None
 */
static inline void
dp_rx_tm_wait_vdev_del_event(struct dp_rx_tm_handle *rx_tm_hdl)
{
	QDF_STATUS qdf_status;
	int i;
	struct dp_rx_thread *rx_thread;
	int wait_timeout = DP_RX_THREAD_FLUSH_TIMEOUT;
	uint64_t entry_time, exit_time, wait_time;

	for (i = 0; i < rx_tm_hdl->num_dp_rx_threads; i++) {
		rx_thread = rx_tm_hdl->rx_thread[i];
		if (!rx_thread)
			continue;

		entry_time = qdf_get_log_timestamp();
		qdf_status =
			qdf_wait_single_event(&rx_thread->vdev_del_event,
					      wait_timeout);

		if (QDF_IS_STATUS_SUCCESS(qdf_status))
			dp_debug("thread:%d napi gro flush successfully",
				 rx_thread->id);
		else if (qdf_status == QDF_STATUS_E_TIMEOUT) {
			dp_err("thread:%d timed out waiting for napi gro flush",
			       rx_thread->id);
			/*
			 * If timeout, then force flush in case any rx packets
			 * belong to this vdev is still pending on stack queue,
			 * while net_vdev will be freed soon.
			 */
			dp_rx_thread_gro_flush(rx_thread,
					       DP_RX_GRO_NORMAL_FLUSH);
		} else {
			dp_err("thread:%d event wait failed with status: %d",
			       rx_thread->id, qdf_status);
		}

		exit_time = qdf_get_log_timestamp();
		wait_time = qdf_do_div(qdf_log_timestamp_to_usecs(exit_time -
								  entry_time),
				       QDF_USEC_PER_MSEC);
		/**
		 *
		 * Maximum wait timeout to flush all thread is
		 * DP_RX_THREAD_MIN_FLUSH_TIMEOUT, This logic
		 * limit maximum wait time of all the thread to
		 * DP_RX_THREAD_FLUSH_TIMEOUT. In case if
		 * DP_RX_THREAD_FLUSH_TIMEOUT is exhausted
		 * give remaining thread DP_RX_THREAD_MIN_FLUSH_TIMEOUT.
		 *
		 * Since all the threads are already woken before calling
		 * wait API. So each thread will get
		 * DP_RX_THREAD_FLUSH_TIMEOUT to set the event.
		 */
		if (wait_timeout > DP_RX_THREAD_MIN_FLUSH_TIMEOUT)
			wait_timeout = (wait_timeout > wait_time) ?
				       (wait_timeout - wait_time) :
				       DP_RX_THREAD_MIN_FLUSH_TIMEOUT;
	}
}

/**
 * dp_rx_tm_flush_by_vdev_id() - flush rx packets by vdev_id in all
 * rx thread queues
 * @rx_tm_hdl: dp_rx_tm_handle containing the overall thread
 * infrastructure
 * @vdev_id: vdev id for which packets are to be flushed
 *
 * Return: QDF_STATUS_SUCCESS
 */
QDF_STATUS dp_rx_tm_flush_by_vdev_id(struct dp_rx_tm_handle *rx_tm_hdl,
				     uint8_t vdev_id)
{
	uint64_t entry_time, exit_time;

	entry_time = qdf_get_log_timestamp();
	dp_rx_tm_flush_nbuf_list(rx_tm_hdl, vdev_id);
	dp_rx_tm_wait_vdev_del_event(rx_tm_hdl);
	exit_time = qdf_get_log_timestamp();
	dp_info("Vdev: %u total flush time: %llu us",
		vdev_id,
		qdf_log_timestamp_to_usecs(exit_time - entry_time));

	return QDF_STATUS_SUCCESS;
}

/**
 * dp_rx_tm_resume() - resume DP RX threads
 * @rx_tm_hdl: dp_rx_tm_handle containing the overall thread
 * infrastructure
 *
 * Return: QDF_STATUS_SUCCESS on resume success. QDF error otherwise.
 */
QDF_STATUS dp_rx_tm_resume(struct dp_rx_tm_handle *rx_tm_hdl)
{
	int i;

	if (rx_tm_hdl->state != DP_RX_THREADS_SUSPENDED &&
	    rx_tm_hdl->state != DP_RX_THREADS_SUSPENDING) {
		dp_info("resume callback received w/o suspend! Ignoring.");
		return QDF_STATUS_E_INVAL;
	}

	for (i = 0; i < rx_tm_hdl->num_dp_rx_threads; i++) {
		if (!rx_tm_hdl->rx_thread[i])
			continue;
		dp_debug("calling thread %d to resume", i);

		/* postively reset event_flag for DP_RX_THREADS_SUSPENDING
		 * state
		 */
		qdf_clear_bit(RX_SUSPEND_EVENT,
			      &rx_tm_hdl->rx_thread[i]->event_flag);
		qdf_event_set(&rx_tm_hdl->rx_thread[i]->resume_event);
	}

	rx_tm_hdl->state = DP_RX_THREADS_RUNNING;

	return QDF_STATUS_SUCCESS;
}

/**
 * dp_rx_refill_thread_resume() - Resume DP RX refill threads
 * @refill_thread: refill_thread containing the overall thread infrastructure
 *
 * Return: QDF_STATUS_SUCCESS on resume success. QDF error otherwise.
 */
QDF_STATUS dp_rx_refill_thread_resume(struct dp_rx_refill_thread *refill_thread)
{
	dp_debug("calling refill thread to resume");

	if (refill_thread->state != DP_RX_REFILL_THREAD_SUSPENDED &&
	    refill_thread->state != DP_RX_REFILL_THREAD_SUSPENDING) {
		dp_info("resume callback received in %d state ! Ignoring.",
			refill_thread->state);
		return QDF_STATUS_E_INVAL;
	}

	/* postively reset event_flag for DP_RX_REFILL_THREAD_SUSPENDING
	 * state
	 */
	qdf_clear_bit(RX_REFILL_SUSPEND_EVENT,
		      &refill_thread->event_flag);
	qdf_event_set(&refill_thread->resume_event);

	refill_thread->state = DP_RX_REFILL_THREAD_RUNNING;

	return QDF_STATUS_SUCCESS;
}

/**
 * dp_rx_tm_shutdown() - shutdown all DP RX threads
 * @rx_tm_hdl: dp_rx_tm_handle containing the overall thread infrastructure
 *
 * Return: QDF_STATUS_SUCCESS
 */
static QDF_STATUS dp_rx_tm_shutdown(struct dp_rx_tm_handle *rx_tm_hdl)
{
	int i;

	for (i = 0; i < rx_tm_hdl->num_dp_rx_threads; i++) {
		if (!rx_tm_hdl->rx_thread[i] ||
		    rx_tm_hdl->state == DP_RX_THREADS_INVALID)
			continue;
		qdf_set_bit(RX_SHUTDOWN_EVENT,
			    &rx_tm_hdl->rx_thread[i]->event_flag);
		qdf_set_bit(RX_POST_EVENT,
			    &rx_tm_hdl->rx_thread[i]->event_flag);
		qdf_wake_up_interruptible(&rx_tm_hdl->rx_thread[i]->wait_q);
	}

	for (i = 0; i < rx_tm_hdl->num_dp_rx_threads; i++) {
		if (!rx_tm_hdl->rx_thread[i] ||
		    rx_tm_hdl->state == DP_RX_THREADS_INVALID)
			continue;
		dp_debug("waiting for shutdown of thread %d", i);
		qdf_wait_single_event(&rx_tm_hdl->rx_thread[i]->shutdown_event,
				      0);
	}
	rx_tm_hdl->state = DP_RX_THREADS_INVALID;
	return QDF_STATUS_SUCCESS;
}

/**
 * dp_rx_tm_deinit() - de-initialize RX thread infrastructure
 * @rx_tm_hdl: dp_rx_tm_handle containing the overall thread
 * infrastructure
 *
 * Return: QDF_STATUS_SUCCESS
 */
QDF_STATUS dp_rx_tm_deinit(struct dp_rx_tm_handle *rx_tm_hdl)
{
	int i = 0;

	if (!rx_tm_hdl->rx_thread) {
		dp_err("rx_tm_hdl->rx_thread not initialized!");
		return QDF_STATUS_SUCCESS;
	}

	dp_rx_tm_shutdown(rx_tm_hdl);

	for (i = 0; i < rx_tm_hdl->num_dp_rx_threads; i++) {
		if (!rx_tm_hdl->rx_thread[i])
			continue;
		dp_rx_tm_thread_deinit(rx_tm_hdl->rx_thread[i]);
		qdf_mem_free(rx_tm_hdl->rx_thread[i]);
	}

	/* free the array of RX thread pointers*/
	qdf_mem_free(rx_tm_hdl->rx_thread);
	rx_tm_hdl->rx_thread = NULL;

	return QDF_STATUS_SUCCESS;
}

/**
 * dp_rx_tm_select_thread() - select a DP RX thread for a nbuf
 * @rx_tm_hdl: dp_rx_tm_handle containing the overall thread
 * infrastructure
 * @reo_ring_num: REO ring number corresponding to the thread
 *
 * The function relies on the presence of QDF_NBUF_CB_RX_CTX_ID passed to it
 * from the nbuf list. Depending on the RX_CTX (copy engine or reo
 * ring) on which the packet was received, the function selects
 * a corresponding rx_thread.
 *
 * Return: rx thread ID selected for the nbuf
 */
static uint8_t dp_rx_tm_select_thread(struct dp_rx_tm_handle *rx_tm_hdl,
				      uint8_t reo_ring_num)
{
	uint8_t selected_rx_thread;

	selected_rx_thread = reo_ring_num % rx_tm_hdl->num_dp_rx_threads;
	dp_debug("ring_num %d, selected thread %u", reo_ring_num,
		 selected_rx_thread);

	return selected_rx_thread;
}

QDF_STATUS dp_rx_tm_enqueue_pkt(struct dp_rx_tm_handle *rx_tm_hdl,
				qdf_nbuf_t nbuf_list)
{
	uint8_t selected_thread_id;

	selected_thread_id =
		dp_rx_tm_select_thread(rx_tm_hdl,
				       QDF_NBUF_CB_RX_CTX_ID(nbuf_list));
	dp_rx_tm_thread_enqueue(rx_tm_hdl->rx_thread[selected_thread_id],
				nbuf_list);
	return QDF_STATUS_SUCCESS;
}

QDF_STATUS
dp_rx_tm_gro_flush_ind(struct dp_rx_tm_handle *rx_tm_hdl, int rx_ctx_id,
		       enum dp_rx_gro_flush_code flush_code)
{
	uint8_t selected_thread_id;

	selected_thread_id = dp_rx_tm_select_thread(rx_tm_hdl, rx_ctx_id);
	dp_rx_tm_thread_gro_flush_ind(rx_tm_hdl->rx_thread[selected_thread_id],
				      flush_code);

	return QDF_STATUS_SUCCESS;
}

qdf_napi_struct *dp_rx_tm_get_napi_context(struct dp_rx_tm_handle *rx_tm_hdl,
					   uint8_t rx_ctx_id)
{
	uint8_t selected_thread_id;

	selected_thread_id = dp_rx_tm_select_thread(rx_tm_hdl, rx_ctx_id);

	return &rx_tm_hdl->rx_thread[selected_thread_id]->napi;
}

QDF_STATUS dp_rx_tm_set_cpu_mask(struct dp_rx_tm_handle *rx_tm_hdl,
				 qdf_cpu_mask *new_mask)
{
	int i = 0;

	for (i = 0; i < rx_tm_hdl->num_dp_rx_threads; i++) {
		if (!rx_tm_hdl->rx_thread[i])
			continue;
		qdf_thread_set_cpus_allowed_mask(rx_tm_hdl->rx_thread[i]->task,
						 new_mask);
	}
	return QDF_STATUS_SUCCESS;
}

/**
 * dp_rx_refill_thread_schedule() - Schedule rx refill thread
 * @soc: ol_txrx_soc_handle object
 *
 */
#ifdef WLAN_FEATURE_RX_PREALLOC_BUFFER_POOL
static void dp_rx_refill_thread_schedule(ol_txrx_soc_handle soc)
{
	struct dp_rx_refill_thread *rx_thread;
	struct dp_txrx_handle *dp_ext_hdl;

	if (!soc)
		return;

	dp_ext_hdl = cdp_soc_get_dp_txrx_handle(soc);
	if (!dp_ext_hdl)
		return;

	rx_thread = &dp_ext_hdl->refill_thread;
	qdf_set_bit(RX_REFILL_POST_EVENT, &rx_thread->event_flag);
	qdf_wake_up_interruptible(&rx_thread->wait_q);
}
#else
static void dp_rx_refill_thread_schedule(ol_txrx_soc_handle soc)
{
}
#endif

/**
 * dp_get_rx_threads_num() - Get number of threads in use
 * @soc: ol_txrx_soc_handle object
 *
 * Return: number of threads
 */
static uint8_t dp_get_rx_threads_num(ol_txrx_soc_handle soc)
{
	return cdp_get_num_rx_contexts(soc);
}

QDF_STATUS dp_txrx_init(ol_txrx_soc_handle soc, uint8_t pdev_id,
			struct dp_txrx_config *config)
{
	struct dp_txrx_handle *dp_ext_hdl;
	QDF_STATUS qdf_status = QDF_STATUS_SUCCESS;
	uint8_t num_dp_rx_threads;
	struct dp_pdev *pdev;
	struct dp_soc *dp_soc;

	if (qdf_unlikely(!soc)) {
		dp_err("soc is NULL");
		return 0;
	}

	pdev = dp_get_pdev_from_soc_pdev_id_wifi3(cdp_soc_t_to_dp_soc(soc),
						  pdev_id);
	if (!pdev) {
		dp_err("pdev is NULL");
		return 0;
	}

	dp_ext_hdl = qdf_mem_malloc(sizeof(*dp_ext_hdl));
	if (!dp_ext_hdl) {
		QDF_ASSERT(0);
		return QDF_STATUS_E_NOMEM;
	}

	dp_info("dp_txrx_handle allocated");
	dp_ext_hdl->soc = soc;
	dp_ext_hdl->pdev = dp_pdev_to_cdp_pdev(pdev);
	cdp_soc_set_dp_txrx_handle(soc, dp_ext_hdl);
	qdf_mem_copy(&dp_ext_hdl->config, config, sizeof(*config));
	dp_ext_hdl->rx_tm_hdl.txrx_handle_cmn =
				dp_txrx_get_cmn_hdl_frm_ext_hdl(dp_ext_hdl);

	dp_soc = cdp_soc_t_to_dp_soc(soc);
	if (wlan_cfg_is_rx_refill_buffer_pool_enabled(dp_soc->wlan_cfg_ctx)) {
		dp_ext_hdl->refill_thread.soc = soc;
		dp_ext_hdl->refill_thread.enabled = true;
		qdf_status =
			dp_rx_refill_thread_init(&dp_ext_hdl->refill_thread);
		if (qdf_status != QDF_STATUS_SUCCESS) {
			dp_err("Failed to initialize RX refill thread status:%d",
			       qdf_status);
			qdf_mem_free(dp_ext_hdl);
			return qdf_status;
		}
		cdp_register_rx_refill_thread_sched_handler(soc,
							    dp_rx_refill_thread_schedule);
	}

	num_dp_rx_threads = dp_get_rx_threads_num(soc);
	dp_info("%d RX threads in use", num_dp_rx_threads);

	if (dp_ext_hdl->config.enable_rx_threads) {
		qdf_status = dp_rx_tm_init(&dp_ext_hdl->rx_tm_hdl,
					   num_dp_rx_threads);
	}

	if (QDF_IS_STATUS_ERROR(qdf_status))
		dp_txrx_deinit(soc);

	return qdf_status;
}

QDF_STATUS dp_txrx_deinit(ol_txrx_soc_handle soc)
{
	struct dp_txrx_handle *dp_ext_hdl;
	struct dp_soc *dp_soc;

	if (!soc)
		return QDF_STATUS_E_INVAL;

	dp_ext_hdl = cdp_soc_get_dp_txrx_handle(soc);
	if (!dp_ext_hdl)
		return QDF_STATUS_E_FAULT;

	dp_soc = cdp_soc_t_to_dp_soc(soc);
	if (wlan_cfg_is_rx_refill_buffer_pool_enabled(dp_soc->wlan_cfg_ctx)) {
		dp_rx_refill_thread_deinit(&dp_ext_hdl->refill_thread);
		dp_ext_hdl->refill_thread.soc = NULL;
		dp_ext_hdl->refill_thread.enabled = false;
	}

	if (dp_ext_hdl->config.enable_rx_threads)
		dp_rx_tm_deinit(&dp_ext_hdl->rx_tm_hdl);

	qdf_mem_free(dp_ext_hdl);
	dp_info("dp_txrx_handle_t de-allocated");

	cdp_soc_set_dp_txrx_handle(soc, NULL);

	return QDF_STATUS_SUCCESS;
}

/**
 * dp_rx_tm_get_pending() - get number of frame in thread
 * nbuf queue pending
 * @soc: ol_txrx_soc_handle object
 *
 * Return: number of frames
 */
#ifdef FEATURE_WLAN_DP_RX_THREADS
int dp_rx_tm_get_pending(ol_txrx_soc_handle soc)
{
	int i;
	int num_pending = 0;
	struct dp_rx_thread *rx_thread;
	struct dp_txrx_handle *dp_ext_hdl;
	struct dp_rx_tm_handle *rx_tm_hdl;

	if (!soc)
		return 0;

	dp_ext_hdl = cdp_soc_get_dp_txrx_handle(soc);
	if (!dp_ext_hdl)
		return 0;

	rx_tm_hdl = &dp_ext_hdl->rx_tm_hdl;

	for (i = 0; i < rx_tm_hdl->num_dp_rx_threads; i++) {
		rx_thread = rx_tm_hdl->rx_thread[i];
		if (!rx_thread)
			continue;
		num_pending += qdf_nbuf_queue_head_qlen(&rx_thread->nbuf_queue);
	}

	if (num_pending)
		dp_debug("pending frames in thread queue %d", num_pending);

	return num_pending;
}
#else
int dp_rx_tm_get_pending(ol_txrx_soc_handle soc)
{
	return 0;
}
#endif