timeoutmanager.cpp

.net 方面的开发说明资料。

// ========================================================
// Event time-out handler
//
// Design and Implementation by Floris van den Berg
// ========================================================

#pragma warning (disable : 4275)
#pragma warning (disable : 4786)

#include <boost/thread/xtime.hpp>

#include "OpenNet.h"
#include "EventDispatcher.h"
#include "ThreadBalancer.h"
#include "TimeoutManager.h"
#include "Utilities.h"

// --------------------------------------------------------

TimeOutFunctor::TimeOutFunctor(TimeOutManager *manager) :
m_manager(manager) {
}

bool
TimeOutFunctor::iterate() {
	bool finished = false;
	TimeOutList::iterator i = NULL;

	do {		
		DWORD start_time = 0;
		DWORD timeout_value = 0;
		bool post_timeout = false;
		EpTimeOut timeout;
		TRANSPORT_HANDLE transport = NULL;		

		{
			boost::mutex::scoped_lock scoped_lock(m_manager->m_cs);

			// first time initialisation

			if (i == NULL)
				i = m_manager->m_queue.begin();

			// copy data from item

			if (i != m_manager->m_queue.end()) {
				start_time = (*i).m_start_time;
				timeout.protocol = (*i).m_protocol;
				timeout.msg = (*i).m_msg;
				timeout_value= (*i).m_timeout;
				transport = (*i).m_plug;

				// correct timeout when gettickcount wraps to 0

				DWORD tick = EpGetTickCount();

				if (start_time <= tick)
					tick -= start_time;
				else
					tick = (0xFFFFFFFF - start_time) + tick;

				// if the timeout expired, remove the item from the list, then flag
				// that a timeout has to be posted.

				if (tick >= timeout_value) {
					post_timeout = true;
					i = m_manager->m_queue.erase(i);
				} else {
					++i;
				}				
			} else {
				finished = true;
			}
		}

		// check for timeout

		if (post_timeout) {
			EpDispatchSystemEvent(transport, SYSTEM_TIMEOUT, &timeout, sizeof(EpTimeOut));
		}
	} while (!finished);

	boost::mutex::scoped_lock scoped_lock(m_manager->m_cs);
	return m_manager->m_queue.empty();		
}

void
TimeOutFunctor::operator()() {
	while (m_manager->m_thread_running) {
		// poll for timeouts

		iterate();

		// wait a few milliseconds to give other threads time to do something

		boost::xtime xt;
		boost::xtime_get(&xt, boost::TIME_UTC);
		xt.nsec += 1000000 * 55;
		boost::thread::sleep(xt);
	}
}

// --------------------------------------------------------
// Singleton implementation
// --------------------------------------------------------

TimeOutManager *TimeOutManager::instance = NULL;

TimeOutManager *
TimeOutManager::getInstance() {
	if (!isInstantiated())
		TimeOutManager::instance = new TimeOutManager;

	return TimeOutManager::instance;
}

void
TimeOutManager::destroyInstance() {
	if (isInstantiated()) {
		delete TimeOutManager::instance;

		TimeOutManager::instance = NULL;
	}
}

bool
TimeOutManager::isInstantiated() {
	return (TimeOutManager::instance != NULL);
}

// --------------------------------------------------------
// Manager implementation
// --------------------------------------------------------

TimeOutManager::TimeOutManager() :
m_thread(NULL),
m_thread_running(true),
m_cs(),
m_queue() {
	TimeOutFunctor functor(this);
	m_thread = new boost::thread(functor);
}

TimeOutManager::~TimeOutManager() {
	// terminate the thread

	m_thread_running = false;

	// wait until it is terminated

	m_thread->join();
	
	// cleanup

	delete m_thread;

	// clear all remaining items in the queue

	m_queue.clear();
}

void
TimeOutManager::addItem(ITransport *plug, EpAction *action) {
	if (action->timeout > 0) {
		TimeOutEvent time_out_event;
		time_out_event.m_plug = plug;
		time_out_event.m_msg = action->msg;
		time_out_event.m_protocol = action->protocol;
		time_out_event.m_timeout = action->timeout;
		time_out_event.m_start_time = EpGetTickCount();

		boost::mutex::scoped_lock scoped_lock(m_cs);

		m_queue.push_back(time_out_event);
	}
}

bool
TimeOutManager::removeItem(ITransport *plug, GUID protocol, int msg) {
	boost::mutex::scoped_lock scoped_lock(m_cs);

	for (TimeOutList::iterator i = m_queue.begin(); i != m_queue.end(); ++i) {
		if (((*i).m_plug == plug) && (IsEqualGUID((*i).m_protocol, protocol)) && ((*i).m_msg == msg)) {
			m_queue.erase(i);
			return true;
		}
	}

	return false;	
}

bool
TimeOutManager::removeItems() {
	boost::mutex::scoped_lock scoped_lock(m_cs);

	if (m_queue.size() > 0) {
		TimeOutList::iterator i = m_queue.begin();

		while (i != m_queue.end())
			i = m_queue.erase(i);

		return true;
	}

	return false;
}

bool
TimeOutManager::removeItems(ITransport *plug) {
	boost::mutex::scoped_lock scoped_lock(m_cs);
	bool result = false;

	TimeOutList::iterator i = m_queue.begin();

	while (i != m_queue.end()) {
		if (((*i).m_plug == plug)) {
			result = true;
			i = m_queue.erase(i);
		} else {
			++i;
		}
	}

	return result;
}

bool
TimeOutManager::removeItems(ITransport *plug, GUID protocol) {
	boost::mutex::scoped_lock scoped_lock(m_cs);
	bool result = false;

	if (m_queue.size() > 0) {
		TimeOutList::iterator i = m_queue.begin();

		while (i != m_queue.end()) {
			if (((*i).m_plug == plug) && (IsEqualGUID((*i).m_protocol, protocol))) {
				result = true;
				i = m_queue.erase(i);
			} else {
				++i;
			}
		}
	}

	return result;
}