myvtm/spbase/sp_svc.c

#include "precompile.h"

#include "sp_svc.h"
#include "sp_dbg_export.h"
#include "sp_def.h"

#include "array.h"
#include "memutil.h"
#include "list.h"
#include "spinlock.h"

#define DEFAULT_MT_THREAD	1
#define DEFAUTL_MAX_MT_THREAD		32
#define DEFAULT_THREAD_TTL	300000 // 5min
#define MAX_RUNTIMESERIAL_LEVEL	16

typedef struct pkt_handler_t {
	struct hlist_node hentry;
	int key;
	sp_svc_on_pkt on_pkt;
	void *user_data;
}pkt_handler_t;

typedef struct sys_handler_t {
	struct hlist_node hentry;
	int key;
	sp_svc_on_sys on_sys;
	void *user_data;
}sys_handler_t;

struct sp_svc_t
{
	sp_iom_t *iom;
	int mt;
	int id;
	threadpool_t *threadpool;
	sp_tmr_mgr_t *tmr_mgr;
	spinlock_t lock;
	struct hlist_head pkt_handler_list[SP_PKT_MAX_NUM];
	struct hlist_head sys_handler_list;
	int stop;
	sp_uid_t last_runserial;
	sp_rsn_context_t *rsn_stack[MAX_RUNTIMESERIAL_LEVEL];
	int rsn_stack_sp;
};

static __inline void svc_lock(sp_svc_t *svc)
{
	spinlock_enter(&svc->lock, -1);
}

static __inline void svc_unlock(sp_svc_t *svc)
{
	spinlock_leave(&svc->lock);
}

static pkt_handler_t* find_pkt_handler(sp_svc_t *svc, int key, int pkt_type)
{
	pkt_handler_t *tpos;
	struct hlist_node *pos;
	hlist_for_each_entry(tpos, pos, &svc->pkt_handler_list[pkt_type >> SP_TYPE_SHIFT], pkt_handler_t, hentry) {
		if (key == tpos->key)
			return tpos;
	}
	return NULL;
}

static sys_handler_t* find_sys_handler(sp_svc_t *svc, int key)
{
	sys_handler_t *tpos;
	struct hlist_node *pos;
	hlist_for_each_entry(tpos, pos, &svc->sys_handler_list, sys_handler_t, hentry) {
		if (tpos->key == key) {
			return tpos;
		}
	}
	return NULL;
}

static int on_rx_pkt(sp_iom_t *iom, int to_svc_id, int epid, int svc_id, int pkt_type, int pkt_id, iobuffer_t **p_pkt, void *user_data)
{
	sp_svc_t *svc = (sp_svc_t*)user_data;
	pkt_handler_t *pkt_handler;
	struct hlist_node *pos;
	int pkt_handler_slot = SP_GET_PKT_TYPE(pkt_type) >> SP_TYPE_SHIFT;

	if (svc->id != to_svc_id)
		return TRUE;

	svc_lock(svc);
	hlist_for_each_entry(pkt_handler, pos, &svc->pkt_handler_list[pkt_handler_slot], pkt_handler_t, hentry) {
		if (pkt_handler->on_pkt) {
			int count;
			int read_state = iobuffer_get_read_state(*p_pkt);
			int write_state = iobuffer_get_write_state(*p_pkt);
			count = pkt_handler->on_pkt(svc, epid, svc_id, pkt_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;
			}
		}
	}
	svc_unlock(svc);

	return FALSE;
}

static void on_rx_sys(sp_iom_t *iom, int epid, int state, void *user_data)
{
	sp_svc_t *svc = (sp_svc_t*)user_data;
	sys_handler_t *sys_handler;
	struct hlist_node *pos;

	svc_lock(svc);
	hlist_for_each_entry(sys_handler, pos, &svc->sys_handler_list, sys_handler_t, hentry) {
		if (sys_handler->on_sys) {
			sys_handler->on_sys(svc, epid, state, sys_handler->user_data);
		}
	}
	svc_unlock(svc);
}

int sp_svc_create(sp_iom_t *iom, int multithread, int svc_id, int devel_id, sp_svc_t **p_svc)
{
	sp_svc_t *svc;
	int rc;
	int i;

	assert(iom);

	svc = ZALLOC_T(sp_svc_t);

	svc->mt = !!multithread;
	svc->iom = iom;

	rc = threadpool_create(&svc->threadpool);
	if (rc != 0) {
		sp_dbg_debug("create thread pool failed!");
		goto on_error;
	}

	rc = sp_tmr_mgr_create(svc, &svc->tmr_mgr);
	if (rc != 0) {
		sp_dbg_debug("create timer mgr failed!");
		goto on_error;
	}

	spinlock_init(&svc->lock);

	for (i = 0; i < SP_PKT_MAX_NUM; ++i) {
		INIT_HLIST_HEAD(&svc->pkt_handler_list[i]);
	}
	INIT_HLIST_HEAD(&svc->sys_handler_list);

	svc->id = svc_id;
	svc->rsn_stack_sp = 0;
	
	svc->last_runserial = sp_uid_make((unsigned short)devel_id);

	*p_svc = svc;

	return 0;

on_error:

	if (svc->iom) {
		sp_iom_destroy(svc->iom);
		svc->iom = NULL;
	}
	if (svc->tmr_mgr) {
		sp_tmr_mgr_destroy(svc->tmr_mgr);
		svc->tmr_mgr = NULL;
	}
	if (svc->threadpool) {
		threadpool_destroy(svc->threadpool);
		svc->threadpool = NULL;
	}

	free(svc);

	return Error_Unexpect;
}

void sp_svc_destroy(sp_svc_t *svc)
{
	sp_iom_destroy(svc->iom);
	sp_tmr_mgr_destroy(svc->tmr_mgr);
	threadpool_destroy(svc->threadpool);
	free(svc);
}

int sp_svc_add_pkt_handler(sp_svc_t *svc, int key, int pkt_type, sp_svc_on_pkt on_pkt, void *user_data)
{
	int rc = 0;
	pkt_handler_t *pkt_handler;

	svc_lock(svc);
	pkt_handler = find_pkt_handler(svc, key, pkt_type);
	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;
		hlist_add_head(&pkt_handler->hentry, &svc->pkt_handler_list[pkt_type >> SP_TYPE_SHIFT]);
	} else {
		rc = Error_Duplication;
	}
	svc_unlock(svc);

	return rc;
}

int sp_svc_remove_pkt_handler(sp_svc_t *svc, int key, int pkt_type)
{
	int rc = 0;
	pkt_handler_t *pkt_handler;

	svc_lock(svc);
	pkt_handler = find_pkt_handler(svc, key, pkt_type);
	if (pkt_handler) {
		hlist_del(&pkt_handler->hentry);
		free(pkt_handler);
	} else {
		rc = Error_NotExist;
	}
	svc_unlock(svc);

	return rc;
}

int sp_svc_add_sys_handler(sp_svc_t *svc, int key, sp_svc_on_sys on_sys, void *user_data)
{
	int rc = 0;
	sys_handler_t *sys_handler;

	svc_lock(svc);
	sys_handler = find_sys_handler(svc, key);
	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;
		hlist_add_head(&sys_handler->hentry, &svc->sys_handler_list);
	} else {
		rc = Error_Duplication;
	}
	svc_unlock(svc);

	return rc;
}

int sp_svc_remove_sys_handler(sp_svc_t *svc, int key)
{
	int rc = 0;
	sys_handler_t *sys_handler;

	svc_lock(svc);
	sys_handler = find_sys_handler(svc, key);
	if (sys_handler) {
		hlist_del(&sys_handler->hentry);
		free(sys_handler);
	} else {
		rc = Error_NotExist;
	}
	svc_unlock(svc);

	return rc;
}

int sp_svc_start(sp_svc_t *svc)
{
	int rc;

	if (!svc->mt) {
		rc = threadpool_start(svc->threadpool, 1, 0, 0, 0);
	} else {
		rc = threadpool_start(svc->threadpool, 
			DEFAULT_MT_THREAD, DEFAUTL_MAX_MT_THREAD, DEFAULT_THREAD_TTL, 0);
	}

	if (rc == 0) {
		sp_iom_add_pkt_handler(svc->iom, (int)svc, &on_rx_pkt, svc);
		sp_iom_add_sys_handler(svc->iom, (int)svc, &on_rx_sys, svc);
	}

	return rc;
}

int sp_svc_stop(sp_svc_t *svc)
{
	sp_iom_remove_pkt_handler(svc->iom, (int)svc);
	sp_iom_remove_sys_handler(svc->iom, (int)svc);

	sp_tmr_mgr_cancel_all_tmr(svc->tmr_mgr);
	while (sp_tmr_mgr_tmr_cnt(svc->tmr_mgr)) {
		Sleep(1);
	}

	threadpool_stop(svc->threadpool);

	return 0;
}

sp_tmr_mgr_t* sp_svc_get_tmr_mgr(sp_svc_t *svc)
{
	return svc->tmr_mgr;
}

sp_iom_t *sp_svc_get_iom(sp_svc_t *svc)
{
	return svc->iom;
}

threadpool_t *sp_svc_get_threadpool(sp_svc_t *svc)
{
	return svc->threadpool;
}

int sp_svc_get_id(sp_svc_t *svc)
{
	return svc->id;
}

int sp_svc_send(sp_svc_t *svc, int epid, int svc_id, int pkt_type, int pkt_id, iobuffer_t **p_pkt)
{
	return sp_iom_send(svc->iom, svc->id, epid, svc_id, pkt_type, pkt_id, p_pkt);
}

int sp_svc_post(sp_svc_t *svc, int epid, int svc_id, int pkt_type, int pkt_id, iobuffer_t **p_pkt)
{
	return sp_iom_post(svc->iom, svc->id, epid, svc_id, pkt_type, pkt_id, p_pkt);
}

int sp_svc_get_state(sp_svc_t *svc, int epid, int *state)
{
	return sp_iom_get_state(svc->iom, epid, state);
}

sp_uid_t sp_svc_new_runserial(sp_svc_t *svc)
{
	return svc->last_runserial = sp_uid_update(svc->last_runserial);
}

void sp_svc_push_runserial_context(sp_svc_t *svc, sp_rsn_context_t *rsn)
{
	assert(svc->rsn_stack_sp < MAX_RUNTIMESERIAL_LEVEL);
	svc->rsn_stack[svc->rsn_stack_sp++] = rsn;
}

void sp_svc_pop_runserial_context(sp_svc_t *svc)
{
	assert(svc->rsn_stack_sp > 0);
	svc->rsn_stack[--svc->rsn_stack_sp] = NULL;
}

sp_rsn_context_t *sp_svc_get_runserial_context(sp_svc_t *svc)
{
	if (svc->rsn_stack_sp) {
		return svc->rsn_stack[svc->rsn_stack_sp-1];
	} else {
		return NULL;
	}
}