epoller.cpp

socket的事件分发模型

#include "epoller.h"
#include "skfwd.h"
#include <assert.h>
#include <string.h>

#if zPLAT_WIN
#include <io.h>
#else 
#include <unistd.h>
#endif

inline int fdevent2epoll(int events){
	int revents = 0;
	if (events & FDEVENT_IN)
		revents |= EPOLLIN;
	if (events & FDEVENT_OUT)
		revents |= EPOLLOUT;
	if (events & FDEVENT_ERR)
		revents |= EPOLLERR;
	if (events & FDEVENT_PRI)
		revents |= EPOLLPRI;
	if (events & FDEVENT_HUP)
		revents |= EPOLLHUP;
	return revents;
}

void epoller::set_events(int idx, int events)
{
	assert(idx >= 0 && idx < (int)m_set.capacity());

	epoll_event ev;
	ev.events = fdevent2epoll(events);
	ev.data.ptr = m_set.get(idx);
	epoll_ctl(m_epoll_fd, EPOLL_CTL_MOD, m_set.get(idx)->get_fd(), &ev);
}

HRET epoller::create(size_t bufSize)
{
	clear();
	try {
		m_pRevents = new epoll_event[bufSize];
	}catch(...){
		delete[] m_pRevents;
		return -1;
	}
	HRET hret = m_set.create(bufSize);
	if (hret < 0){
		delete[] m_pRevents;
		m_pRevents = 0;
		return -1;
	}
	m_epoll_fd = epoll_create(bufSize);
	if (m_epoll_fd >= 0)
		return 0;
	clear();
	return -1;
}
/**
 *	@brief	
 *	@remark
 *		不改变epoll文件描述符
 */
HRET epoller::expand(size_t bufSize)
{
	size_t oldBufSize = m_set.capacity();

	if (bufSize <= oldBufSize)
		return 0;
	epoll_event* pRevents = 0;
	try{
		pRevents = new epoll_event[bufSize];
	}catch(...){
		delete[] pRevents;
		return -1;
	}
	HRET hret = m_set.expand(bufSize);
	if (hret < 0){
		delete[] pRevents;
		return hret;
	}

	if (oldBufSize > 0){
		memcpy(pRevents, m_pRevents, sizeof(epoll_event)*oldBufSize);
	}
	delete[] m_pRevents;
	m_pRevents = pRevents;
	return 0;
}

HRET epoller::add(fdbase& fb, int events /* = FDEVENT_IN | FDEVENT_OUT */)
{
	assert(m_epoll_fd >= 0);
	
	HRET hret = m_set.add(fb);
	if (hret < 0)
		return hret;
	epoll_event ev;
	ev.events = fdevent2epoll(events);
	ev.data.ptr = &fb;
	epoll_ctl(m_epoll_fd, EPOLL_CTL_ADD, fb.get_fd(), &ev);
	return hret;
}

HRET epoller::remove(int idx)
{
	fdbase* pfb = m_set.get(idx);
	assert(pfb);
	if (!pfb)
		return -1;
	epoll_event ev;
	ev.events = 0;
	ev.data.ptr = pfb;
	epoll_ctl(m_epoll_fd, EPOLL_CTL_DEL, pfb->get_fd(), &ev);
	m_set.remove(idx);
	return 0;
}

void epoller::clear()
{
	delete[] m_pRevents;
	m_pRevents = 0;
	if (m_epoll_fd >= 0){
#if zPLAT_WIN
		::epoll_close(m_epoll_fd);
#else
		::close(m_epoll_fd);
#endif
		m_epoll_fd = -1;
	}
	m_set.clear();
}

void epoller::destroy()
{
	delete[] m_pRevents;
	m_pRevents = 0;
	if (m_epoll_fd >= 0){
#if zPLAT_WIN
		::epoll_close(m_epoll_fd);
#else
		::close(m_epoll_fd);
#endif
		m_epoll_fd = -1;
	}
	m_set.destroy();
}

HRET epoller::wait(const struct timeval* ptv)
{
	int timeout = -1;
	if (ptv)
		timeout = ptv->tv_sec*1000+(ptv->tv_usec+999)/1000;

	int ret = epoll_wait(m_epoll_fd, m_pRevents, m_set.capacity(), timeout);

	if (ret > 0){
		dispatch(ret);
	}
	return ret;
}

void epoller::dispatch(int count)
{
	for(int i = 0 ; i < count ; ++i)
	{
		int events = m_pRevents[i].events;
		assert(m_pRevents[i].data.ptr);
		fdbase* pfd = (fdbase*)m_pRevents[i].data.ptr;
		if ((events & EPOLLIN)){
			if (pfd->do_read() < 0){
				continue;
			}
		}
		if ((events & EPOLLOUT)){
			if (pfd->do_write() < 0){
				continue;
			}
		}
		if ((events & EPOLLERR)){
			if (pfd->do_error() < 0){
				continue;
			}
		}
		if ((events & EPOLLHUP)){
			if (pfd->do_hup() < 0){
				continue;
			}
		}
		if ((events & EPOLLPRI)){
			if (pfd->do_pri() < 0){
				continue;
			}
		}
	}
}

void epoller::refresh()
{
	size_t bufSize = m_set.capacity();
	for(size_t i = 0 ; i < bufSize ; ++i)
	{
		fdbase* pfb = m_set.get(i);
		if (pfb){
			int events = 0;
			pfb->get_events(events);
			set_events(i, events);
		}
	}
}