myvtm/spbase/sp_iom.c

#include "precompile.h"

#include "sp_iom.h"
#include "sp_def.h"
#include "sp_dbg_export.h"

#include "spinlock.h"
#include "timerqueue.h"
#include "array.h"
#include "memutil.h"
#include <time.h>

#define POLL_INTERVAL	10

#define IOM_T_EXIT			0
#define IOM_T_GET_STATE		1
#define IOM_T_SEND_INFO		2

static int translate_error(int bus_error)
{
	int error;

	switch (bus_error) {
	case BUS_E_OK:
		error = Error_Succeed;
		break;
	case BUS_E_FAIL:
		error = Error_Unexpect;
		break;
	case BUS_E_NETBROKEN:
		error = Error_NetBroken;
		break;
	case BUS_E_NOTFOUND:
		error = Error_NotExist;
		break;
	default:
		error = Error_Unexpect;
		break;
	}

	return error;
}

typedef struct pkt_handler_t {
	int key;
	sp_iom_on_pkt on_pkt;
	void *user_data;
}pkt_handler_t;

typedef struct sys_handler_t {
	int key;
	sp_iom_on_sys on_sys;
	void *user_data;
}sys_handler_t;


struct sp_iom_t
{
	bus_endpt_t *endpt;
	timer_queue_t *tm_queue;
	spinlock_t tm_lock;
	spinlock_t pkt_handler_lock;
	spinlock_t sys_handler_lock;
	array_header_t *arr_pkt_handler;
	array_header_t *arr_sys_handler;
	int stop;
	int poll_thread_id;
};

static __inline void iom_pkt_handler_lock(sp_iom_t *iom)
{
	spinlock_enter(&iom->pkt_handler_lock, -1);
}

static __inline void iom_pkt_handler_unlock(sp_iom_t *iom)
{
	spinlock_leave(&iom->pkt_handler_lock);
}

static __inline void iom_sys_handler_lock(sp_iom_t *iom)
{
	spinlock_enter(&iom->sys_handler_lock, -1);
}

static __inline void iom_sys_handler_unlock(sp_iom_t *iom)
{
	spinlock_leave(&iom->sys_handler_lock);
}

static __inline void iom_tm_lock(sp_iom_t *iom)
{
	spinlock_enter(&iom->tm_lock, -1);
}

static __inline void iom_tm_unlock(sp_iom_t *iom)
{
	spinlock_leave(&iom->tm_lock);
}

static pkt_handler_t* find_pkt_handler(sp_iom_t *iom, int key, int *idx)
{
	int i;
	for (i = 0; i < iom->arr_pkt_handler->nelts; ++i) {
		pkt_handler_t *t = ARRAY_IDX(iom->arr_pkt_handler, i, pkt_handler_t*);
		if (t->key == key) {
			if (idx)
				*idx = i;
			return t;
		}
	}
	return NULL;
}

static sys_handler_t* find_sys_handler(sp_iom_t *iom, int key, int *idx)
{
	int i;
	for (i = 0; i < iom->arr_sys_handler->nelts; ++i) {
		sys_handler_t *t = ARRAY_IDX(iom->arr_sys_handler, i, sys_handler_t*);
		if (t->key == key) {
			if (idx)
				*idx = i;
			return t;
		}
	}
	return NULL;
}

static void on_pkt(bus_endpt_t *endpt, int epid, int type, iobuffer_t **p_pkt, void *user_data)
{
	sp_iom_t *iom = (sp_iom_t *)user_data;
	int i;

	iom_pkt_handler_lock(iom);
	for (i = 0; i < iom->arr_pkt_handler->nelts; ++i) {
		pkt_handler_t *pkt_handler = ARRAY_IDX(iom->arr_pkt_handler, i, pkt_handler_t*);
		if (pkt_handler->on_pkt) {
			int read_state = iobuffer_get_read_state(*p_pkt);
			int write_state = iobuffer_get_write_state(*p_pkt);
			int svc_id;
			int from_svc_id;
			int pkt_id;
			int count;
			iobuffer_format_read(*p_pkt, "444", &from_svc_id, &svc_id, &pkt_id);
			count = pkt_handler->on_pkt(iom, svc_id, epid, from_svc_id, type, pkt_id, p_pkt, pkt_handler->user_data);
			if (count && p_pkt && *p_pkt) {
				iobuffer_restore_read_state(*p_pkt, read_state);
				iobuffer_restore_write_state(*p_pkt, write_state);
			} else {
				break;
			}
		}
	}
	iom_pkt_handler_unlock(iom);
}

static void on_msg(bus_endpt_t *endpt, int msg, int nparam, int params[], int *result, void *user_data)
{
	sp_iom_t *iom = (sp_iom_t *)user_data;

	if (msg == IOM_T_SEND_INFO) {
		int pkt_type = params[0];
		int this_svc_id = params[1];
		int epid = params[2];
		int svc_id = params[3];
		int pkt_id = params[4];
		iobuffer_t *pkt = (iobuffer_t*)params[5];
		int read_state, write_state;
		int ret;
		read_state = iobuffer_get_read_state(pkt);
		write_state = iobuffer_get_write_state(pkt);
		iobuffer_write_head(pkt, IOBUF_T_I4, &pkt_id, 0);
		iobuffer_write_head(pkt, IOBUF_T_I4, &svc_id, 0);
		iobuffer_write_head(pkt, IOBUF_T_I4, &this_svc_id, 0);
		if (result) {
			int state;
			ret = bus_endpt_get_state(endpt, epid, &state);
			if (ret == 0)
				ret = (state == BUS_STATE_ON) ? 0 : -1;
			if (ret == 0)
				ret = bus_endpt_send_info(endpt, epid, pkt_type, pkt);
			*result = ret;
			if (ret != 0) {
				iobuffer_restore_read_state(pkt, read_state);
				iobuffer_restore_write_state(pkt, write_state);
			} else {
				iobuffer_dec_ref(pkt);
			}
		} else {
			bus_endpt_send_info(endpt, epid, pkt_type, pkt);
			iobuffer_dec_ref(pkt);
		}
		//sp_dbg_debug("on_msg send_info end, %d, %d, %d, %d", this_svc_id, epid, svc_id, pkt_id);
	} else if (msg == IOM_T_GET_STATE) {
		int epid = params[0];
		int *state = (int*)params[1];
		int *rc = (int*)params[2];
		*rc = bus_endpt_get_state(endpt, epid, state);
	} else if (msg == IOM_T_EXIT) {
		sp_iom_stop(iom);
	} else {
		assert(0);
	}

	
}

static void on_sys(bus_endpt_t *endpt, int epid, int state, void *user_data)
{
	sp_iom_t *iom = (sp_iom_t *)user_data;
	int i;
	
	iom_sys_handler_lock(iom);
	for (i = 0; i < iom->arr_sys_handler->nelts; ++i) {
		sys_handler_t *sys_handler = ARRAY_IDX(iom->arr_sys_handler, i, sys_handler_t*);
		if (sys_handler->on_sys) {
			sys_handler->on_sys(iom, epid, state, sys_handler->user_data);
		}
	}
	iom_sys_handler_unlock(iom);
}

int sp_iom_create(const char *url, int epid, sp_iom_t **p_iom)
{
	sp_iom_t *iom;
	bus_endpt_callback callback;
	int rc;

	iom = ZALLOC_T(sp_iom_t);

	callback.on_evt = NULL;
	callback.on_sys = &on_sys;
	callback.on_pkt = &on_pkt;
	callback.on_inf = &on_pkt;
	callback.on_msg = &on_msg;
	callback.user_data = iom;

	rc = bus_endpt_create(url, epid, &callback, &iom->endpt);
	if (rc != 0)
		goto on_error;
	timer_heap_create(&iom->tm_queue);
	spinlock_init(&iom->pkt_handler_lock);
	spinlock_init(&iom->sys_handler_lock);
	spinlock_init(&iom->tm_lock);
	iom->arr_pkt_handler = array_make(3, sizeof(pkt_handler_t*));
	iom->arr_sys_handler = array_make(3, sizeof(sys_handler_t*));
	iom->poll_thread_id = 0;

	*p_iom = iom;

	return 0;

on_error:
	free(iom);
	return Error_Unexpect;
}

void sp_iom_destroy(sp_iom_t *iom)
{
	timer_queue_destroy(iom->tm_queue);
	bus_endpt_destroy(iom->endpt);
	array_free(iom->arr_pkt_handler);
	array_free(iom->arr_sys_handler);
	free(iom);
}

int sp_iom_add_pkt_handler(sp_iom_t *iom, int key, sp_iom_on_pkt on_pkt, void *user_data)
{
	int rc = 0;
	pkt_handler_t *pkt_handler;

	iom_pkt_handler_lock(iom);
	pkt_handler = find_pkt_handler(iom, key, NULL);
	if (!pkt_handler) {
		pkt_handler = MALLOC_T(pkt_handler_t);
		pkt_handler->key = key;
		pkt_handler->on_pkt = on_pkt;
		pkt_handler->user_data = user_data;
		ARRAY_PUSH(iom->arr_pkt_handler, pkt_handler_t*) = pkt_handler;
	} else {
		rc = Error_Duplication;
	}
	iom_pkt_handler_unlock(iom);

	return rc;
}

int sp_iom_remove_pkt_handler(sp_iom_t *iom, int key)
{
	int rc = 0;
	pkt_handler_t *pkt_handler;
	int i;

	iom_pkt_handler_lock(iom);
	pkt_handler = find_pkt_handler(iom, key, &i);
	if (pkt_handler) {
		if (i < iom->arr_pkt_handler->nelts-1)
			ARRAY_IDX(iom->arr_pkt_handler, i, pkt_handler_t*) = ARRAY_IDX(iom->arr_pkt_handler, iom->arr_pkt_handler->nelts-1, pkt_handler_t*);
		array_pop(iom->arr_pkt_handler);
	} else {
		rc = Error_NotExist;
	}
	iom_pkt_handler_unlock(iom);

	return rc;
}

int sp_iom_add_sys_handler(sp_iom_t *iom, int key, sp_iom_on_sys on_sys, void *user_data)
{
	int rc = 0;
	sys_handler_t *sys_handler;

	iom_sys_handler_lock(iom);
	sys_handler = find_sys_handler(iom, key, NULL);
	if (!sys_handler) {
		sys_handler = MALLOC_T(sys_handler_t);
		sys_handler->key = key;
		sys_handler->on_sys = on_sys;
		sys_handler->user_data = user_data;
		ARRAY_PUSH(iom->arr_sys_handler, sys_handler_t*) = sys_handler;
	} else {
		rc = Error_Duplication;
	}
	iom_sys_handler_unlock(iom);

	return rc;
}

int sp_iom_remove_sys_handler(sp_iom_t *iom, int key)
{
	int rc = 0;
	sys_handler_t *sys_handler;
	int i;

	iom_sys_handler_lock(iom);
	sys_handler = find_sys_handler(iom, key, &i);
	if (sys_handler) {
		ARRAY_DEL(iom->arr_sys_handler, i, sys_handler_t*);
	} else {
		rc = Error_NotExist;
	}
	iom_sys_handler_unlock(iom);

	return rc;
}

static int sp_iom_poll(sp_iom_t *iom, int *timeout)
{
	int rc;	
	if (iom->poll_thread_id == 0) {
		iom->poll_thread_id = (int)GetCurrentThreadId();
	}
	rc = bus_endpt_poll(iom->endpt, *timeout);

	for (;;) {
		int cnt;
		long timerBeginTime, timerPlayTime;
		iom_tm_lock(iom);
		timerBeginTime = clock();
		cnt = timer_queue_poll_one(iom->tm_queue, NULL, timeout);		// timeout返回下一个定时器超时间隔
		timerPlayTime = clock() - timerBeginTime;
		if (timerPlayTime > 500)	//timer play time over than 500ms, it may make the main thread run slow
			sp_dbg_info("cur Timer %d has run %d ms", cnt, timerPlayTime);
		iom_tm_unlock(iom);
		if (!cnt)	 {
			//sp_dbg_debug("no timer execute current times.");
			break;
		}
	}
	return rc;
}

int sp_iom_run(sp_iom_t *iom)
{
	int timeout = POLL_INTERVAL;
	while (
#ifdef _WIN32
		InterlockedExchangeAdd((LONG*)&iom->stop, 0) == 0
#else
		/*the adapte func 'InterlockedExchangeAdd' implemented under winpr went wrong, maybe 64bit has responsibility*/
		iom->stop == 0
#endif //_WIN32
		|| timer_queue_get_count(iom->tm_queue) > 0) 
{
		int rc = sp_iom_poll(iom, &timeout);
		if (rc >= 0) {
			if (timeout > POLL_INTERVAL || timeout < 0)
				timeout = POLL_INTERVAL;
		} else {
			sp_dbg_debug("iom poll failed!");
			ExitProcess(-1);
			return rc;
		}
	}
	sp_dbg_debug("iom run exit ok!");
	return 0;
}

int sp_iom_stop(sp_iom_t *iom)
{
	sp_dbg_debug("set iom stop flag!");
	InterlockedExchange((LONG*)&iom->stop, 1);
	return 0;
}

int sp_iom_send(sp_iom_t *iom, int this_svc_id, int epid, int svc_id, int pkt_type, int pkt_id, iobuffer_t **p_pkt)
{
	iobuffer_t *pkt = p_pkt ? *p_pkt : NULL;
	int rc;

	if (!pkt) {
		pkt = iobuffer_create(-1, -1);
	}

	{
		int params[] = {
			pkt_type, 
			this_svc_id, 
			epid, 
			svc_id, 
			pkt_id,
			(int)pkt,
		};
		rc = bus_endpt_send_msg(iom->endpt, IOM_T_SEND_INFO, array_size(params), params);
	}

	if (!p_pkt || !*p_pkt) {
		if (rc != 0) {
			iobuffer_dec_ref(pkt);
		}
	} else {
		if (rc == 0)
			*p_pkt = NULL;
	}

	if (rc != 0)
		rc = Error_IO;

	return rc;
}

int sp_iom_post(sp_iom_t *iom, int this_svc_id, int epid, int svc_id, int pkt_type, int pkt_id, iobuffer_t **p_pkt)
{
	iobuffer_t *pkt = p_pkt ? *p_pkt : NULL;
	int rc;

	if (!pkt) {
		pkt = iobuffer_create(-1, -1);
	}

	{
		int params[] = {
			pkt_type, 
			this_svc_id, 
			epid, 
			svc_id, 
			pkt_id,
			(int)pkt,
		};
		rc = bus_endpt_post_msg(iom->endpt, IOM_T_SEND_INFO, array_size(params), params);
	}

	if (!p_pkt || !*p_pkt) {
		if (rc != 0)
			iobuffer_dec_ref(pkt);
	} else {
		if (rc == 0)
			*p_pkt = NULL;
	}

	if (rc != 0)
		rc = Error_Unexpect;

	return rc;
}

int sp_iom_get_state(sp_iom_t *iom, int epid, int *state)
{
	int rc = 0;
	int rc1;
	int params[] = {epid, (int)state, (int)&rc};

	// use -1 for get state
	rc1 = bus_endpt_send_msg(iom->endpt, IOM_T_GET_STATE, array_size(params), params);

	if (rc != 0)
		rc = Error_Unexpect;

	return rc1 == 0 ? rc : Error_Unexpect;
}

int sp_iom_schedule_timer(sp_iom_t *iom, timer_entry *entry, unsigned int delay)
{
	int rc;
	iom_tm_lock(iom);
	rc = timer_queue_schedule(iom->tm_queue, entry, delay);
	iom_tm_unlock(iom);
	return rc ? Error_Unexpect : 0;
}

int sp_iom_cancel_timer(sp_iom_t *iom, timer_entry *entry, int cancel)
{
	int rc;
	iom_tm_lock(iom);
	rc = timer_queue_cancel(iom->tm_queue, entry, cancel);
	iom_tm_unlock(iom);
	return rc ? Error_Unexpect : 0;
}

int sp_iom_get_epid(sp_iom_t *iom)
{
	return bus_endpt_get_epid(iom->endpt);
}

const char *sp_iom_get_comm_url(sp_iom_t *iom)
{
	return bus_endpt_get_url(iom->endpt);
}

int sp_iom_post_quit(sp_iom_t *iom)
{
	int rc = bus_endpt_post_msg(iom->endpt, IOM_T_EXIT, 0, 0);
	if (rc != 0)
		rc = Error_Unexpect;
	return rc;
}

int sp_iom_get_poll_thread_id(sp_iom_t *iom)
{
	return iom->poll_thread_id;
}