/*

 * Copyright (c) 2001-2020 Mellanox Technologies, Ltd. All rights reserved.

 *

 * This software is available to you under a choice of one of two

 * licenses.  You may choose to be licensed under the terms of the GNU

 * General Public License (GPL) Version 2, available from the file

 * COPYING in the main directory of this source tree, or the

 * BSD license below:

 *

 *     Redistribution and use in source and binary forms, with or

 *     without modification, are permitted provided that the following

 *     conditions are met:

 *

 *      - Redistributions of source code must retain the above

 *        copyright notice, this list of conditions and the following

 *        disclaimer.

 *

 *      - Redistributions in binary form must reproduce the above

 *        copyright notice, this list of conditions and the following

 *        disclaimer in the documentation and/or other materials

 *        provided with the distribution.

 *

 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,

 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF

 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND

 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS

 * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN

 * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN

 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE

 * SOFTWARE.

 */

#include <vma/sock/fd_collection.h>

#include <vma/iomux/epfd_info.h>

#define MODULE_NAME "epfd_info:"

#define SUPPORTED_EPOLL_EVENTS (EPOLLIN|EPOLLOUT|EPOLLERR|EPOLLHUP|EPOLLRDHUP|EPOLLONESHOT|EPOLLET)

#define NUM_LOG_INVALID_EVENTS 10

#define EPFD_MAX_OFFLOADED_STR 150

#define CQ_FD_MARK 0xabcd

int epfd_info::remove_fd_from_epoll_os(int fd)

{

	int ret = orig_os_api.epoll_ctl(m_epfd, EPOLL_CTL_DEL, fd, NULL);

	BULLSEYE_EXCLUDE_BLOCK_START

	if (ret < 0) {

		__log_dbg("failed to remove fd=%d from os epoll epfd=%d (errno=%d %m)", fd, m_epfd, errno);

	}

	BULLSEYE_EXCLUDE_BLOCK_END

	return ret;

}

epfd_info::epfd_info(int epfd, int size) :

	lock_mutex_recursive("epfd_info"), m_epfd(epfd), m_size(size), m_ring_map_lock("epfd_ring_map_lock"),

	m_lock_poll_os("epfd_lock_poll_os"), m_sysvar_thread_mode(safe_mce_sys().thread_mode),

	m_b_os_data_available(false)

{

	__log_funcall("");

	int max_sys_fd = get_sys_max_fd_num();

	if (m_size<=max_sys_fd)

	{

		m_size=max_sys_fd;

		__log_dbg("using open files max limit of %d file descriptors", m_size);

	}

	m_ready_fds.set_id("epfd_info (%p) : m_ready_fds", this);

	m_p_offloaded_fds = new int[m_size];

	m_n_offloaded_fds = 0;

	memset(&(m_local_stats.stats), 0, sizeof(m_local_stats.stats));

	/* This initialization is not needed (because it is also done in shmem) but for proper code

	 * we do it in any case

	 */

	m_local_stats.enabled = true;

	m_local_stats.epfd = m_epfd;

	m_stats = &m_local_stats;

	m_log_invalid_events = NUM_LOG_INVALID_EVENTS;

	vma_stats_instance_create_epoll_block(m_epfd, &(m_stats->stats));

	// Register this socket to read nonoffloaded data

	g_p_event_handler_manager->update_epfd(m_epfd, EPOLL_CTL_ADD, EPOLLIN | EPOLLPRI | EPOLLONESHOT);

	wakeup_set_epoll_fd(m_epfd);

}

epfd_info::~epfd_info()

{

	__log_funcall("");

	socket_fd_api* sock_fd;

	// Meny: going over all handled fds and removing epoll context.

	lock();

	while(!m_ready_fds.empty())

	{

		sock_fd = m_ready_fds.get_and_pop_front();

		sock_fd->m_epoll_event_flags = 0;

	}

	while(!m_fd_offloaded_list.empty())

	{

		sock_fd = m_fd_offloaded_list.get_and_pop_front();

		sock_fd->m_fd_rec.reset();

	}

	for (int i = 0; i < m_n_offloaded_fds; i++) {

		sock_fd = fd_collection_get_sockfd(m_p_offloaded_fds[i]);

		BULLSEYE_EXCLUDE_BLOCK_START

		if (sock_fd) {

			unlock();

			m_ring_map_lock.lock();

			sock_fd->remove_epoll_context(this);

			m_ring_map_lock.unlock();

			lock();

		} else {

			__log_err("Invalid temp_sock_fd_api==NULL. Deleted fds should have been removed from epfd.");

		}

		BULLSEYE_EXCLUDE_BLOCK_END

	}

	g_p_event_handler_manager->update_epfd(m_epfd, EPOLL_CTL_DEL, EPOLLIN | EPOLLPRI | EPOLLONESHOT);

	unlock();

	vma_stats_instance_remove_epoll_block(&m_stats->stats);

	delete [] m_p_offloaded_fds;

}

int epfd_info::ctl(int op, int fd, epoll_event *event)

{

	int ret;

	epoll_event event_dummy;

	if (event == NULL) {

		memset(&event_dummy, 0, sizeof(event_dummy));

		event = &event_dummy;

	}

	// YossiE TODO make "event table" - and add index in that table instead

	// of real event (in orig_os_api.epoll_ctl). must have this because fd's can

	// be added after the cq.

	lock();

	switch (op) {

	case EPOLL_CTL_ADD:

		ret = add_fd(fd, event);

		break;

	case EPOLL_CTL_DEL:

		ret = del_fd(fd);

		break;

	case EPOLL_CTL_MOD:

		ret = mod_fd(fd, event);

		break;

	default:

		errno = EINVAL;

		ret = -1;

		break;

	}

	unlock();

	return ret;

}

void epfd_info::get_offloaded_fds_arr_and_size(int **p_p_num_offloaded_fds,

					       int **p_p_offloadded_fds)

{

	*p_p_num_offloaded_fds = &m_n_offloaded_fds;

	*p_p_offloadded_fds = m_p_offloaded_fds;

}

bool epfd_info::is_cq_fd(uint64_t data)

{

	if ((data >> 32) != CQ_FD_MARK)

		return false;

	lock();

	//todo consider making m_ready_cq_fd_q a set instead of queue

	m_ready_cq_fd_q.push_back((int)(data & 0xffff));

	unlock();

	return true;

}

int epfd_info::add_fd(int fd, epoll_event *event)

{

	int ret;

	epoll_fd_rec fd_rec;

	epoll_event evt = {0, {0}};

	bool is_offloaded = false;

	__log_funcall("fd=%d", fd);

	socket_fd_api* temp_sock_fd_api = fd_collection_get_sockfd(fd);

	if (temp_sock_fd_api && temp_sock_fd_api->get_type()== FD_TYPE_SOCKET) {

		is_offloaded = true;

	}

	// Make sure that offloaded fd has a correct event mask

	if (is_offloaded) {

		if (m_log_invalid_events && (event->events & ~SUPPORTED_EPOLL_EVENTS)) {

			__log_dbg("invalid event mask 0x%x for offloaded fd=%d", event->events, fd);

			__log_dbg("(event->events & ~%s)=0x%x", TO_STR(SUPPORTED_EPOLL_EVENTS),

			          event->events & ~SUPPORTED_EPOLL_EVENTS);

			m_log_invalid_events--;

		}

	}

	if (temp_sock_fd_api && temp_sock_fd_api->skip_os_select()) {

		__log_dbg("fd=%d must be skipped from os epoll()", fd);

		// Checking for duplicate fds

		if (get_fd_rec(fd)) {

			errno = EEXIST;

			__log_dbg("epoll_ctl: fd=%d is already registered with this epoll instance %d (errno=%d %m)", fd, m_epfd, errno);

			return -1;

		}

	}

	else {

		// Add an event which indirectly point to our event

		evt.events = event->events;

		evt.data.u64 = 0; //zero all data

		evt.data.fd = fd;

		ret = orig_os_api.epoll_ctl(m_epfd, EPOLL_CTL_ADD, fd, &evt);

		BULLSEYE_EXCLUDE_BLOCK_START

		if (ret < 0) {

			__log_dbg("failed to add fd=%d to epoll epfd=%d (errno=%d %m)", fd, m_epfd, errno);

			return ret;

		}

		BULLSEYE_EXCLUDE_BLOCK_END

	}

	fd_rec.events = event->events;

	fd_rec.epdata = event->data;

	if (is_offloaded) {  // TODO: do we need to handle offloading only for one of read/write?

		if (m_n_offloaded_fds >= m_size) {

			__log_dbg("Reached max fds for epoll (%d)", m_size);

			errno = ENOMEM;

			return -1;

		}

		//NOTE: when supporting epoll on epfd, need to add epfd ring list

		//NOTE: when having rings in pipes, need to overload add_epoll_context

		unlock();

		m_ring_map_lock.lock();

		ret = temp_sock_fd_api->add_epoll_context(this);

		m_ring_map_lock.unlock();

		lock();

		if (ret < 0) {

			switch (errno) {

			case EEXIST:

				__log_dbg("epoll_ctl: fd=%d is already registered with this epoll instance %d (errno=%d %m)", fd, m_epfd, errno);

				break;

			case ENOMEM:

				__log_dbg("epoll_ctl: fd=%d is already registered with another epoll instance %d, cannot register to epoll %d (errno=%d %m)", fd, temp_sock_fd_api->get_epoll_context_fd(), m_epfd, errno);

				break;

			default:

				__log_dbg("epoll_ctl: failed to add fd=%d to epoll epfd=%d (errno=%d %m)", fd, m_epfd, errno);

				break;

			}

			return ret;

		}

		m_p_offloaded_fds[m_n_offloaded_fds] = fd;

		++m_n_offloaded_fds;

		m_fd_offloaded_list.push_back(temp_sock_fd_api);

		fd_rec.offloaded_index = m_n_offloaded_fds;

		temp_sock_fd_api->m_fd_rec = fd_rec;

		// if the socket is ready, add it to ready events

		uint32_t events = 0;

		if ((event->events & EPOLLIN) && temp_sock_fd_api->is_readable(NULL, NULL)) {

			events |=  EPOLLIN;

		}

		if ((event->events & EPOLLOUT) && temp_sock_fd_api->is_writeable()) {

			// MNY: udp_socket is always ready to write. Both VMA and the OS will notify it.

			// Can't remove notification in VMA in case user decides to skip the OS using VMA params.

			// Meaning: user will get 2 ready WRITE events on startup of socket

			events |= EPOLLOUT;

		}

		if (events != 0) {

			insert_epoll_event(temp_sock_fd_api, events);

		}

		else{

			do_wakeup();

		}

	} else {

		fd_rec.offloaded_index = -1;

		m_fd_non_offloaded_map[fd] = fd_rec;

	}

	__log_func("fd %d added in epfd %d with events=%#x and data=%#x", 

		   fd, m_epfd, event->events, event->data);

	return 0;

}

void epfd_info::increase_ring_ref_count(ring* ring)

{

	m_ring_map_lock.lock();

	ring_map_t::iterator iter = m_ring_map.find(ring);

	if (iter != m_ring_map.end()) {

		//increase ref count

		iter->second++;

	} else {

		m_ring_map[ring] = 1;

		// add cq channel fd to the epfd

		int num_ring_rx_fds = ring->get_num_resources();

		int *ring_rx_fds_array = ring->get_rx_channel_fds();

		for (int i = 0; i < num_ring_rx_fds; i++) {

			epoll_event evt = {0, {0}};

			evt.events = EPOLLIN | EPOLLPRI;

			int fd = ring_rx_fds_array[i];

			evt.data.u64 = (((uint64_t)CQ_FD_MARK << 32) | fd);

			int ret = orig_os_api.epoll_ctl(m_epfd, EPOLL_CTL_ADD, fd, &evt);

			BULLSEYE_EXCLUDE_BLOCK_START

			if (ret < 0) {

				__log_dbg("failed to add cq fd=%d to epoll epfd=%d (errno=%d %m)",

						fd, m_epfd, errno);

			} else {

				__log_dbg("add cq fd=%d to epfd=%d", fd, m_epfd);

			}

			BULLSEYE_EXCLUDE_BLOCK_END

		}

	}

	m_ring_map_lock.unlock();

}

void epfd_info::decrease_ring_ref_count(ring* ring)

{

	m_ring_map_lock.lock();

	ring_map_t::iterator iter = m_ring_map.find(ring);

	BULLSEYE_EXCLUDE_BLOCK_START

	if (iter == m_ring_map.end()) {

		__log_err("expected to find ring %p here!", ring);

		m_ring_map_lock.unlock();

		return;

	}

	BULLSEYE_EXCLUDE_BLOCK_END

	//decrease ref count

	iter->second--;

	if (iter->second == 0) {

		m_ring_map.erase(iter);

		// remove cq channel fd from the epfd

		int num_ring_rx_fds = ring->get_num_resources();

		int *ring_rx_fds_array = ring->get_rx_channel_fds();

		for (int i = 0; i < num_ring_rx_fds; i++) {

			// delete cq fd from epfd

			int ret = orig_os_api.epoll_ctl(m_epfd, EPOLL_CTL_DEL, ring_rx_fds_array[i], NULL);

			BULLSEYE_EXCLUDE_BLOCK_START

			if (ret < 0) {

				__log_dbg("failed to remove cq fd=%d from epfd=%d (errno=%d %m)",

						ring_rx_fds_array[i], m_epfd, errno);

			} else {

				__log_dbg("remove cq fd=%d from epfd=%d", ring_rx_fds_array[i], m_epfd);

			}

			BULLSEYE_EXCLUDE_BLOCK_END

		}

	}

	m_ring_map_lock.unlock();

}

/*

 * del_fd have two modes:

 * 1. not passthrough (default) - remove the fd from the epfd, both from OS epfd and VMA epfd

 * 2. passthrough - remove the fd as offloaded fd, and keep it only on OS epfd if it was there.

 *    this is a 1 way direction from both offloaded/not-offloaded to not-offloaded only.

 */

int epfd_info::del_fd(int fd, bool passthrough)

{

	__log_funcall("fd=%d", fd);

	epoll_fd_rec* fi;

	socket_fd_api* temp_sock_fd_api = fd_collection_get_sockfd(fd);

	if (temp_sock_fd_api && temp_sock_fd_api->skip_os_select()) {

		__log_dbg("fd=%d must be skipped from os epoll()", fd);

	}

	else if (!passthrough) {

		remove_fd_from_epoll_os(fd);

	}

	fi = get_fd_rec(fd);

	if (!fi) {

		errno = ENOENT;

		return -1;

	}

	if (temp_sock_fd_api && temp_sock_fd_api->get_epoll_context_fd() == m_epfd) {

		m_fd_offloaded_list.erase(temp_sock_fd_api);

		if (passthrough) {

			// In case the socket is not offloaded we must copy it to the non offloaded sockets map.

			// This can happen after bind(), listen() or accept() calls.

			m_fd_non_offloaded_map[fd] = *fi;

			m_fd_non_offloaded_map[fd].offloaded_index = -1;

		}

	} else {

		fd_info_map_t::iterator fd_iter = m_fd_non_offloaded_map.find(fd);

		if (fd_iter != m_fd_non_offloaded_map.end()) {

			m_fd_non_offloaded_map.erase(fd_iter);

		}

	}

	if (temp_sock_fd_api && temp_sock_fd_api->ep_ready_fd_node.is_list_member()) {

		temp_sock_fd_api->m_epoll_event_flags = 0;

		m_ready_fds.erase(temp_sock_fd_api);

	}

	// handle offloaded fds

	if (fi->offloaded_index > 0) {

		//check if the index of fd, which is being removed, is the last one.

		//if does, it is enough to decrease the val of m_n_offloaded_fds in order

		//to shrink the offloaded fds array.

		if (fi->offloaded_index < m_n_offloaded_fds) {

			// remove fd and replace by last fd

			m_p_offloaded_fds[fi->offloaded_index - 1] =

					m_p_offloaded_fds[m_n_offloaded_fds - 1];

			socket_fd_api* last_socket = fd_collection_get_sockfd(m_p_offloaded_fds[m_n_offloaded_fds - 1]);

			if (last_socket && last_socket->get_epoll_context_fd() == m_epfd) {

				last_socket->m_fd_rec.offloaded_index = fi->offloaded_index;

			} else {

				__log_warn("Failed to update the index of offloaded fd: %d last_socket %p\n",

						m_p_offloaded_fds[m_n_offloaded_fds - 1], last_socket);

			}

		}

		--m_n_offloaded_fds;

	}

	if (temp_sock_fd_api) {

		temp_sock_fd_api->m_fd_rec.reset();

		unlock();

		m_ring_map_lock.lock();

		temp_sock_fd_api->remove_epoll_context(this);

		m_ring_map_lock.unlock();

		lock();

	}

	__log_func("fd %d removed from epfd %d", fd, m_epfd);

	return 0;

}

int epfd_info::mod_fd(int fd, epoll_event *event)

{

	epoll_event evt;

	epoll_fd_rec* fd_rec;

	int ret;

	__log_funcall("fd=%d", fd);

	// find the fd in local table

	fd_rec = get_fd_rec(fd);

	if (!fd_rec) {

		errno = ENOENT;

		return -1;

	}

	socket_fd_api* temp_sock_fd_api = fd_collection_get_sockfd(fd);

	// check if fd is offloaded that new event mask is OK 

	if (temp_sock_fd_api && temp_sock_fd_api->m_fd_rec.offloaded_index > 0) {

		if (m_log_invalid_events && (event->events & ~SUPPORTED_EPOLL_EVENTS)) {

			__log_dbg("invalid event mask 0x%x for offloaded fd=%d", event->events, fd);

			__log_dbg("(event->events & ~%s)=0x%x", TO_STR(SUPPORTED_EPOLL_EVENTS),

					event->events & ~SUPPORTED_EPOLL_EVENTS);

			m_log_invalid_events--;

		}

	}

	if (temp_sock_fd_api && temp_sock_fd_api->skip_os_select()) {

		__log_dbg("fd=%d must be skipped from os epoll()", fd);

	}

	else {

		// modify fd

		evt.events = event->events;

		evt.data.u64 = 0; //zero all data

		evt.data.fd = fd;

		ret = orig_os_api.epoll_ctl(m_epfd, EPOLL_CTL_MOD, fd, &evt);

		BULLSEYE_EXCLUDE_BLOCK_START

		if (ret < 0) {

			__log_err("failed to modify fd=%d in epoll epfd=%d (errno=%d %m)", fd, m_epfd, errno);

			return ret;

		}

		BULLSEYE_EXCLUDE_BLOCK_END

	}

	// modify fd data in local table

	fd_rec->epdata = event->data;

	fd_rec->events = event->events;

	bool is_offloaded = temp_sock_fd_api && temp_sock_fd_api->get_type()== FD_TYPE_SOCKET;

	uint32_t events = 0;

	if (is_offloaded) {

		// if the socket is ready, add it to ready events

		if ((event->events & EPOLLIN) && temp_sock_fd_api->is_readable(NULL, NULL)) {

			events |=  EPOLLIN;

		}

		if ((event->events & EPOLLOUT) && temp_sock_fd_api->is_writeable()) {

			// MNY: udp_socket is always ready to write. Both VMA and the OS will notify it.

			// Can't remove notification in VMA in case user decides to skip the OS using VMA params.

			// Meaning: user will get 2 ready WRITE events on startup of socket

			events |= EPOLLOUT;

		}

		if (events != 0) {

			insert_epoll_event(temp_sock_fd_api, events);

		}

	}

	if (event->events == 0 || events == 0) {

		if (temp_sock_fd_api && temp_sock_fd_api->ep_ready_fd_node.is_list_member()) {

			temp_sock_fd_api->m_epoll_event_flags = 0;

			m_ready_fds.erase(temp_sock_fd_api);

		}

	}

	__log_func("fd %d modified in epfd %d with events=%#x and data=%#x", 

		   fd, m_epfd, event->events, event->data);

	return 0;

}

epoll_fd_rec* epfd_info::get_fd_rec(int fd)

{

	epoll_fd_rec* fd_rec = NULL;

	socket_fd_api* temp_sock_fd_api = fd_collection_get_sockfd(fd);

	lock();

	if (temp_sock_fd_api && temp_sock_fd_api->get_epoll_context_fd() == m_epfd) {

		fd_rec = &temp_sock_fd_api->m_fd_rec;

	} else {

		fd_info_map_t::iterator iter = m_fd_non_offloaded_map.find(fd);

		if (iter != m_fd_non_offloaded_map.end()) {

			fd_rec = &iter->second;

		}

	}

	unlock();

	return fd_rec;

}

void epfd_info::fd_closed(int fd, bool passthrough)

{

	lock();

	if (get_fd_rec(fd)) {

		del_fd(fd, passthrough);

	}

	unlock();

}

void epfd_info::insert_epoll_event_cb(socket_fd_api* sock_fd, uint32_t event_flags)

{

	lock();

	//EPOLLHUP | EPOLLERR are reported without user request

	if (event_flags & (sock_fd->m_fd_rec.events | EPOLLHUP | EPOLLERR)) {

		insert_epoll_event(sock_fd, event_flags);

	}

	unlock();

}

void epfd_info::insert_epoll_event(socket_fd_api *sock_fd, uint32_t event_flags)

{

	// assumed lock

	if (sock_fd->ep_ready_fd_node.is_list_member()) {

		sock_fd->m_epoll_event_flags |= event_flags;

	}

	else {

		sock_fd->m_epoll_event_flags = event_flags;

		m_ready_fds.push_back(sock_fd);

	}

	do_wakeup();

}

void epfd_info::remove_epoll_event(socket_fd_api *sock_fd, uint32_t event_flags)

{

	sock_fd->m_epoll_event_flags &= ~event_flags;

	if (sock_fd->m_epoll_event_flags == 0) {

		m_ready_fds.erase(sock_fd);

	}

}

epoll_stats_t *epfd_info::stats()

{

	return m_stats;

}

int epfd_info::ring_poll_and_process_element(uint64_t *p_poll_sn, void* pv_fd_ready_array/* = NULL*/)

{

	__log_func("");

	int ret_total = 0;

	if (m_ring_map.empty()) {

		return ret_total;

	}

	m_ring_map_lock.lock();

	for (ring_map_t::iterator iter = m_ring_map.begin(); iter != m_ring_map.end(); iter++) {

		int ret = iter->first->poll_and_process_element_rx(p_poll_sn, pv_fd_ready_array);

		BULLSEYE_EXCLUDE_BLOCK_START

		if (ret < 0 && errno != EAGAIN) {

			__log_err("Error in ring->poll_and_process_element() of %p (errno=%d %m)", iter->first, errno);

			m_ring_map_lock.unlock();

			return ret;

		}

		BULLSEYE_EXCLUDE_BLOCK_END

		if (ret > 0)

			__log_func("ring[%p] Returned with: %d (sn=%d)", iter->first, ret, *p_poll_sn);

		ret_total += ret;

	}

	m_ring_map_lock.unlock();

	if (m_sysvar_thread_mode == THREAD_MODE_PLENTY && ret_total == 0 && errno == EAGAIN) pthread_yield();

	if (ret_total) {

		__log_func("ret_total=%d", ret_total);

	} else {

		__log_funcall("ret_total=%d", ret_total);

	}

	return ret_total;

}

int epfd_info::ring_request_notification(uint64_t poll_sn)

{

	__log_func("");

	int ret_total = 0;

	if (m_ring_map.empty()) {

		return ret_total;

	}

	m_ring_map_lock.lock();

	for (ring_map_t::iterator iter = m_ring_map.begin(); iter != m_ring_map.end(); iter++) {

		int ret = iter->first->request_notification(CQT_RX, poll_sn);

		BULLSEYE_EXCLUDE_BLOCK_START

		if (ret < 0) {

			__log_err("Error ring[%p]->request_notification() (errno=%d %m)", iter->first, errno);

			m_ring_map_lock.unlock();

			return ret;

		}

		BULLSEYE_EXCLUDE_BLOCK_END

		__log_func("ring[%p] Returned with: %d (sn=%d)", iter->first, ret, poll_sn);

		ret_total += ret;

	}

	m_ring_map_lock.unlock();

	return ret_total;

}

int epfd_info::ring_wait_for_notification_and_process_element(uint64_t *p_poll_sn, void* pv_fd_ready_array /* = NULL*/)

{

	__log_func("");

	int ret_total = 0;

	while (!m_ready_cq_fd_q.empty()) {

		lock();

		if (m_ready_cq_fd_q.empty()) {

			unlock();

			break;

		}

		int fd = m_ready_cq_fd_q.back();

		m_ready_cq_fd_q.pop_back();

		unlock();

		cq_channel_info* p_cq_ch_info = g_p_fd_collection->get_cq_channel_fd(fd);

		if (p_cq_ch_info) {

			ring* p_ready_ring = p_cq_ch_info->get_ring();

			// Handle the CQ notification channel

			int ret = p_ready_ring->wait_for_notification_and_process_element(fd, p_poll_sn, pv_fd_ready_array);

			if (ret < 0) {

				if (errno == EAGAIN) {

					__log_dbg("Error in ring->wait_for_notification_and_process_element() of %p (errno=%d %m)", p_ready_ring, errno);

				}

				else {

					__log_err("Error in ring->wait_for_notification_and_process_element() of %p (errno=%d %m)", p_ready_ring, errno);

				}

				continue;

			}

			if (ret > 0) {

				__log_func("ring[%p] Returned with: %d (sn=%d)", p_ready_ring, ret, *p_poll_sn);

			}

			ret_total += ret;

		}

		else {

			__log_dbg("failed to find channel fd. removing cq fd=%d from epfd=%d", fd, m_epfd);

			BULLSEYE_EXCLUDE_BLOCK_START

			if ((orig_os_api.epoll_ctl(m_epfd, EPOLL_CTL_DEL,

					fd, NULL)) && (!(errno == ENOENT || errno == EBADF))) {

				__log_err("failed to del cq channel fd=%d from os epfd=%d (errno=%d %m)", fd, m_epfd, errno);

			}

			BULLSEYE_EXCLUDE_BLOCK_END

		}

	}

	if (ret_total) {

		__log_func("ret_total=%d", ret_total);

	} else {

		__log_funcall("ret_total=%d", ret_total);

	}

	return ret_total;

}

void epfd_info::clean_obj()

{

	if (g_p_fd_collection)

		g_p_fd_collection->remove_epfd_from_list(this);

	cleanable_obj::clean_obj();

}

void epfd_info::statistics_print(vlog_levels_t log_level /* = VLOG_DEBUG */)

{

	size_t num_rings, num_ready_fds, num_ready_cq_fd;

	int offloaded_str_place, i = 0;

	char offloaded_str[VLOGGER_STR_SIZE];

	// Prepare data

	num_rings = m_ring_map.size();

	iomux_func_stats_t temp_iomux_stats = m_stats->stats;

	num_ready_fds = m_ready_fds.size();

	num_ready_cq_fd = m_ready_cq_fd_q.size();

	// Epoll data

	vlog_printf(log_level, "Fd number : %d\n", m_epfd);

	vlog_printf(log_level, "Size : %d\n", m_size);

	vlog_printf(log_level, "Offloaded Fds : %d\n", m_n_offloaded_fds);

	while (i < m_n_offloaded_fds) {

		memset(offloaded_str, 0, sizeof(offloaded_str));

		for (offloaded_str_place = 0; offloaded_str_place < EPFD_MAX_OFFLOADED_STR && i < m_n_offloaded_fds; i++) {

			int n = snprintf(&offloaded_str[offloaded_str_place], sizeof(offloaded_str) - offloaded_str_place - 1, " %d", m_p_offloaded_fds[i]);

			if (!likely((0 < n) && (n < (int)(sizeof(offloaded_str) - offloaded_str_place - 1)))) {

				break;

			}

			offloaded_str_place += n;

		}

		offloaded_str[offloaded_str_place] = '\0';

		vlog_printf(log_level, "Offloaded Fds list: %s\n", offloaded_str);

	}

	vlog_printf(log_level, "Number of rings : %u\n", num_rings);

	vlog_printf(log_level, "Number of ready Fds : %u\n", num_ready_fds);

	vlog_printf(log_level, "Number of ready CQ Fds : %u\n", num_ready_cq_fd);

	if (temp_iomux_stats.n_iomux_os_rx_ready || temp_iomux_stats.n_iomux_rx_ready || temp_iomux_stats.n_iomux_timeouts || temp_iomux_stats.n_iomux_errors ||

			temp_iomux_stats.n_iomux_poll_miss || temp_iomux_stats.n_iomux_poll_hit) {

		vlog_printf(log_level, "Polling CPU : %d%%\n", temp_iomux_stats.n_iomux_polling_time);

		if (temp_iomux_stats.threadid_last != 0)

			vlog_printf(log_level, "Thread Id : %5u\n", temp_iomux_stats.threadid_last);

		if (temp_iomux_stats.n_iomux_os_rx_ready || temp_iomux_stats.n_iomux_rx_ready)

			vlog_printf(log_level, "Rx fds ready : %u / %u [os/offload]\n", temp_iomux_stats.n_iomux_os_rx_ready, temp_iomux_stats.n_iomux_rx_ready);

		if (temp_iomux_stats.n_iomux_poll_miss + temp_iomux_stats.n_iomux_poll_hit) {

			double iomux_poll_hit = (double)temp_iomux_stats.n_iomux_poll_hit;

			double iomux_poll_hit_percentage = (iomux_poll_hit / (iomux_poll_hit + (double)temp_iomux_stats.n_iomux_poll_miss)) * 100;

			vlog_printf(log_level, "Polls [miss/hit] : %u / %u (%2.2f%%)\n", temp_iomux_stats.n_iomux_poll_miss, temp_iomux_stats.n_iomux_poll_hit, iomux_poll_hit_percentage);

			if (temp_iomux_stats.n_iomux_timeouts)

				vlog_printf(log_level, "Timeouts : %u\n", temp_iomux_stats.n_iomux_timeouts);

			if (temp_iomux_stats.n_iomux_errors)

				vlog_printf(log_level, "Errors : %u\n", temp_iomux_stats.n_iomux_errors);

		}

	}

}

void epfd_info::set_os_data_available()

{

	auto_unlocker locker(m_lock_poll_os);

	m_b_os_data_available = true;

}

void epfd_info::register_to_internal_thread()

{

	auto_unlocker locker(m_lock_poll_os);

	m_b_os_data_available = false;

	// Reassign EPOLLIN event

	g_p_event_handler_manager->update_epfd(m_epfd, EPOLL_CTL_MOD, EPOLLIN | EPOLLPRI | EPOLLONESHOT);

}

bool epfd_info::get_and_unset_os_data_available()

{

	auto_unlocker locker(m_lock_poll_os);

	bool ret = m_b_os_data_available;

	m_b_os_data_available = false;

	return ret;

}