selector.cpp

socket的事件分发模型

#include "selector.h"
#include <malloc.h>

selector::selector(size_t bufSize) : m_set(0), m_pfds(0), m_pevents(0)
{
	if (bufSize > 0)
		create(bufSize);
}

HRET selector::create(size_t bufSize)
{
	clear();
	HRET hret = m_set.create(bufSize);
	if (hret < 0)
		return hret;
	m_pfds = (int*)malloc(sizeof(int)*bufSize);
	m_pevents = (int*)malloc(sizeof(int)*bufSize);
	if (!m_pfds || !m_pevents){
		clear();
		return -1;
	}
	for(size_t i = 0 ; i < bufSize ; ++i)
	{
		m_pfds[i] = -1;
		m_pevents[i] = 0;
	}
	if ((hret = m_rset.create(bufSize)) < 0)
		goto failed;
	if ((hret = m_wset.create(bufSize)) < 0)
		goto failed;
	if ((hret = m_eset.create(bufSize)) < 0)
		goto failed;
	return 0;
failed:
	clear();
	return hret;
}
/**
 *	@brief	调整poller的大小（扩大）
 *	@remark	
 *		如果失败，就保持原有内容不变
 */
HRET selector::expand(size_t bufSize)
{
	size_t oldBufSize = m_set.capacity();
	if (bufSize <= oldBufSize)
		return 0;
	int* pfds = (int*)malloc(sizeof(int)*bufSize);
	int* pevents = (int*)malloc(sizeof(int)*bufSize);
	if (!pfds || !pevents){
		if (pfds)
			free(pfds);
		if (pevents)
			free(pevents);
		return -1;
	}
	HRET hret = m_set.expand(bufSize);
	if (hret < 0){
		free(pfds); 
		free(pevents);
		return hret;
	}
	memcpy(pfds, m_pfds, sizeof(int)*oldBufSize);
	memcpy(pevents, m_pevents, sizeof(int)*oldBufSize);
	for(size_t i = oldBufSize ; i < bufSize ; ++i)
	{
		pfds[i] = -1;
		pevents[i] = 0;
	}
	free(m_pfds); 
	free(m_pevents);
	m_pfds = pfds;
	m_pevents = pevents;
	return 0;
}

HRET selector::add(fdbase& fb, int events /* = POLLIN | POLLOUT */)
{
	HRET hret = m_set.add(fb);
	if (hret < 0)
		return hret;
	int fd = fb.get_fd();
	m_pevents[hret] = events;
	m_pfds[hret] = fd;
	return hret;
}

HRET selector::remove(int idx)
{
	assert(idx >= 0 && idx < (int)m_set.capacity());

	HRET hret = m_set.remove(idx);
	if (hret < 0)
		return hret;
	assert(m_pfds[idx] >= 0);
	m_pevents[idx] = 0;
	m_pfds[idx] = -1;
	return 0;
}

void selector::clear()
{
	m_set.clear();
	clr_set();
	if (m_pfds){
		free(m_pfds);
		m_pfds = 0;
	}
	if (m_pevents){
		free(m_pevents);
		m_pevents = 0;
	}
}

void selector::destroy()
{
	m_set.destroy();
	clr_set();

	if (m_pfds){
		free(m_pfds);
		m_pfds = 0;
	}
	if (m_pevents){
		free(m_pevents);
		m_pevents = 0;
	}
}

void selector::dispatch_select(size_t cnt, fd_set* pr, fd_set* pw, fd_set* pe)
{
	size_t bufSize = m_set.capacity();
	size_t i;
	size_t num;
	if (pr){
		num = 0;
		for(i = 0 ; i < bufSize ; ++i)
		{
			if (m_pfds[i] != -1 && FD_ISSET(m_pfds[i], pr)){
				m_set.get(i)->do_read();
				if (++num >= cnt)
					break;
			}
		}
	}
	if (pw){
		num = 0;
		for(i = 0 ; i < bufSize ; ++i)
		{
			if (m_pfds[i] != -1 && FD_ISSET(m_pfds[i], pw)){
				m_set.get(i)->do_write();
				if (++num >= cnt)
					break;
			}
		}
	}
	if (pe){
		num = 0;
		for(i = 0 ; i < bufSize ; ++i)
		{
			if (m_pfds[i] != -1 && FD_ISSET(m_pfds[i], pe)){
				m_set.get(i)->do_error();
				if (++num >= cnt)
					break;
			}
		}
	}
}

HRET selector::wait(const struct timeval* ptv)
{
	fd_set* pr = 0;
	fd_set* pw = 0;
	fd_set* pe = 0;
	
	fill_set();
	if (m_rset.count() > 0){
		pr = &m_rset.get_fdset();
	}
	if (m_wset.count() > 0){
		pw = &m_wset.get_fdset();
	}
	if (m_eset.count() > 0){
		pe = &m_eset.get_fdset();
	}
#if zPLAT_POSIX
	timeval tv;
	if (ptv)
		tv = *ptv;
	int ret = ::select(FD_SETSIZE, pr, pw, pe, ptv ? &tv : 0);
#else
	int ret = ::select(0, pr, pw, pe, ptv);
#endif

	if (ret > 0){
		dispatch_select(ret, pr, pw, pe);
	}
	return ret;
}

void selector::refresh()
{
	clr_set();

	size_t bufSize = m_set.capacity();
	for(size_t i = 0 ; i < bufSize ; ++i)
	{
		fdbase* p = m_set.get(i);
		m_pfds[i] = -1;
		if (p){
			int events;
			int fd = p->get_events(events);
			if (fd >= 0){
				m_pevents[i] = events;
				m_pfds[i] = fd;
			}
		}
	}
}

#if zDEBUG

void selector::print_freeze() const
{
	size_t cnt = m_set.capacity();
	for(size_t i = 0 ; i < cnt ; ++i)
	{
		if (m_pfds[i] >= 0){
			sockaddr_in addr;
			socklen_t len = sizeof(addr);
			int ret = ::getsockname(m_pfds[i], (sockaddr*)&addr, &len);
			if (ret < 0){
				DUMP3("error: selector find a freeze socket, index: %d, fd: %d, ptr: %p\n", i, m_pfds[i], m_set.get(i));
			}
		}		
	}
}

#endif