myvtm/libtoolkit/unix/evtpoll.c

#include <assert.h>
#include <errno.h>
#include "memutil.h"
#include "spinlock.h"
#include "list.h"
#include "array.h"
#include "refcnt.h"
#include "evtpoll.h"
#include "core.h"
#include "queue.h"
#include <winpr/wlog.h>
#define TAG TOOLKIT_TAG("evtpoll")

#ifndef _WIN32

#include <unistd.h>

struct event_epoll_data_s {
	int type;
	int fd;
	void* owner;
};

struct event_epoll_s {
	int fd;
	/*insterest*/
	spinlock_t interest_list_lock;
	struct list_head interest_list;

	/*new mechanism*/
	event_epoll_data_t** interests;
	unsigned int cnt_of_intersests;
	unsigned int nfds;
};

struct evtpoll_interest_entry_s {
	struct list_head entry;
	//int type;
	int events;
	int pending;
	//void* key;
	void* data;

	spinlock_t lock;
	struct evtpoll_interest_s* owner;
};

struct evtpoll_interest_s {
	struct list_head node; // for evtpoll_t::interest_list
	struct list_head entry_list; //no used

	int fd;
	int events;
	int type; //no used
	evtpoll_t* owner;

	array_header_t* entries;
	spinlock_t lock;
	volatile int entry_counts;
};
typedef struct evtpoll_interest_entry_s evtpoll_interest_entry_t;
typedef struct evtpoll_interest_s evtpoll_interest_t;

static int evtpoll__interest_empty(evtpoll_t* ep)
{
	int ret;
	assert(ep);
	spinlock_enter(&ep->interest_list_lock, -1);
	ret = list_empty(&ep->interest_list);
	spinlock_leave(&ep->interest_list_lock);
	return ret;
}

event_epoll_data_t* evtpoll_data_new()
{
	event_epoll_data_t* data = ZALLOC_T(event_epoll_data_t);
	return data;
}

void evtpoll_data_destroy(event_epoll_data_t* epoll_data)
{
	if (epoll_data) {
		free(epoll_data);
	}
}

int evtpoll_data_set(event_epoll_data_t* epoll_data, int type, int fd, void* data)
{
	assert(epoll_data);
	epoll_data->fd = fd;
	epoll_data->type = type;
	epoll_data->owner = data;

	return 0;
}

int evtpoll_data_get_fd(const event_epoll_data_t* const epoll_data)
{
	assert(epoll_data);
	return epoll_data->fd;
}

int evtpoll_data_get_type(const event_epoll_data_t* const epoll_data)
{
	assert(epoll_data);
	return epoll_data->type;
}

void evtpoll_data_set_data(event_epoll_data_t* epoll_data, void* data)
{
	assert(epoll_data);
	epoll_data->owner = data;
}

void* evtpoll_data_get_data(const event_epoll_data_t* const epoll_data)
{
	assert(epoll_data);
	return epoll_data->owner;
}


static unsigned int next_power_of_two(unsigned int val) {
	val -= 1;
	val |= val >> 1;
	val |= val >> 2;
	val |= val >> 4;
	val |= val >> 8;
	val |= val >> 16;
	val += 1;
	return val;
}

static void resize_interests_capacity(evtpoll_t*ep, int len)
{
	event_epoll_data_t** interests;
	void* fake_interest_list;
	void* fake_interest_count;
	unsigned int cnt_of_intersests;
	unsigned int i;

	if (len <= ep->cnt_of_intersests)
		return;

	/* Preserve fake interest list and count at the end of the interests */
	if (ep->interests != NULL) {
		fake_interest_list = ep->interests[ep->cnt_of_intersests];
		fake_interest_count = ep->interests[ep->cnt_of_intersests + 1];
	}
	else {
		fake_interest_list = NULL;
		fake_interest_count = NULL;
	}

	cnt_of_intersests = next_power_of_two(len + 2) - 2;
	interests = REALLOC(ep->interests, (cnt_of_intersests + 2) * sizeof(ep->interests[0]));
	if (interests == NULL) {
		WLog_ERR(TAG, "no enough memory!!");
		abort();
	}
	for (i = ep->cnt_of_intersests; i < cnt_of_intersests; i++)
		interests[i] = NULL;
	interests[cnt_of_intersests] = fake_interest_list;
	interests[cnt_of_intersests + 1] = fake_interest_count;

	ep->interests = interests;
	ep->cnt_of_intersests = cnt_of_intersests;
}

void evtpoll__add(evtpoll_t* ep, event_epoll_data_t* evtpoll_data)
{
	resize_interests_capacity(ep, evtpoll_data->fd + 1);
	if (ep->interests[evtpoll_data->fd] == NULL) {
		ep->interests[evtpoll_data->fd] = evtpoll_data;
	}
}


evtpoll_t* evtpoll_create()
{
	int fd;
	evtpoll_t* ep = ZALLOC_T(evtpoll_t);
	if (!ep)
		return NULL;
	/*
	since Linux 2.6.8 the argument is ignored but must be g0 for compatibility,
	bcz kernel would allocate space dynamically.
	*/
	fd = epoll_create1(EPOLL_CLOEXEC);
	if (fd == -1 && (errno == ENOSYS || errno == EINVAL)) {
		fd = epoll_create(MAX_EPOLL_EVENT);
		if (fd != -1) {
			make_fd_cloexec(fd, 1);
		}
	}
	if (fd == -1) {
		WLog_ERR(TAG, "epoll fd create failed: %d", (errno));
		goto on_error;
	}
	ep->fd = fd;

	spinlock_init(&ep->interest_list_lock);
	INIT_LIST_HEAD(&ep->interest_list);

	ep->cnt_of_intersests = 0;
	ep->interests = NULL;
	ep->nfds = 0;

	return ep;
on_error:
	free(ep);
	return NULL;
}

int evtpoll_get_epoll_fd(const evtpoll_t* const evt_poll)
{
	return evt_poll->fd;
}

int evtpoll_get_raw_fd(evtpoll_t* ep)
{
	return ep->fd;
}

void evtpoll__interest_entry_set(
	evtpoll_interest_entry_t* entry,
	int events,
	int type,
	int pending,
	void* key,
	void* data)
{
	assert(entry);
	assert(entry->owner);
	//entry->type = type;
	entry->events = events;
	entry->data = data;
	entry->pending = pending;
	//entry->key = key;
}

void evtpoll__interest_entry_reset(evtpoll_interest_entry_t* entry) {
	evtpoll__interest_entry_set(entry, 0, 0, 0, NULL, NULL);
}

static evtpoll_interest_entry_t* evtpoll__interest_entry_create(evtpoll_interest_t* interest)
{
	evtpoll_interest_entry_t* entry = MALLOC_T(evtpoll_interest_entry_t);
	assert(entry);
	entry->owner = interest;
	evtpoll__interest_entry_reset(entry);
	spinlock_init(&entry->lock);
	return entry;
}


static int evtpoll__interest_entry_is_ready(const evtpoll_interest_entry_t* const entry)
{
	if (entry->pending != 0) {
		return 0;
	}
	if (!entry->data) {
		return 0;
	}
	if (entry->events == 0) {
		return 0;
	}
	return 1;
}

void evtpoll__interest_entry_destroy(evtpoll_interest_entry_t* entry)
{
	free(entry);
}

void evtpoll__interest_lock(evtpoll_interest_t* interest)
{
	spinlock_enter(&interest->lock, 0);
}

void evtpoll__interest_unlock(evtpoll_interest_t* interest)
{
	spinlock_leave(&interest->lock);
}

void evtpoll___add_interest_list(evtpoll_interest_t* inst)
{
	evtpoll_t* ep = inst->owner;
	assert(ep);
	spinlock_enter(&ep->interest_list_lock, -1);
	list_add(&inst->node, &ep->interest_list);
	spinlock_leave(&ep->interest_list_lock);
}

void evtpoll___del_interest_list(evtpoll_interest_t* inst)
{
	evtpoll_t* ep = inst->owner;
	assert(ep);
	spinlock_enter(&ep->interest_list_lock, -1);
	list_del(&inst->node);
	inst->node.next = inst->node.prev = NULL;
	spinlock_leave(&ep->interest_list_lock);
}

/*remove from inst list*/
void evtpoll__interest_free(evtpoll_interest_t* inst)
{
	evtpoll___del_interest_list(inst);
}


static int evtpoll__ctl(evtpoll_t* ep, int event_mask, int ctrl_mod, int fd, evtpoll_interest_t* inst)
{
	int ret = -1;
	struct epoll_event ee;
	assert(ep);
	ee.events = event_mask;
	if (inst) {
		ee.data.ptr = inst;
	}
	else {//data is union type.
		ee.data.fd = fd;
	}
	ret = epoll_ctl(evtpoll_get_epoll_fd(ep), ctrl_mod, fd, &ee);
	if (ret == -1) {
		WLog_ERR(TAG, "epoll ctl failed: %d", (errno));
	}
	else {
		WLog_INFO(TAG, "epoll_ctl(%d): OUT:%d, IN:%d, OTHERS:0x%08X, ADD:%d, MOD:%d, inst:0x%X ret:%d",
			fd,
			(event_mask & EPOLLOUT) > 0 ? 1 : 0, (event_mask & EPOLLIN) > 0 ? 1 : 0,
			(event_mask & ~(EPOLLOUT | EPOLLIN)),
			ctrl_mod == EPOLL_CTL_ADD ? 1 : 0, ctrl_mod == EPOLL_CTL_MOD ? 1 : 0, inst, ret);
	}
	return ret;
}

/*no use for now.*/
//static int evtpoll__ctl_remove(evtpoll_t* ep, int event_mask, int fd, void* data)
//{
//	int ret = -1;
//	struct epoll_event ee;
//	ee.events = event_mask;
//	if (data)
//		ee.data.ptr = data;
//	else
//		ee.data.fd = fd;
//	ret = epoll_ctl(ep->fd, EPOLL_CTL_DEL, fd, &ee);
//	if (ret == 0 || (ret == -1 && errno == ENOENT)) {
//		return 0;
//	}
//	WLog_ERR(TAG, "epoll remove ctrl failed: %d", (errno));
//	return -1;
//}

/*!
 * @brief
		 one 'evtpoll_interest_t' maps to one fd.
		 each interest instance has {DEFAULT_INTEREST_OP_COUNT} evtpoll_interest_entry_t type object.
 * @param[in]
 * @return : SUCC: new evtpoll_interest_t type pointer object; NULL: failed
 */
static evtpoll_interest_t* evtpoll__interest_create(int fd)
{
	int i;
	evtpoll_interest_t* inst = ZALLOC_T(evtpoll_interest_t);
	if (!inst) {
		return NULL;
	}
	inst->fd = fd;
	INIT_LIST_HEAD(&inst->entry_list);
	inst->entry_counts = 0;
	inst->owner = NULL;
	inst->type = 0;
	inst->events = 0;

	inst->entries = array_make(DEFAULT_INTEREST_OP_COUNT, sizeof(evtpoll_interest_entry_t*));
	for (i = 0; i < DEFAULT_INTEREST_OP_COUNT; ++i) {
		ARRAY_IDX(inst->entries, i, evtpoll_interest_entry_t*) = evtpoll__interest_entry_create(inst);
	}
	return inst;
}

/*free inst and its entry*/
static void evtpoll__interest_destroy(evtpoll_interest_t* inst)
{
	int i;
	assert(inst);
	assert(list_empty(&inst->entry_list));
	for (i = 0; i < inst->entries->nelts; ++i)
		evtpoll__interest_entry_destroy(ARRAY_IDX(inst->entries, i, evtpoll_interest_entry_t*));
	array_free(inst->entries);
	free(inst);
}


static evtpoll_interest_t* evtpoll__find_interest(const evtpoll_t* const cst_ep, int fd)
{
	struct list_head* pos;
	struct list_head* tmp;
	if (list_empty(&cst_ep->interest_list))
		return NULL;
	list_for_each_safe(pos, tmp, &cst_ep->interest_list) {
		evtpoll_interest_t* node = list_entry(pos, evtpoll_interest_t, node);
		if (node->fd == fd) {
			return node;
		}
	}
	return NULL;
}

static void evtpoll__interest_clear(evtpoll_t* ep)
{
	struct list_head* pos;
	struct list_head* tmp;
	if (list_empty(&ep->interest_list))
		return;
	spinlock_enter(&ep->interest_list_lock, -1);
	list_for_each_safe(pos, tmp, &ep->interest_list) {
		evtpoll_interest_t* node = list_entry(pos, evtpoll_interest_t, node);
		list_del(pos);
		if (node->events != 0) {
			evtpoll_unsubscribe(node->owner, EV_READ_WRITE_WITH_LT_PURE, node->fd, 0, NULL, NULL);
		}
		evtpoll__interest_destroy(node);
	}
	spinlock_leave(&ep->interest_list_lock);
	return;
}

static int evtpoll__register_interest(evtpoll_t* ep, evtpoll_interest_t* inst)
{
	int ret = -1;
	assert(ep);
	assert(inst);
	evtpoll_interest_t* exist;
#if 1
	//×¢²áºó²»Á¢¼´½øÐмàÌý
	ret = 0;
#else
	ret = evtpoll__ctl(ep, EV_READ_WRITE_WITH_LT_FULL, EPOLL_CTL_ADD, inst->fd, NULL);
	if (ret != 0)
		return ret;
#endif
	evtpoll___add_interest_list(inst);
#if 1 //DEBUG
	exist = evtpoll__find_interest(ep, inst->fd);
	assert(exist);
	assert(exist == inst);
#endif
	return ret;
}

static void evtpoll__unregister_interest(evtpoll_interest_t* inst)
{
	evtpoll__interest_free(inst);
	evtpoll__interest_destroy(inst);
}

/*remove inst from list and destroy it*/
//static int evtpoll__unregister_interest(evtpoll_t* ep, int fd)
//{
//	int ret = 0;
//	evtpoll_interest_t* exist;
//	exist = evtpoll__find_interest(ep, fd);
//	if (exist) {
//		evtpoll__unregister_interest(exist);
//	}
//	return ret;
//}

int evtpoll_attach(evtpoll_t* ep, int interest_fd)
{
	int ret = 0;
	assert(interest_fd > 0);
	evtpoll_interest_t* exist;
	WLog_DBG(TAG, "==> attach interest fd(%d) ", interest_fd);
	exist = evtpoll__find_interest(ep, interest_fd);
	if (exist) {
		WLog_WARN(TAG, "detect interest fd %d already exist.", interest_fd);
		return -1;
	}
	exist = evtpoll__interest_create(interest_fd);
	if (!exist) {
		WLog_ERR(TAG, "create epueue insterest fd %d failed!", interest_fd);
		return -1;
	}
	exist->owner = ep;
	ret = evtpoll__register_interest(ep, exist);
	if (ret != 0) {
		evtpoll__interest_destroy(exist);
		return -1;
	}
	WLog_DBG(TAG, "<== normal attach interest fd(%d) succ.", interest_fd);
	return 0;
}

/*unsubscribe R|W event and unregist it */
static void evtpoll__detach(evtpoll_interest_t* inst)
{
	if (inst->events != 0) {
		evtpoll_unsubscribe(inst->owner, EV_READ_WRITE_WITH_LT_PURE, inst->fd, 0, NULL, NULL);
	}
	evtpoll__unregister_interest(inst);
}

void evtpoll_detach(evtpoll_t* ep, int interest_fd)
{
	evtpoll_interest_t* exist;
	exist = evtpoll__find_interest(ep, interest_fd);
	if (exist) {
		evtpoll__detach(exist);
	}
}

static int evtpoll__subscribe_precheck(evtpoll_interest_entry_t* entry, int exist, void* data)
{
	if (exist) {
		if (entry->pending) {
			WLog_ERR(TAG, "the entry is still pending...");
			return -1;
		}
		else if ((uintptr_t)entry->data == (uintptr_t)data) {
			WLog_WARN(TAG, "entry already exists and seems same, return previously.");
			return 0;
		}
		WLog_DBG(TAG, "data:0x%08X vs data:0x%08X", entry->data, data);
	}
	//if (data != NULL) {
	//	const int io_type = ioqueue_overlapped_get_type(data);
	//	WLog_INFO(TAG, "ioqueue ov type for subscribing: 0x%X", io_type);
	//}
	//else if (!entry->data) {
	//	WLog_ERR(TAG, "no data data for event!");
	//	return -1;
	//}
	return 1;
}

static int evtpoll__subscribe_read(evtpoll_t* ep, evtpoll_interest_t* inst, void* data)
{
	int ret;
	int has = 0;
	int op = EPOLL_CTL_ADD;
	int events = EPOLLIN;
	evtpoll_interest_entry_t* entry;
	if (inst->events & EV_READ) {
		has = 1;
	}

	entry = ARRAY_IDX(inst->entries, EV_INTEREST_ENTRY_IN_IDX, evtpoll_interest_entry_t*);
	assert(entry);

	ret = evtpoll__subscribe_precheck(entry, has, data);
	if (ret < 1) {
		return ret;
	}

	ret = 0;

	if (has) {
		evtpoll__interest_entry_set(entry, EV_READ, 0, 0, NULL, data);
		WLog_INFO(TAG, "fd(%d): read entry existed, only update the ov and returned.", inst->fd);
		return 1;
	}

	if (inst->events & EV_WRITE) {
		op = EPOLL_CTL_MOD;
		events |= EPOLLOUT;
	}
	ret = evtpoll__ctl(ep, events, op, inst->fd, inst);
	if (ret) {
		WLog_ERR(TAG, "fd(%d): set read register failed !", inst->fd);
	}
	else {
		inst->events |= EV_READ;
		evtpoll__interest_entry_set(entry, EV_READ, 0, 0, NULL, data);
	}

	return ret;
}

static int evtpoll__unsubscribe_read(evtpoll_t* ep, evtpoll_interest_t* inst)
{
	int ret;
	int events = EPOLLIN;
	int op = EPOLL_CTL_DEL;
	evtpoll_interest_entry_t* entry = NULL;
	if (!(inst->events & EV_READ)) {
		WLog_WARN(TAG, "read event is not existed");
		return 0;
	}
	if (inst->events & EV_WRITE) {
		op = EPOLL_CTL_MOD;
		events = EPOLLOUT;
	}

	ret = evtpoll__ctl(ep, events, op, inst->fd, inst);
	if (ret) {
		WLog_ERR(TAG, "un read register failed !");
	}
	else {
		inst->events &= ~(EV_READ);
		entry = ARRAY_IDX(inst->entries, EV_INTEREST_ENTRY_IN_IDX, evtpoll_interest_entry_t*);
		evtpoll__interest_entry_reset(entry);
	}
	return ret;
}

static int evtpoll__subscribe_write(evtpoll_t* ep, evtpoll_interest_t* inst, void* data)
{
	int ret;
	//int io_type = -1;
	int has = 0;
	int op = EPOLL_CTL_ADD;
	int events = EPOLLOUT;
	evtpoll_interest_entry_t* entry = NULL;
	if (inst->events & EV_WRITE) {
		has = 1;
	}

	entry = ARRAY_IDX(inst->entries, EV_INTEREST_ENTRY_OUT_IDX, evtpoll_interest_entry_t*);
	assert(entry);

	ret = evtpoll__subscribe_precheck(entry, has, data);
	if (ret < 1) {
		return ret;
	}

	ret = 0;
	//io_type = ioqueue_overlapped_get_type(data);

	if (has) {
		evtpoll__interest_entry_set(entry, EV_WRITE, 0, 0, NULL, data);
		WLog_INFO(TAG, "fd(%d): write entry existed, only update the ov and returned.", inst->fd);
		return 1;
	}

	if (inst->events & EV_READ) {
		op = EPOLL_CTL_MOD;
		events |= EPOLLIN;
	}
	ret = evtpoll__ctl(ep, events, op, inst->fd, inst);
	if (ret) {
		WLog_ERR(TAG, "fd(%d): set write register failed !", inst->fd);
	}
	else {
		inst->events |= EV_WRITE;
		evtpoll__interest_entry_set(entry, EV_WRITE, 0, 0, NULL, data);
	}

	return ret;
}

static int evtpoll__unsubscribe_write(evtpoll_t* ep, evtpoll_interest_t* inst)
{
	int ret;
	int events = EPOLLOUT;
	int op = EPOLL_CTL_DEL;

	if (!(inst->events & EV_WRITE)) {
		WLog_WARN(TAG, "write event is not existed");
		return 0;
	}
	if (inst->events & EV_READ) {
		op = EPOLL_CTL_MOD;
		events = EPOLLIN;
	}

	ret = evtpoll__ctl(ep, events, op, inst->fd, inst);
	if (ret) {
		WLog_ERR(TAG, "un read register failed !");
	}
	else {
		evtpoll_interest_entry_t* entry = ARRAY_IDX(inst->entries, EV_INTEREST_ENTRY_OUT_IDX, evtpoll_interest_entry_t*);
		evtpoll__interest_entry_reset(entry);
		inst->events &= ~(EV_WRITE);
	}
	return ret;
}

int evtpoll_subscribe(evtpoll_t* ep, int event_mask, int fd, void* rdata, void* wdata)
{
	int ret = 0;
	evtpoll_interest_t* inst;
	inst = evtpoll__find_interest(ep, fd);
	if (!inst) {
		WLog_ERR(TAG, "the fd(%d) have not been registered", fd);
		return -1;
	}
	if (event_mask & EPOLLET) {
		WLog_WARN(TAG, "EPOLLET event type is not supported now.");
		event_mask &= ~EPOLLET;
	}
	if (event_mask & ~(EV_READ_WRITE_WITH_LT_PURE)) {
		WLog_ERR(TAG, "event type exclude read or write is not supported now.");
		return  -1;
	}
	if (!ret && (event_mask & EV_READ)) {
		ret = evtpoll__subscribe_read(ep, inst, rdata);
		if (ret > 0) {
			WLog_WARN(TAG, "read event has been registered");
			ret = 0;
		}
	}

	if (!ret && (event_mask & EV_WRITE)) {
		ret = evtpoll__subscribe_write(ep, inst, wdata);
		if (ret > 0) {
			WLog_WARN(TAG, "write event has been registered");
			ret = 0;
		}
	}

	return ret;
}

int evtpoll_unsubscribe(evtpoll_t* ep, int event_mask, int fd, int only_reset, void** rdata, void** wdata)
{
	int ret;
	evtpoll_interest_t* exist;
	exist = evtpoll__find_interest(ep, fd);
	if (!exist) {
		WLog_ERR(TAG, "the fd %d have not been registered", fd);
		return -1;
	}
	if (event_mask & ~(EV_READ_WRITE_WITH_LT_PURE)) {
		WLog_WARN(TAG, "event type exclude read or write is not supported now.");
		return  -1;
	}

	ret = 0;
	
	if (!ret && (event_mask & EV_READ)) {
		evtpoll_interest_entry_t* entry =
			ARRAY_IDX(exist->entries, EV_INTEREST_ENTRY_IN_IDX, evtpoll_interest_entry_t*);
		if (rdata)
			*rdata = entry->data;
		if (only_reset) {
			evtpoll__interest_entry_reset(entry);
		}
		else {
			ret = evtpoll__unsubscribe_read(ep, exist);
		}
	}

	if (!ret && (event_mask & EV_WRITE)) {
		evtpoll_interest_entry_t* entry =
			ARRAY_IDX(exist->entries, EV_INTEREST_ENTRY_OUT_IDX, evtpoll_interest_entry_t*);
		if (wdata)
			*wdata = entry->data;
		if (only_reset) {
			evtpoll__interest_entry_reset(entry);
		}
		else {
			ret = evtpoll__unsubscribe_write(ep, exist);
		}
	}
	return ret;
}

static int evtpoll__deal_inner(evtpoll_interest_t* inst, int idx, void** data, int reset)
{
	evtpoll_interest_entry_t* entry = ARRAY_IDX(inst->entries, idx, evtpoll_interest_entry_t*);
	assert(entry);
	if (evtpoll__interest_entry_is_ready(entry)) {
		WLog_DBG(TAG, "interest entry is ready.");
		*data = entry->data;
		if (reset) {
			evtpoll__interest_entry_reset(entry);
		}
		return 0;
	}
	WLog_WARN(TAG, "interest entry is not ready.");
	return -1;
}

int evtpoll_deal(evtpoll_t* ep, struct epoll_event* event, void** data, int cancel)
{
	int ret = 0;
	evtpoll_interest_t* inst;
	evtpoll_interest_t* exist;

	if (!(event->events & (EPOLLIN | EPOLLOUT))) {
		WLog_DBG(TAG, "no event mask.");
		return -1;
	}

	inst = (evtpoll_interest_t*)event->data.ptr;
	assert(inst);
#if 1 //DEBUG
	assert(inst->owner == ep);
	exist = evtpoll__find_interest(ep, inst->fd);
	assert(exist);
	assert(exist == inst);
#endif

	if (event->events & EPOLLIN) { // read
		ret = evtpoll__deal_inner(inst, EV_INTEREST_ENTRY_IN_IDX, data, cancel);
		if (!ret && cancel) {
			evtpoll__unsubscribe_read(ep, exist);
		}
		event->events &= ~(EPOLLIN);
		return ret;
	}
	if (event->events & EPOLLOUT) { // write
		ret = evtpoll__deal_inner(inst, EV_INTEREST_ENTRY_OUT_IDX, data, cancel);
		if (!ret && cancel) {
			evtpoll__unsubscribe_write(ep, exist);
		}
		event->events &= ~(EPOLLOUT);
		return ret;
	}
	return -1;
}

int evtpoll_wait(evtpoll_t* ep, struct epoll_event event_array[], int event_array_size, int timeout)
{
	int nfds;

	do 
	{

		nfds = epoll_wait(ep->fd, event_array, event_array_size, timeout);

	} while (nfds == -1 && errno == EINTR);

	if (nfds == 0) {
		assert(timeout != -1);
		return 0;
	}
	if (nfds == -1) {
		WLog_ERR(TAG, "epoll wait error: %d", errno);
		return -1;
	}

	return nfds;
}

int evtpoll_loop(evtpoll_t* ep, int timeout)
{
	int nfds;
	int ret = -1;
	return ret;
}

void evtpoll_destroy(evtpoll_t* evt_poll)
{
	evtpoll__interest_clear(evt_poll);
	assert(evtpoll__interest_empty(evt_poll));
	close(evt_poll->fd);
	free(evt_poll);
}

#endif //NOT _WIN32