myvtm/libtoolkit/stm.c

#include "precompile.h"
#include "stm.h"

#include <WinSock2.h>
#include <Mswsock.h>
#define WIN32_LEAN_AND_MEAN
#include <Windows.h>

#include "memutil.h"
#include "strutil.h"
#include "timerqueue.h"

#ifndef SO_UPDATE_CONNECT_CONTEXT
#define SO_UPDATE_CONNECT_CONTEXT 0x7010
#endif

#ifndef WSAID_CONNECTEX
#define WSAID_CONNECTEX \
	{0x25a207b9,0xddf3,0x4660,{0x8e,0xe9,0x76,0xe5,0x8c,0x74,0x06,0x3e}}
#endif

#define MAX_TIMEOUT 500
#define MAX_UDP_CONCURENT	16

static struct list_head g_stm_list = LIST_HEAD_INIT(g_stm_list);

static stm_event_t *default_duplicate(const stm_event_t *src)
{
	stm_event_t *new_evt = (stm_event_t *)memdup(src, sizeof(stm_event_t)+src->ctx_size);
	new_evt->__ref_cnt = 1;
	new_evt->__entry.next = new_evt->__entry.prev = NULL;
	return new_evt;
}

static __inline const stm_machine_base_t *stm_find_no_lock(int machine_cat, int machine_cls)
{
	stm_machine_base_t *result = NULL;
	stm_machine_base_t *pos;
	list_for_each_entry(pos, &g_stm_list, stm_machine_base_t, entry) {
		if (pos->cat == machine_cat && pos->cls == machine_cls) {
			result = pos;
			break;
		}
	}
	return result;
}

TOOLKIT_API stm_event_t *__stm_event_create_base(int type, const char *type_desc, int src_stm_id, int dst_stm_id, int param1, int param2, int ctx_size, void **p_ctx)
{
	stm_event_t *e;
	assert(ctx_size >= 0);
	e = (stm_event_t *)malloc(sizeof(stm_event_t) + ctx_size);
	if (e) {
		e->__entry.next = e->__entry.prev = NULL;
		e->type = type;
		e->ctx_size = ctx_size;
		e->src_stm_id = src_stm_id;
		e->dst_stm_id = dst_stm_id;
		e->param1 = param1;
		e->param2 = param2;
		e->__ref_cnt = 1;
		e->on_destroy = NULL;
		e->duplicate = &default_duplicate;
		e->desc = type_desc;
		if (ctx_size && p_ctx) {
			*p_ctx = (void*)(e+1);
		}
		if (src_stm_id) {
			stm_machine_t *stm = stm_machine_id2ptr(src_stm_id);
			stm_machine_inc_ref(stm);
		}
		if (dst_stm_id) {
			stm_machine_t *stm = stm_machine_id2ptr(dst_stm_id);
			stm_machine_inc_ref(stm);
		}
		//printf("event %s create(%08x) %08x->%08x !\n", e->desc, e, src_stm_id, dst_stm_id);
	}
	return e;
}

TOOLKIT_API void __stm_event_destroy(stm_event_t *e)
{
	//printf("event %s %08x destroy!\n", e->desc, e);
	assert(e->__ref_cnt == 0);
	if (e->on_destroy)
		(*e->on_destroy)(e);
	if (e->src_stm_id) {
		stm_machine_t *stm = stm_machine_id2ptr(e->src_stm_id);
		stm_machine_dec_ref(stm);
	}
	if (e->dst_stm_id) {
		stm_machine_t *stm = stm_machine_id2ptr(e->dst_stm_id);
		stm_machine_dec_ref(stm);
	}
	free(e);
}

TOOLKIT_API long stm_event_inc_ref(stm_event_t *e) 
{ 
	if (e->src_stm_id) {
		stm_machine_t *src_stm = stm_machine_id2ptr(e->src_stm_id);
		if (src_stm->host_dispatcher != src_stm->report_dispatcher) {
			return InterlockedIncrement((LONG*)&e->__ref_cnt);
		} else {
			return ++e->__ref_cnt;
		}
	} else {
		return InterlockedIncrement((LONG*)&e->__ref_cnt);
	}
}

TOOLKIT_API long stm_event_dec_ref(stm_event_t *e)
{
	long l;
	if (e->src_stm_id) {
		stm_machine_t *src_stm = stm_machine_id2ptr(e->src_stm_id);
		if (src_stm->host_dispatcher != src_stm->report_dispatcher) {
			l = InterlockedDecrement((LONG*)&e->__ref_cnt);
		} else {
			l = --e->__ref_cnt;
		}
	} else {
		l = InterlockedDecrement((LONG*)&e->__ref_cnt);
	}
	if (l == 0) {
		__stm_event_destroy(e);
	}
	return l;
}

TOOLKIT_API void stm_tcp_conn_initial_arg_free(stm_event_t *e)
{
	stm_tcp_conn_initial_arg_t *arg = (stm_tcp_conn_initial_arg_t *)stm_event_get_context(e);
	if (arg->arr_server) {
		array_free(arg->arr_server);
		arg->arr_server = NULL;
	}
}

static __inline int validate_state(const stm_machine_base_t *machine_base, int state)
{
	return state >= 0 && state < machine_base->arr_state_size;
}

static int stm_machine_base_check(stm_machine_base_t *machine_base)
{

	if (machine_base->arr_state_size <= 0) {
		return FALSE;
	}

	return TRUE;
}

static int __stm_sort(stm_machine_base_t *machine_base)
{
	int rc = 0;
	int i;
	for (i = 0; i < machine_base->arr_state_size; ++i) {
		stm_state_t *pos = &machine_base->arr_state[i];
		if (pos->state != i) {
			int j;
			for (j = i+1; j < machine_base->arr_state_size; ++j) {
				stm_state_t *tpos = &machine_base->arr_state[j];
				if (tpos->state == i)
					break;
			}
			if (j < machine_base->arr_state_size) {
				stm_state_t tmp;
				memcpy(&tmp, pos, sizeof(stm_state_t));
				memcpy(pos, &machine_base->arr_state[j], sizeof(stm_state_t));
				memcpy(&machine_base->arr_state[j], &tmp, sizeof(stm_state_t));
			} else {
				assert(0);
				rc = -1;
				break;
			}
		}
	}
	for (i = 0; i < machine_base->arr_state_size; ++i) {
		stm_state_t *pos = &machine_base->arr_state[i];
		INIT_LIST_HEAD(&pos->rule_list);
	}
	for (i = 0; i < machine_base->arr_state_rule_size; ++i) {
		stm_state_rule_t *pos = &machine_base->arr_state_rule[i];
		stm_state_t *state = &machine_base->arr_state[pos->state];
		list_add_tail(&pos->entry, &state->rule_list);
	}
	return rc;
}

static int __stm_get_state_node_id(stm_machine_base_t *machine_base, stm_state_t *st, int depth)
{
	int ret = STM_INVALID_NODE;
	if (st->node_id != STM_INVALID_NODE) {
		ret = st->node_id;
	} else if (st->state == STM_STATE_ROOT) {
		ret = st->node_id;
	} else {
		if (depth < STM_MAX_STATE_DEPTH) {
			int parent_node_id = __stm_get_state_node_id(machine_base, &machine_base->arr_state[st->parent_state], depth+1);
			int level = STM_NODE_LEVEL(parent_node_id) + 1;
			STM_NODE_SET(st->node_id, level, st->state, parent_node_id);
			STM_NODE_SET_LEAF(st->node_id);
			ret = st->node_id;
			STM_NODE_CLEAR_LEAF(machine_base->arr_state[st->parent_state].node_id);
			if (ret != STM_INVALID_NODE) {
				int level = STM_NODE_LEVEL(ret);
				if (level > machine_base->depth)
					machine_base->depth = level;
			}
		}
	}
	return ret;
}

static int __stm_calc_nodes(stm_machine_base_t *machine_base)
{
	int i;
	for (i = 0; i < machine_base->arr_state_size; ++i) {
		stm_state_t *st = &machine_base->arr_state[i];
		if (st->state != STM_STATE_ROOT && st->node_id == STM_INVALID_NODE) {
			int depth = 0;
			st->node_id = __stm_get_state_node_id(machine_base, st, depth);
			if (st->node_id == STM_INVALID_NODE) 
				return -1;
		}
	}
	return 0;
}

static unsigned int __stm_on_start_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	return stm->base->on_init(stm, state, e);
}

static void __stm_on_exit_init(stm_machine_t *stm, const stm_state_t *state)
{
	stm->base->on_exit(stm);
}

TOOLKIT_API int stm_register(stm_machine_base_t *machine_base)
{
	int rc = 0;
	assert(stm_machine_base_check(machine_base));

	if (!stm_find_no_lock(machine_base->cat, machine_base->cls)) {
		if (!machine_base->depth) {
			rc = __stm_sort(machine_base);
			if (rc == 0) {
				rc = __stm_calc_nodes(machine_base);
			}
			if (rc == 0) {
				machine_base->arr_state[STM_STATE_INIT].on_state_event = &__stm_on_start_event;
				machine_base->arr_state[STM_STATE_EXIT].on_state_init = &__stm_on_exit_init;
			}
		}
		list_add_tail(&machine_base->entry, &g_stm_list);
	} else {
		rc = -1;
	}

	return rc;
}

TOOLKIT_API stm_machine_base_t *stm_unregister(int machine_cat, int machine_cls)
{
	stm_machine_base_t *resule = NULL;
	stm_machine_base_t *pos;

	list_for_each_entry(pos, &g_stm_list, stm_machine_base_t, entry) {
		if (pos->cat == machine_cat && pos->cls == machine_cls) {
			resule = pos;
			list_del(&pos->entry);
			break;
		}
	}

	return resule;
}

TOOLKIT_API const stm_machine_base_t *stm_find(int machine_cat, int machine_cls)
{
	const stm_machine_base_t *result = NULL;

	result = stm_find_no_lock(machine_cat, machine_cls);

	return result;
}

static void __stm_machine_clear_defer_list(stm_machine_t *stm)
{
	while (stm->dispatch_sp >= 0) {
		struct list_head *p = &stm->dispatch_stack[stm->dispatch_sp];
		while (!list_empty(p)) {
			stm_event_t *evt = list_first_entry(p, stm_event_t, __entry);
			list_del(&evt->__entry);
			stm_event_dec_ref(evt);
		}
		stm->dispatch_sp--;
	}
}

TOOLKIT_API stm_machine_t *stm_machine_create2(stm_dispatcher_t *host_dispatcher, stm_dispatcher_t *report_dispatcher, 
	int machine_cat, int machine_cls, 
	int report_stm_id, stm_event_t *initial_evt, 
	void (*user_on_destroy)(stm_machine_t *stm, void *arg),
	void (*on_state_trans)(stm_machine_t *stm, const stm_state_t *src_state, const stm_state_t *dst_state, void *arg), 
	void *arg)
{
	stm_machine_t *machine;
	const stm_machine_base_t *stm_base = stm_find(machine_cat, machine_cls);
	assert(stm_base->ctx_size >= 0);
	assert(initial_evt);
	if (!stm_base)
		return NULL;
	machine = (stm_machine_t *)malloc(sizeof(stm_machine_t)+stm_base->ctx_size);
	if (machine) {
		int i;
		machine->base = stm_base;
		machine->user_on_destroy = user_on_destroy;
		machine->tag = arg;
		machine->on_state_trans = on_state_trans;
		machine->current_state = STM_STATE_INIT;
		machine->last_error = 0;
		machine->host_dispatcher = host_dispatcher;
		machine->report_dispatcher = report_dispatcher;
		machine->entry.next = machine->entry.prev = NULL;
		machine->stm_id = stm_machine_ptr2id(machine);
		machine->report_stm_id = report_stm_id;
		machine->__ref_cnt = 1;
		machine->src_trans_evt_mask = NULL;
		for (i = 0; i < array_size(machine->dispatch_stack); ++i) {
			INIT_LIST_HEAD(&machine->dispatch_stack[i]);
		}
		machine->dispatch_sp = 0;
		if (host_dispatcher) {
			stm_dispatcher_lock(host_dispatcher);
			list_add_tail(&machine->entry, &host_dispatcher->machine_list);
			stm_dispatcher_unlock(host_dispatcher);
		}
		stm_event_change_dst_stm(initial_evt, machine->stm_id);
		if (host_dispatcher) {
			stm_dispatcher_post_event(host_dispatcher, initial_evt);
		} else {
			stm_machine_process(machine, initial_evt);
		}
	}
	return machine;
}

TOOLKIT_API void __stm_machine_destroy(stm_machine_t *stm)
{
	assert(stm->__ref_cnt == 0);
	if (stm->host_dispatcher) {
		stm_dispatcher_lock(stm->host_dispatcher);
		list_del(&stm->entry);
		stm_dispatcher_unlock(stm->host_dispatcher);
	} else {
		printf("caution: no delete from machine_list!!!\n");
	}
	__stm_machine_clear_defer_list(stm);
	if (stm->user_on_destroy) {
		(*stm->user_on_destroy)(stm, stm->tag);
	}
	if (stm->base->on_destroy) {
		(*stm->base->on_destroy)(stm);
	}
	if (stm->src_trans_evt_mask)
		free(stm->src_trans_evt_mask);
	free(stm);
}

static void __stm_trans_state(stm_machine_t *stm, int dst_state)
{
	const stm_state_t *src_st = &stm->base->arr_state[stm->current_state];
	const stm_state_t *dst_st = &stm->base->arr_state[dst_state];
	const stm_state_t *p = src_st;
	stm->current_state = dst_state;
	while (!stm_node_is_ancestor(p->node_id, dst_st->node_id)) {
		if (p->on_state_exit) {
			(*p->on_state_exit)(stm, p);
		}
		p = stm_get_parent_state(stm->base, p->state);
	}
	if (!stm->last_error && p != dst_st) {
		int v = STM_NODE_LEVEL(p->node_id);
		int n = dst_st->node_id;
		do {
			int vv = (1 << 5*(1+v)) - (1 << 5*v);
			p = &stm->base->arr_state[(vv & n) >> 5*v];
			v ++;
			if (p->on_state_init) {
				(*p->on_state_init)(stm, p);
			}
		} while (p != dst_st && !stm->last_error);
	}
	if (stm->last_error) {
		stm->current_state = STM_STATE_EXIT;
		p = &stm->base->arr_state[stm->current_state];
		p->on_state_init(stm, p);
	}

	if (stm->base->on_state_trans) {
		(*stm->base->on_state_trans)(stm, src_st, p);
	}
	if (stm->on_state_trans) {
		(*stm->on_state_trans)(stm, src_st, p, stm->tag);
	}
	if (stm->report_dispatcher && stm->report_stm_id) {
		if (p->state == STM_STATE_EXIT || src_st->state == STM_STATE_INIT || (stm->src_trans_evt_mask && (stm->src_trans_evt_mask[p->state] & (1 << src_st->state)))) {
			stm_event_t *e = stm_event_create_base(STM_SYS_EVT_MSG_TRANS, stm->stm_id, stm->report_stm_id, src_st->state, p->state, 0, NULL);
			stm_dispatcher_post_event(stm->report_dispatcher, e);
			stm_event_dec_ref(e);
		}
	}
}

static unsigned int __stm_decide_next_state(stm_machine_t *stm, const stm_state_t *state, int type, unsigned int result)
{
	const stm_state_rule_t *rule;
	list_for_each_entry(rule, &state->rule_list, const stm_state_rule_t, entry) {
		if (rule->evt_type == type) {
			if (result >= rule->result_start && result <= rule->result_end) {
				return rule->action;
			}
		}
	}
	return STM_STATE_INIT; // no jump
}

static int __stm_need_process_event(const stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	const stm_state_rule_t *rule;
	list_for_each_entry(rule, &state->rule_list, const stm_state_rule_t, entry) {
		if (rule->evt_type == e->type) {
			//printf("stm %08x state %s need process %s %08x\n", stm, state->name, e->desc, e);
			return TRUE;
		}
	}
	//printf("stm %08x state %s no need process %s %08x\n", stm, state->name, e->desc, e);
	return FALSE;
}

static int __stm_state_process_event(stm_machine_t *stm, const stm_state_t *stm_state, stm_event_t *e, unsigned int *action, int *handled_state)
{
	const stm_state_t *curr = stm_state;
	int rc = -1;
	while (curr && curr->state != STM_STATE_ROOT && !stm->last_error) {
		if (__stm_need_process_event(stm, curr, e)) {
			int e_type = e->type;
			unsigned int result = curr->on_state_event(stm, curr, e);
			//printf("stm %08x on_state_event state(%s), evt(%s) %08x, result=%d\n", stm, curr->name, e->desc, e, result);
			*action = __stm_decide_next_state(stm, curr, e_type, result);
			*handled_state = curr->state;
			rc = 0;
			break;
		} else {
			curr = stm_get_parent_state(stm->base, curr->state);
		}
	}
	if (rc != 0) {
		//printf("stm %08x state %s has no process event %s %08x \n", stm, stm_state->name, e->desc, e);
	}
	return rc;
}

TOOLKIT_API void stm_machine_process(stm_machine_t *stm, stm_event_t *e)
{
	assert(e);

	//printf("begin machine %s process evt:%s %08x\n", stm->base->name, e->desc, e);

	if (stm->last_error || stm->current_state == STM_STATE_EXIT) {
		//printf("end machine %s process evt:%s %08x, stm->last_error = %d, stm->state=%d\n", stm->base->name, e->desc, e, stm->last_error, stm->current_state);
		return;
	}

	stm_event_inc_ref(e);
	list_add_tail(&e->__entry, &stm->dispatch_stack[stm->dispatch_sp]);

	while (stm->current_state != STM_STATE_EXIT && !list_empty(&stm->dispatch_stack[stm->dispatch_sp])) {
		const stm_state_t *stm_state = &stm->base->arr_state[stm->current_state];
		stm_event_t *evt = list_first_entry(&stm->dispatch_stack[stm->dispatch_sp], stm_event_t, __entry);
		int handled_state;
		unsigned int action;
		list_del(&evt->__entry);
		//printf("fetch event:%s %08x, stm->state:%s\n", evt->desc, evt, stm_state->name);
		if (__stm_state_process_event(stm, stm_state, evt, &action, &handled_state) == 0) {
			if (stm->last_error) {
				__stm_trans_state(stm, STM_STATE_EXIT);
			} else {
				if (action != STM_STATE_INIT) {
					int dst_state = action;
					if (dst_state != stm->current_state) {
						__stm_trans_state(stm, dst_state);
					}
				}
			}
		}
		stm_event_dec_ref(evt);
	}

	if (stm->current_state == STM_STATE_EXIT) {
		__stm_machine_clear_defer_list(stm);
	}
}

TOOLKIT_API void stm_machine_defer_process(stm_machine_t *stm, stm_event_t *e)
{
	assert(e);
	stm_event_inc_ref(e);
	list_add_tail(&e->__entry, &stm->dispatch_stack[stm->dispatch_sp]);
}

TOOLKIT_API void stm_machine_defer_process_high_priority(stm_machine_t *stm, stm_event_t *e)
{
	assert(e);
	stm_event_inc_ref(e);
	list_add(&e->__entry, &stm->dispatch_stack[stm->dispatch_sp]);
}

TOOLKIT_API int stm_machine_inc_dispatch_level(stm_machine_t *stm)
{
	assert(stm->dispatch_sp < STM_MAX_STATE_DEPTH);
	list_splice(&stm->dispatch_stack[stm->dispatch_sp], &stm->dispatch_stack[stm->dispatch_sp+1]);
	stm->dispatch_sp++;
	return stm->dispatch_sp;
}

TOOLKIT_API int stm_machine_dec_dispatch_level(stm_machine_t *stm)
{
	assert(stm->dispatch_sp > 0);
	list_splice(&stm->dispatch_stack[stm->dispatch_sp], &stm->dispatch_stack[stm->dispatch_sp-1]);
	stm->dispatch_sp--;
	return stm->dispatch_sp;
}

TOOLKIT_API void stm_machine_process_at_dispatch_level(stm_machine_t *stm, int level, stm_event_t *e)
{
	assert(e);
	assert(level >= 0);
	assert(level <= stm->dispatch_sp);
	stm_event_inc_ref(e);
	list_add_tail(&e->__entry, &stm->dispatch_stack[level]);
}

TOOLKIT_API int stm_machine_get_current_dispatch_level(stm_machine_t *stm)
{
	return stm->dispatch_sp;
}

TOOLKIT_API void stm_machine_goto_error(stm_machine_t *stm, int error)
{
	assert(error);
	stm->last_error = error;
}

TOOLKIT_API long stm_machine_inc_ref(stm_machine_t *stm) 
{
	if (stm->host_dispatcher == stm->report_dispatcher) {
		return ++stm->__ref_cnt;
	} else {
		return InterlockedIncrement((LONG*)&stm->__ref_cnt);
	}
}

TOOLKIT_API long stm_machine_dec_ref(stm_machine_t *stm) 
{
	long l;
	if (stm->host_dispatcher == stm->report_dispatcher) {
		l = --stm->__ref_cnt;
	} else {
		l = InterlockedDecrement((LONG*)&stm->__ref_cnt);
	}
	if (l == 0) {
		__stm_machine_destroy(stm);
	}
	return l;
}

TOOLKIT_API const char *stm_machine_get_state_name(stm_machine_t *stm, int state)
{
	if (state == STM_STATE_INIT) {
		return "$INIT$";
	} else if (state == STM_STATE_EXIT) {
		return "$EXIT$";
	} else {
		const stm_state_t *stm_state = stm_get_state(stm->base, state);
		if (stm_state) {
			return stm_state->name;
		} else {
			return "$UNKNOWN$";
		}
	}
}

TOOLKIT_API void stm_machine_need_trans_evt(stm_machine_t *stm, int state, int from_state)
{
	if (!stm->src_trans_evt_mask) {
		stm->src_trans_evt_mask = (int*)zalloc(stm->base->arr_state_size*sizeof(int));
	}
	if (from_state == 0) {
		stm->src_trans_evt_mask[state] = 0;
	} else if (from_state > STM_MAX_STATE) {
		stm->src_trans_evt_mask[state] = 0xffffffff; // accept any
	} else {
		stm->src_trans_evt_mask[state] = 1 << from_state; // accept specific one
	}
}

enum e_stm_common_internal_event
{
	__STM_SYS_INTERNAL_EVT_MSG_QUIT, 
};

#define STM_SYS_INTERNAL_EVT_MSG_QUIT STM_MAKE_EVENT_TYPE(STM_CTX_TRUE, STM_CMD_MSG, STM_CAT_SYSTEM, STM_CLS_COMMON, __STM_SYS_INTERNAL_EVT_MSG_QUIT)



TOOLKIT_API stm_dispatcher_t *stm_dispatcher_create_base(int ctx_size, void **p_ctx)
{
	stm_dispatcher_t *disp;
	assert(ctx_size >= 0);
	disp = (stm_dispatcher_t*)malloc(sizeof(stm_dispatcher_t)+ctx_size);
	if (disp) {
		INIT_LIST_HEAD(&disp->machine_list);
		INIT_LIST_HEAD(&disp->defer_event_list);
		spinlock_init(&disp->lock);
		disp->vtbl = NULL; // set by stm_dispatcher_t inherits
		disp->thread_id = 0; // set by stm_dispatcher_t inherits
		disp->b_in_process = 0;
		if (ctx_size && p_ctx) {
			*p_ctx = (void*)(disp+1);
		}
	}
	return disp;
}

TOOLKIT_API void stm_dispatcher_destroy(stm_dispatcher_t *disp)
{
	assert(disp);
	assert(list_empty(&disp->defer_event_list));
#if defined(DEBUG) || defined(_DEBUG)
	if (!list_empty(&disp->machine_list)) {
		stm_machine_t *pos;
		list_for_each_entry(pos, &disp->machine_list, stm_machine_t, entry) {
			DEBUG_TRACE("stm_dispatcher_destroy %s not destroy!", pos->base->name);
		}
	}
#endif
	assert(list_empty(&disp->machine_list));
	assert(disp->vtbl);
	if (disp->vtbl->on_destroy) {
		(*disp->vtbl->on_destroy)(disp);
	}
	free(disp);	
}

TOOLKIT_API int stm_dispatcher_new_machine(stm_dispatcher_t *host_dispatcher, stm_dispatcher_t *report_dispatcher, int cat, int cls, int report_stm_id, stm_event_t *initial_evt)
{
	assert(host_dispatcher);
	return stm_machine_ptr2id(stm_machine_create(host_dispatcher, report_dispatcher, cat, cls, report_stm_id, initial_evt));
}

TOOLKIT_API void stm_dispatcher_delete_machine(stm_dispatcher_t *host_dispatcher, int stm_id)
{
	if (stm_id) {
		stm_machine_t *stm = stm_machine_id2ptr(stm_id);
		stm_machine_dec_ref(stm);
	}
}

static void __stm_dispatcher_process_event(stm_dispatcher_t *disp, stm_event_t *e)
{
	if (e->dst_stm_id) {
		stm_machine_t *stm;
		stm = stm_machine_id2ptr(e->dst_stm_id);
		DEBUG_TRACE("dispatcher stm:%s process evt:%s", stm->base->name, e->desc);
		stm_machine_inc_ref(stm);
		stm_machine_process(stm, e);
		stm_machine_dec_ref(stm);
	}
}

TOOLKIT_API void stm_dispatcher_process(stm_dispatcher_t *disp, stm_event_t *e)
{
	assert(disp);
	assert(e);

	disp->b_in_process ++;

	__stm_dispatcher_process_event(disp, e);

	while (!list_empty(&disp->defer_event_list)) {
		stm_event_t *te = list_first_entry(&disp->defer_event_list, stm_event_t, __entry);
		list_del(&te->__entry);
		__stm_dispatcher_process_event(disp, te);
		stm_event_dec_ref(te);
	}

	disp->b_in_process --;
}

TOOLKIT_API int stm_dispatcher_post_event(stm_dispatcher_t *disp, stm_event_t *e)
{
	int rc = -1;
	if (disp && e) {
		assert(disp->vtbl);
		if (GetCurrentThreadId() == disp->thread_id && disp->b_in_process) {
			stm_event_inc_ref(e);
			list_add_tail(&e->__entry, &disp->defer_event_list);
		} else {
			rc = (*disp->vtbl->post_event)(disp, e);
		}
	}
	return rc;
}



//
// iocp dispatcher
//

#define IOCP_OP_NONE		-1
#define IOCP_OP_CONNECT		0
#define IOCP_OP_RX_ANY		1
#define IOCP_OP_RX_N		2
#define IOCP_OP_RX_UNTIL	3
#define IOCP_OP_TX			4
#define IOCP_OP_ACCEPT		5
#define IOCP_OP_RX_UDP		6
#define IOCP_OP_TX_UDP		7

#define ACCEPT_ADDR_LEN	(16+sizeof(SOCKADDR))

enum e_udp_conn_internal_evt
{
	__EVT_MSG_UDP_CONN_RECV_RESULT, // param1: error, param2: transfer bytes
	__EVT_MSG_UDP_CONN_SEND_RESULT, // param1: error, param2: transfer bytes
	__EVT_MSG_UDP_CONN_LATE_CLOSE,
};

#define EVT_MSG_UDP_CONN_RECV_RESULT		STM_MAKE_EVENT_TYPE(STM_CTX_TRUE, STM_CMD_MSG, STM_CAT_SYSTEM, STM_CLS_ID_UDP_CONNECTION, __EVT_MSG_UDP_CONN_RECV_RESULT)
#define EVT_MSG_UDP_CONN_SEND_RESULT		STM_MAKE_EVENT_TYPE(STM_CTX_TRUE, STM_CMD_MSG, STM_CAT_SYSTEM, STM_CLS_ID_UDP_CONNECTION, __EVT_MSG_UDP_CONN_SEND_RESULT)
#define EVT_MSG_UDP_CONN_LATE_CLOSE			STM_MAKE_EVENT_TYPE(STM_CTX_TRUE, STM_CMD_MSG, STM_CAT_SYSTEM, STM_CLS_ID_UDP_CONNECTION, __EVT_MSG_UDP_CONN_LATE_CLOSE)

enum e_tcp_conn_internal_evt
{
	__EVT_MSG_TCP_CONN_CONNECT_RESULT, // param1: error
	__EVT_MSG_TCP_CONN_RECV_RESULT, // param1: error, param2: transfer bytes
	__EVT_MSG_TCP_CONN_SEND_RESULT, // param1: error, param2: transfer bytes
	__EVT_MSG_TCP_CONN_LATE_CLOSE,
};

#define EVT_MSG_TCP_CONN_CONNECT_RESULT		STM_MAKE_EVENT_TYPE(STM_CTX_TRUE, STM_CMD_MSG, STM_CAT_SYSTEM, STM_CLS_ID_TCP_CONNECTION, __EVT_MSG_TCP_CONN_CONNECT_RESULT)
#define EVT_MSG_TCP_CONN_RECV_RESULT		STM_MAKE_EVENT_TYPE(STM_CTX_TRUE, STM_CMD_MSG, STM_CAT_SYSTEM, STM_CLS_ID_TCP_CONNECTION, __EVT_MSG_TCP_CONN_RECV_RESULT)
#define EVT_MSG_TCP_CONN_SEND_RESULT		STM_MAKE_EVENT_TYPE(STM_CTX_TRUE, STM_CMD_MSG, STM_CAT_SYSTEM, STM_CLS_ID_TCP_CONNECTION, __EVT_MSG_TCP_CONN_SEND_RESULT)
#define EVT_MSG_TCP_CONN_LATE_CLOSE			STM_MAKE_EVENT_TYPE(STM_CTX_TRUE, STM_CMD_MSG, STM_CAT_SYSTEM, STM_CLS_ID_TCP_CONNECTION, __EVT_MSG_TCP_CONN_LATE_CLOSE)

enum e_tcp_ap_internal_evt
{
	__EVT_MSG_TCP_AP_ACCEPT_RESULT, // param1: error, param2: id
	__EVT_MSG_TCP_AP_CONNECTION_EXIT,
};

#define EVT_MSG_TCP_AP_ACCEPT_RESULT		STM_MAKE_EVENT_TYPE(STM_CTX_TRUE, STM_CMD_MSG, STM_CAT_SYSTEM, STM_CLS_ID_TCP_ACCEPTOR, __EVT_MSG_TCP_AP_ACCEPT_RESULT)
#define EVT_MSG_TCP_AP_CONNECTION_EXIT		STM_MAKE_EVENT_TYPE(STM_CTX_TRUE, STM_CMD_MSG, STM_CAT_SYSTEM, STM_CLS_ID_TCP_ACCEPTOR, __EVT_MSG_TCP_AP_CONNECTION_EXIT)

typedef struct iocp_op_t
{
	OVERLAPPED ov;
	int type;
	stm_machine_t *stm;
	DWORD dwLastError;
	union {
		struct {
			WSABUF wsabuf;
			char *buf;
			int offset;
			int size;
			DWORD dwFlags;
		}any;
		struct {
			WSABUF wsabuf;
			char *buf;
			int offset;
			int size;
			DWORD dwFlags;
		}n;
		struct {
			WSABUF wsabuf;
			char *buf;
			int offset;
			int size;
			int max_len;
			const char *delimer;
			DWORD dwFlags;
		}t;
		struct {
			WSABUF wsabuf;
			char *buf;
			int offset;
			int size;
		}tx;
		struct {
			SOCKET fd;
			int id;
			char accept_buf[2*ACCEPT_ADDR_LEN];
		}ap;
		struct {
			WSABUF wsabuf;
			int id;
			char *buf;
			int offset;
			int size;
			struct sockaddr_in src_addr;
			int from_len;
			int user_data;
			DWORD dwFlags;
		}udp_rx;
		struct {
			WSABUF wsabuf;
			int id;
			char *buf;
			int offset;
			int size;
			int user_data;
		}udp_tx;
	};
}iocp_op_t;

static void iocp_op_clear(iocp_op_t *op)
{
	if (op->type == IOCP_OP_RX_ANY) {
		if (op->any.buf) {
			free(op->any.buf);
			op->any.buf = NULL;
		}
	} else if (op->type == IOCP_OP_RX_N) {
		if (op->n.buf) {
			free(op->n.buf);
			op->n.buf = NULL;
		}
	} else if (op->type == IOCP_OP_RX_UNTIL) {
		if (op->t.buf) {
			free(op->t.buf);
			op->t.buf = NULL;
		}
	} else if (op->type == IOCP_OP_TX) {
		if (op->tx.buf) {
			free(op->tx.buf);
			op->tx.buf = NULL;
		}
	} else if (op->type == IOCP_OP_ACCEPT) {
		if (op->ap.fd != INVALID_SOCKET) {
			closesocket(op->ap.fd);
			op->ap.fd = INVALID_SOCKET;
		}
	} else if(op->type == IOCP_OP_RX_UDP) {
		if (op->udp_rx.buf) {
			free(op->udp_rx.buf);
			op->udp_rx.buf = NULL;
		}
	} else if (op->type == IOCP_OP_TX_UDP) {
		if (op->udp_tx.buf) {
			free(op->udp_tx.buf);
			op->udp_tx.buf = NULL;
		}
	} else {

	}
	op->dwLastError = 0;
	op->type = IOCP_OP_NONE;
}

static void iocp_op_prepare_for_rx_any(iocp_op_t *op, int suggest_size)
{
	if (op->type == IOCP_OP_RX_ANY) {
		if (op->any.size != suggest_size) {
			free(op->any.buf);
			op->any.buf = (char*)malloc(suggest_size);
			op->any.size = suggest_size;
		}
	} else {
		iocp_op_clear(op);
		op->type = IOCP_OP_RX_ANY;
		op->any.size = suggest_size;
		op->any.buf = (char*)malloc(op->any.size);
	}
	op->any.offset = 0;
	memset(&op->ov, 0, sizeof(OVERLAPPED));
}

static void iocp_op_prepare_for_rx_n(iocp_op_t *op, int n)
{
	if (op->type == IOCP_OP_RX_N) {
		if (op->n.size != n) {
			free(op->n.buf);
			op->n.buf = (char*)malloc(n);
			op->n.size = n;
		}
	} else {
		iocp_op_clear(op);
		op->n.buf = (char*)malloc(n);
		op->n.size = n;
		op->type = IOCP_OP_RX_N;
	}
	op->n.offset = 0;
	memset(&op->ov, 0, sizeof(OVERLAPPED));
}

static void iocp_op_prepare_for_rx_until(iocp_op_t *op, int max_len, const char *delimer)
{
	if (op->type == IOCP_OP_RX_UNTIL) {
		if (op->t.size < max_len) {
			free(op->t.buf);
			op->t.size = max_len;
			op->t.buf = (char*)malloc(max_len);
		}
		op->t.max_len = max_len;
		op->t.delimer = delimer;
	} else {
		iocp_op_clear(op);
		op->type = IOCP_OP_RX_UNTIL;
		op->t.delimer = delimer;
		op->t.size = max(1<<12, max_len);
		op->t.max_len = max_len;
		op->t.buf = (char*)malloc(op->t.size);
	}
	op->t.offset = 0;
	memset(&op->ov, 0, sizeof(OVERLAPPED));
}

static void iocp_op_prepare_for_tx(iocp_op_t *op, char *buf, int offset, int size)
{
	iocp_op_clear(op);
	op->tx.buf = (char*)memdup(buf+offset, size);
	op->tx.offset = 0;
	op->tx.size = size;
	op->type = IOCP_OP_TX;
	memset(&op->ov, 0, sizeof(OVERLAPPED));
}

static void iocp_op_prepare_for_ap(iocp_op_t *op, SOCKET fd, int id)
{
	iocp_op_clear(op);
	op->ap.fd = fd;
	op->type = IOCP_OP_ACCEPT;
	op->ap.id = id;
	memset(&op->ov, 0, sizeof(OVERLAPPED));
}

static void iocp_op_prepare_for_udp_rx(iocp_op_t *op, int id, int max_pkt_size, int user_data)
{
	if (op->type == IOCP_OP_RX_UDP) {
		if (op->udp_rx.size != max_pkt_size) {
			op->udp_rx.buf = (char*)realloc(op->udp_rx.buf, max_pkt_size);
			op->udp_rx.size = max_pkt_size;
		}
	} else {
		iocp_op_clear(op);
		op->type = IOCP_OP_RX_UDP;
		op->udp_rx.buf = (char*)malloc(max_pkt_size);
		op->udp_rx.size = max_pkt_size;
	}
	op->udp_rx.offset = 0;
	op->udp_rx.id = id;
	op->udp_rx.from_len = sizeof(SOCKADDR);
	op->udp_rx.user_data = user_data;
	memset(&op->ov, 0, sizeof(OVERLAPPED));
}

static void iocp_op_prepare_for_udp_tx(iocp_op_t *op, int id, char *buf, int offset, int size, int user_data)
{
	iocp_op_clear(op);
	op->udp_tx.buf = (char*)memdup(buf+offset, size);
	op->udp_tx.size = size;
	op->udp_tx.offset = 0;
	op->udp_tx.id = id;
	op->udp_tx.user_data = user_data;
	op->type = IOCP_OP_TX_UDP;
	memset(&op->ov, 0, sizeof(OVERLAPPED));
}

static void iocp_op_prepare_for_connect(iocp_op_t *op)
{
	iocp_op_clear(op);
	op->type = IOCP_OP_CONNECT;
	memset(&op->ov, 0, sizeof(OVERLAPPED));
}

typedef struct iocp_dispatcher_t
{
	HANDLE iocp_handle;
	timer_queue_t *timer_queue;
	HANDLE worker_thread_handle;
}iocp_dispatcher_t;

static stm_event_t *iocp_op_make_event(iocp_op_t *op, BOOL bRet, ULONG_PTR pKey, DWORD dwTransfer);
static void term_iocp_dispatcher(stm_dispatcher_t *disp);

static unsigned int __stdcall iocp_dispatcher_worker_proc(void *arg)
{
	stm_dispatcher_t *disp = (stm_dispatcher_t*)arg;
	stm_iocp_dispatcher_run(disp);
	return 0;
}

static __inline int iocp_post_event(stm_dispatcher_t *disp, stm_event_t *e)
{
	iocp_dispatcher_t *iocp_ctx = (iocp_dispatcher_t *)stm_dispatcher_get_context(disp);
	assert(e);
	stm_event_inc_ref(e);
	if (!PostQueuedCompletionStatus(iocp_ctx->iocp_handle, -1, (ULONG_PTR)e, (LPOVERLAPPED)e)) {
		stm_event_dec_ref(e);
		return -1;
	}
	return 0;
}

static int init_iocp_dispatcher(stm_dispatcher_t *disp, unsigned int thread_id)
{
	iocp_dispatcher_t *iocp_ctx = (iocp_dispatcher_t *)stm_dispatcher_get_context(disp);
	int rc;
	memset(iocp_ctx, 0, sizeof(iocp_dispatcher_t));
	rc = timer_heap_create(&iocp_ctx->timer_queue);
	if (rc == 0) {
		iocp_ctx->iocp_handle = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
		if (iocp_ctx->iocp_handle) {
			if (thread_id == 0) {
				iocp_ctx->worker_thread_handle = (HANDLE)_beginthreadex(NULL, 
					0, &iocp_dispatcher_worker_proc, disp, 0, &disp->thread_id);
				if (!iocp_ctx->worker_thread_handle) {
					rc = -1;
				}
			} else {
				iocp_ctx->worker_thread_handle = NULL;
				disp->thread_id = thread_id;
			}
		} else {
			rc = -1;
		}
		if (rc == -1) {
			term_iocp_dispatcher(disp);
		}
	}
	return rc;
}

static void term_iocp_dispatcher(stm_dispatcher_t *disp)
{
	iocp_dispatcher_t *iocp_ctx = (iocp_dispatcher_t *)stm_dispatcher_get_context(disp);
	if (iocp_ctx->worker_thread_handle) {
		stm_iocp_dispatcher_quit(disp);
		WaitForSingleObject(iocp_ctx->worker_thread_handle, INFINITE);
		CloseHandle(iocp_ctx->worker_thread_handle);
		iocp_ctx->worker_thread_handle = NULL;
	}
	if (iocp_ctx->timer_queue) {
		timer_queue_destroy(iocp_ctx->timer_queue);
		iocp_ctx->timer_queue = NULL;
	}
	if (iocp_ctx->iocp_handle) {
		CloseHandle(iocp_ctx->iocp_handle);
		iocp_ctx->iocp_handle = NULL;
	}
}

static void iocp_dispatcher_timer_callback(timer_queue_t *q, timer_entry *timer, int err)
{
	if (!err) {
		stm_dispatcher_t *disp = (stm_dispatcher_t *)timer->user_data;
		int stm_id = timer->id;
		stm_event_t *e = stm_event_create_base(STM_SYS_EVT_MSG_TIMER, stm_id, stm_id, (int)timer, 0, 0, 0);
		stm_dispatcher_process(disp, e);
		stm_event_dec_ref(e);
	}
}

static void iocp_dispatcher_on_destroy(stm_dispatcher_t *disp)
{
	term_iocp_dispatcher(disp);
}

static int iocp_dispatcher_post_event(stm_dispatcher_t *disp, stm_event_t *e)
{
	return iocp_post_event(disp, e);
}

static int iocp_dispatcher_schedule_timer(stm_dispatcher_t *disp, int stm_id, unsigned int interval)
{
	iocp_dispatcher_t *iocp_ctx = (iocp_dispatcher_t *)stm_dispatcher_get_context(disp);
	timer_entry *e = MALLOC_T(timer_entry);
	e->id = stm_id;
	e->user_data = disp;
	e->cb = &iocp_dispatcher_timer_callback;
	if (timer_queue_schedule(iocp_ctx->timer_queue, e, interval) != 0) {
		free(e);
		return STM_INVALID_TIMER;
	}
	return (int)e;
}

static void iocp_dispatcher_cancel_timer(stm_dispatcher_t *disp, int timer_id)
{
	iocp_dispatcher_t *iocp_ctx = (iocp_dispatcher_t *)stm_dispatcher_get_context(disp);
	timer_entry *e = (timer_entry *)timer_id;
	assert(timer_id);
	timer_queue_cancel(iocp_ctx->timer_queue, e, 0);
	free(e);
}

static const struct stm_dispatcher_vtbl_t iocp_dispatcher_vtbl = 
{
	"windows iocp dispatcher",
	&iocp_dispatcher_on_destroy,
	&iocp_dispatcher_post_event,
	&iocp_dispatcher_schedule_timer,
	&iocp_dispatcher_cancel_timer,
};

TOOLKIT_API stm_dispatcher_t *stm_iocp_dispatcher_create(unsigned int thread_id)
{
	iocp_dispatcher_t *iocp_ctx;
	stm_dispatcher_t *disp = stm_dispatcher_create_base(sizeof(iocp_dispatcher_t), (void**)&iocp_ctx);
	if (disp) {
		disp->vtbl = &iocp_dispatcher_vtbl;
		if (init_iocp_dispatcher(disp, thread_id) != 0) {
			stm_dispatcher_destroy(disp);
			disp = NULL;
		}
	}
	return disp;
}

TOOLKIT_API int stm_iocp_dispatcher_run(stm_dispatcher_t *disp)
{
	iocp_dispatcher_t *iocp_ctx = (iocp_dispatcher_t *)stm_dispatcher_get_context(disp);
	DWORD dwTimeout = MAX_TIMEOUT;

	for (;;) {
		DWORD dwTransfer = 0;
		ULONG_PTR pKey = 0;
		LPOVERLAPPED lpov = NULL;
		int next = 0;
		stm_event_t *e = NULL;
		BOOL bRet = GetQueuedCompletionStatus(iocp_ctx->iocp_handle, &dwTransfer, &pKey, &lpov, dwTimeout);
		if (bRet) {
			if (dwTransfer == -1) { // event
				e = (stm_event_t*)lpov;
				if (e->type == STM_SYS_INTERNAL_EVT_MSG_QUIT) {
					stm_event_dec_ref(e);
					break;
				}
			} else {
				iocp_op_t *op = container_of(lpov, iocp_op_t, ov);
				e = iocp_op_make_event(op, bRet, pKey, dwTransfer);
			//	if (dwTransfer ==0 && op->type == IOCP_OP_RX_N) {
					//printf("GetQueuedCompletionStatus, dwTransfer:%d, bRet:%d, optype:%d\n", dwTransfer, bRet, op->type);
				//}
				//printf("GetQueuedCompletionStatus, dwTransfer:%d, bRet:%d, optype:%d\n", dwTransfer, bRet, op->type);
			}
		} else {
			if (lpov) {
				iocp_op_t *op = container_of(lpov, iocp_op_t, ov);
				e = iocp_op_make_event(op, bRet, pKey, dwTransfer);
			}
			
		}
		if (e) {
			stm_dispatcher_process(disp, e);
			stm_event_dec_ref(e);
		}
		timer_queue_poll(iocp_ctx->timer_queue, &next);
		dwTimeout = min(next, MAX_TIMEOUT);
	}
	return 0;
}

TOOLKIT_API int stm_iocp_dispatcher_quit(stm_dispatcher_t *disp)
{
	stm_event_t *e = stm_event_create_base(STM_SYS_INTERNAL_EVT_MSG_QUIT, 0, 0, 0, 0, 0, 0);
	int rc;
	if (e) {
		rc = iocp_post_event(disp, e);
		stm_event_dec_ref(e);
	} else {
		rc = -1;
	}
	return rc;
}

// tcp connection stm

typedef struct tcp_conn_ctx_t
{
	SOCKET fd;
	array_header_t *arr_server;
	iocp_op_t tx;
	iocp_op_t rx;
	int pending_tx;
	int pending_rx;
	char *cache_rx_buf;
	int cache_rx_len;
	int cache_rx_offset;
	int connected_level;
	stm_conn_endpt_t *current_endpt;
	int acceptor_stm_id;
	int connect_timer_id;
	int connect_timeout;
}tcp_conn_ctx_t;

static int tcp_conn_prepare_socket(stm_machine_t *stm, SOCKET fd)
{
	u_long v = 1;
	int rc = ioctlsocket(fd, FIONBIO, &v);
	if (rc == 0) {
		iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
		tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
		if (!CreateIoCompletionPort((HANDLE)fd, iocp_disp->iocp_handle, (ULONG_PTR)stm, 0))
			rc = -1;
	}
	return rc;
}

TOOLKIT_API stm_conn_endpt_t *tcp_conn_get_best_endpt(tcp_conn_ctx_t *conn_ctx)
{
	if (conn_ctx->arr_server->nelts) {
		int i;
		int n;
		for (i = 0; i < conn_ctx->arr_server->nelts; ++i) {
			stm_conn_endpt_t *t = &ARRAY_IDX(conn_ctx->arr_server, i, stm_conn_endpt_t);
			if (t->b_primary) {
				ARRAY_DEL_STRUCT(conn_ctx->arr_server, i, stm_conn_endpt_t);
				return &ARRAY_IDX(conn_ctx->arr_server, conn_ctx->arr_server->nelts, stm_conn_endpt_t);
			}
		}
		n = 0;
		for (i = 0; i < conn_ctx->arr_server->nelts; ++i) {
			stm_conn_endpt_t *t = &ARRAY_IDX(conn_ctx->arr_server, i, stm_conn_endpt_t);
			n += t->weight;
		}
		if (n != 0) {
			int p = 0;
			int q;
			srand((unsigned int)time(NULL));
			q = rand() % n;
			for (i = 0; i < conn_ctx->arr_server->nelts; ++i) {
				stm_conn_endpt_t *t = &ARRAY_IDX(conn_ctx->arr_server, i, stm_conn_endpt_t);
				if (q >= p && q <= p+t->weight) {
					ARRAY_DEL_STRUCT(conn_ctx->arr_server, i, stm_conn_endpt_t);
					return &ARRAY_IDX(conn_ctx->arr_server, conn_ctx->arr_server->nelts, stm_conn_endpt_t);
				}
				p += t->weight;
			}
		}
		ARRAY_DEL_STRUCT(conn_ctx->arr_server, 0, stm_conn_endpt_t);
		return &ARRAY_IDX(conn_ctx->arr_server, conn_ctx->arr_server->nelts, stm_conn_endpt_t);
	}
	return NULL;
}

static int start_connect(stm_machine_t *stm)
{
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);

	stm_conn_endpt_t *endpt = tcp_conn_get_best_endpt(conn_ctx);
	if (endpt) {		
		struct sockaddr_in addr = {0};
		addr.sin_family = AF_INET;
		addr.sin_addr.s_addr = endpt->server_ip;
		addr.sin_port = htons(endpt->server_port);
		conn_ctx->current_endpt = endpt;
		{
			DWORD dwBytes;
			BOOL (PASCAL FAR * lpfnConnectEx) (IN SOCKET s,
				IN const struct sockaddr FAR *name,
				IN int namelen,
				IN PVOID lpSendBuffer OPTIONAL,
				IN DWORD dwSendDataLength,
				OUT LPDWORD lpdwBytesSent,
				IN LPOVERLAPPED lpOverlapped
				);
			GUID GuidConnectEx = WSAID_CONNECTEX;
			if (WSAIoctl(conn_ctx->fd, SIO_GET_EXTENSION_FUNCTION_POINTER, &GuidConnectEx, sizeof(GuidConnectEx), &lpfnConnectEx, sizeof(lpfnConnectEx), &dwBytes, NULL, NULL) == 0) {
				iocp_op_prepare_for_connect(&conn_ctx->tx);
				if ((*lpfnConnectEx)(conn_ctx->fd, (struct sockaddr*)&addr, sizeof(addr), 0, 0, 0, &conn_ctx->tx.ov)) {
					return 0;
				}
			}
		}
	}

	return -1;
}

static int RecvEx(tcp_conn_ctx_t *conn_ctx, LPWSABUF lpBuf, LPDWORD pdwBytesTransfer)
{
	if (conn_ctx->cache_rx_buf) {
		int cache_size = conn_ctx->cache_rx_len - conn_ctx->cache_rx_offset;
		if ((int)lpBuf->len >= cache_size) {
			memcpy(lpBuf->buf, conn_ctx->cache_rx_buf+conn_ctx->cache_rx_offset, cache_size);
			conn_ctx->cache_rx_offset += cache_size;
			*pdwBytesTransfer = cache_size;
		} else {
			memcpy(lpBuf->buf, conn_ctx->cache_rx_buf+conn_ctx->cache_rx_offset, lpBuf->len);
			conn_ctx->cache_rx_offset += lpBuf->len;
			*pdwBytesTransfer = lpBuf->len;
		}
		if (conn_ctx->cache_rx_len == conn_ctx->cache_rx_offset) {
			free(conn_ctx->cache_rx_buf);
			conn_ctx->cache_rx_buf = NULL;
		}
		return 0;
	} else {
		iocp_op_t *op = &conn_ctx->rx;
		DWORD *lpdwFlag;
		if (op->type == IOCP_OP_RX_ANY) {
			lpdwFlag = &op->any.dwFlags;
		} else if (op->type == IOCP_OP_RX_N) {
			lpdwFlag = &op->n.dwFlags;
		} else if (op->type == IOCP_OP_RX_UNTIL) {
			lpdwFlag = &op->t.dwFlags;
		} else {
			assert(0);
		}
		*lpdwFlag = 0;
		return WSARecv(conn_ctx->fd, lpBuf, 1, pdwBytesTransfer, lpdwFlag, &conn_ctx->rx.ov, NULL);
	}
}

static int start_recv_any(stm_machine_t *stm, int suggest_size)
{
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	DWORD dwBytesTransfer;
	iocp_op_prepare_for_rx_any(&conn_ctx->rx, suggest_size);
	conn_ctx->rx.any.wsabuf.buf = conn_ctx->rx.any.buf;
	conn_ctx->rx.any.wsabuf.len = conn_ctx->rx.any.size;
	return RecvEx(conn_ctx, &conn_ctx->rx.any.wsabuf, &dwBytesTransfer);
}

static int start_recv_n(stm_machine_t *stm, int n)
{
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	DWORD dwBytesTransfer = 0;
	iocp_op_prepare_for_rx_n(&conn_ctx->rx, n);
	conn_ctx->rx.n.wsabuf.buf = conn_ctx->rx.n.buf;
	conn_ctx->rx.n.wsabuf.len = conn_ctx->rx.n.size;
	return RecvEx(conn_ctx, &conn_ctx->rx.n.wsabuf, &dwBytesTransfer);
}

static int continue_recv_n(stm_machine_t *stm)
{
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	DWORD dwBytesTransfer = 0;
	conn_ctx->rx.n.wsabuf.buf = conn_ctx->rx.n.buf+conn_ctx->rx.n.offset;
	conn_ctx->rx.n.wsabuf.len = conn_ctx->rx.n.size-conn_ctx->rx.n.offset;
	memset(&conn_ctx->rx.ov, 0, sizeof(OVERLAPPED));
	return RecvEx(conn_ctx, &conn_ctx->rx.n.wsabuf, &dwBytesTransfer);
}

static int start_recv_until(stm_machine_t *stm, int max_len, const char *delimer)
{
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	DWORD dwBytesTransfer = 0;
	iocp_op_prepare_for_rx_until(&conn_ctx->rx, max_len, delimer);
	conn_ctx->rx.t.wsabuf.buf = conn_ctx->rx.t.buf;
	conn_ctx->rx.t.wsabuf.len = conn_ctx->rx.t.size;
	return RecvEx(conn_ctx, &conn_ctx->rx.t.wsabuf, &dwBytesTransfer);
}

static __inline char *find_recv_until_delimer(stm_machine_t *stm)
{
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	return memstr1(conn_ctx->rx.t.buf, conn_ctx->rx.t.offset, conn_ctx->rx.t.delimer);
}

static int continue_recv_until(stm_machine_t *stm)
{
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	int rc;
	DWORD dwBytesTransfer;
	if (conn_ctx->rx.t.offset < conn_ctx->rx.t.size) {
		conn_ctx->rx.t.wsabuf.buf = conn_ctx->rx.t.buf + conn_ctx->rx.t.offset;
		conn_ctx->rx.t.wsabuf.len = conn_ctx->rx.t.size - conn_ctx->rx.t.offset;
		memset(&conn_ctx->rx.ov, 0, sizeof(OVERLAPPED));
		rc = RecvEx(conn_ctx, &conn_ctx->rx.t.wsabuf, &dwBytesTransfer);
	} else {
		rc = -1;
	}
	return rc;
}

static int start_send(stm_machine_t *stm)
{
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	DWORD dwBytesTransfer = 0;
	conn_ctx->tx.tx.wsabuf.buf = conn_ctx->tx.tx.buf + conn_ctx->tx.tx.offset;
	conn_ctx->tx.tx.wsabuf.len = conn_ctx->tx.tx.size - conn_ctx->tx.tx.offset;
	return WSASend(conn_ctx->fd, &conn_ctx->tx.tx.wsabuf, 1, &dwBytesTransfer, 0, &conn_ctx->tx.ov, NULL);	
}

static int continue_send(stm_machine_t *stm)
{
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	DWORD dwBytesTransfer = 0;
	conn_ctx->tx.tx.wsabuf.buf = conn_ctx->tx.tx.buf + conn_ctx->tx.tx.offset;
	conn_ctx->tx.tx.wsabuf.len = conn_ctx->tx.tx.size - conn_ctx->tx.tx.offset;
	memset(&conn_ctx->tx.ov, 0, sizeof(OVERLAPPED));
	return WSASend(conn_ctx->fd, &conn_ctx->tx.tx.wsabuf, 1, &dwBytesTransfer, 0, &conn_ctx->tx.ov, NULL);	
}

static stm_event_t *make_tcp_rx_res_event(stm_machine_t *stm, int rc, const char *buf, int size)
{
	stm_tcp_conn_rx_msg_t *rx_msg;
	int ctx_len = sizeof(stm_tcp_conn_rx_msg_t) + size;
	stm_event_t *te = stm_event_create_base(STM_TCP_CONN_EVT_RES_RX, stm->stm_id, stm->report_stm_id, rc, rc == 0 ? size : WSAGetLastError(), ctx_len, (void**)&rx_msg);
	rx_msg->buf = (char*)(rx_msg + 1);
	rx_msg->offset = 0;
	rx_msg->size = size;
	if (rc == 0 && size > 0) {
		memcpy(rx_msg->buf, buf, rx_msg->size);
	}
	return te;
}

static int tcp_conn_handle_tx_cmd(stm_machine_t *stm, stm_event_t *e)
{
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	int rc;
	stm_tcp_conn_tx_t *tx_evt = (stm_tcp_conn_tx_t *)stm_event_get_context(e);

	iocp_op_prepare_for_tx(&conn_ctx->tx, tx_evt->buf, tx_evt->offset, tx_evt->size);
	rc = start_send(stm);
	if (rc && WSAGetLastError() != WSA_IO_PENDING) {
		stm_event_t *te = stm_event_create_base(STM_TCP_CONN_EVT_RES_TX, stm->stm_id, stm->report_stm_id, rc, WSAGetLastError(), 0, NULL);
		stm_dispatcher_post_event(stm->report_dispatcher, te);
		stm_event_dec_ref(te);
	} else {
		conn_ctx->pending_tx++;
		rc = 0;
	}
	return rc;
}

static int tcp_conn_handle_rx_cmd(stm_machine_t *stm, stm_event_t *e)
{
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	stm_tcp_conn_rx_t *rx_evt = (stm_tcp_conn_rx_t *)stm_event_get_context(e);
	int rc = 0;
	if (rx_evt->type == STM_TCP_CONN_RX_ANY) {
		int suggest_size = rx_evt->length == 0 ? 1<<12 : rx_evt->length;
		rc = start_recv_any(stm, suggest_size);
	} else if (rx_evt->type == STM_TCP_CONN_RX_N) {
		int n = rx_evt->length;
		rc = start_recv_n(stm, n);
	} else if (rx_evt->type == STM_TCP_CONN_RX_UNTIL) {
		int max_len = rx_evt->length;
		const char *delimer = rx_evt->delimer;
		rc = start_recv_until(stm, max_len, delimer);
	} else {
		assert(0);
		rc = -1;
	}
	if (rc && WSAGetLastError() != WSA_IO_PENDING) {
		stm_event_t *te = stm_event_create_base(STM_TCP_CONN_EVT_RES_RX, stm->stm_id, stm->report_stm_id, rc, WSAGetLastError(), 0, 0);
		stm_dispatcher_post_event(stm->report_dispatcher, te);
		stm_event_dec_ref(te);
	} else {
		conn_ctx->pending_rx++;
		rc = 0;
	}
	return rc;
}

static int tcp_conn_handle_rx_result(stm_machine_t *stm, stm_event_t *e)
{
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	int rc = e->param1;
	//printf("tcp_conn_handle_rx_result rc:%d, size:%d GetLastError=%d\n", e->param1, e->param2, WSAGetLastError());
	WSASetLastError(0);
	if (rc == 0) {
		int size = e->param2;
		if (conn_ctx->rx.type == IOCP_OP_RX_ANY) {
			if (size == 0) {
				rc = -1;
			} else {
				stm_event_t *te = make_tcp_rx_res_event(stm, rc, conn_ctx->rx.any.buf, conn_ctx->rx.any.offset);
				stm_dispatcher_post_event(stm->report_dispatcher, te);
				stm_event_dec_ref(te);
				conn_ctx->pending_rx--;
			}
		} else if (conn_ctx->rx.type == IOCP_OP_RX_N) {
			if (size == 0) {
				rc = -1;
			} else {
				if (conn_ctx->rx.n.offset != conn_ctx->rx.n.size) {
					rc = continue_recv_n(stm);
				} else {
					stm_event_t *te = make_tcp_rx_res_event(stm, rc, conn_ctx->rx.n.buf, conn_ctx->rx.n.offset);
					stm_dispatcher_post_event(stm->report_dispatcher, te);
					stm_event_dec_ref(te);
					conn_ctx->pending_rx--;
				}
			}
		} else if (conn_ctx->rx.type == IOCP_OP_RX_UNTIL) {
			if (size == 0) {
				rc = -1;
			} else {
				char *str;
				if ((str = find_recv_until_delimer(stm)) == NULL && conn_ctx->rx.t.offset != conn_ctx->rx.t.max_len) {
					rc = continue_recv_until(stm);
				} else {
					if (str == NULL && conn_ctx->rx.t.offset == conn_ctx->rx.t.max_len)  {
						rc = -1;
					}
					if (rc == 0) {
						const char *delimer = conn_ctx->rx.t.delimer;
						int left_size = conn_ctx->rx.t.offset - (str - conn_ctx->rx.t.buf);
						stm_event_t *te;
						str += strlen(delimer);
						if (left_size) {
							assert(!conn_ctx->cache_rx_buf);
							conn_ctx->cache_rx_buf = (char*)memdup(str, left_size);
							conn_ctx->cache_rx_len = left_size;
							conn_ctx->cache_rx_offset = 0;
						}
						te = make_tcp_rx_res_event(stm, rc, conn_ctx->rx.t.buf, conn_ctx->rx.t.offset - left_size);
						stm_dispatcher_post_event(stm->report_dispatcher, te);
						stm_event_dec_ref(te);
						conn_ctx->pending_rx--;
					}
				}
			}
		} else {
			assert(0);
			rc = -1;
		}
	}
	if (rc) {
		if (WSAGetLastError() != WSA_IO_PENDING) {
			stm_event_t *te = make_tcp_rx_res_event(stm, rc, 0, 0);
			stm_dispatcher_post_event(stm->report_dispatcher, te);
			stm_event_dec_ref(te);
			conn_ctx->pending_rx--;
		} else {
			rc = 0;
		}
	}
	return rc;
}

static int tcp_conn_handle_tx_result(stm_machine_t *stm, stm_event_t *e)
{
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	int rc = e->param1;
	WSASetLastError(0);
	if (rc == 0) {
		int size = e->param2;
		if (conn_ctx->tx.tx.offset != conn_ctx->tx.tx.size) {
			rc = continue_send(stm);
		} else {
			stm_event_t *te = stm_event_create_base(STM_TCP_CONN_EVT_RES_TX, stm->stm_id, stm->report_stm_id, rc, conn_ctx->tx.tx.offset, 0, 0);
			stm_dispatcher_post_event(stm->report_dispatcher, te);
			stm_event_dec_ref(te);
			conn_ctx->pending_tx--;
		}
	}
	if (rc != 0) {
		if (WSAGetLastError() != WSA_IO_PENDING) {
			stm_event_t *te = stm_event_create_base(STM_TCP_CONN_EVT_RES_TX, stm->stm_id, stm->report_stm_id, rc, WSAGetLastError(), 0, 0);
			stm_dispatcher_post_event(stm->report_dispatcher, te);
			stm_event_dec_ref(te);
			conn_ctx->pending_tx--;
		} else {
			rc = 0;
		}
	}
	return rc;
}

static void on_tcp_start_init(stm_machine_t *stm, const stm_state_t *state)
{

}

static unsigned int on_tcp_start_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	if (e->type == STM_TCP_CONN_EVT_MSG_INITIAL_ARG) {
		stm_tcp_conn_initial_arg_t *conn_arg = (stm_tcp_conn_initial_arg_t *)stm_event_get_context(e);
		if (conn_arg->socket_fd != INVALID_SOCKET) {
			conn_ctx->acceptor_stm_id = conn_arg->acceptor_stm_id;
			conn_ctx->fd = conn_arg->socket_fd;
			conn_arg->socket_fd = INVALID_SOCKET;
			if (tcp_conn_prepare_socket(stm, conn_ctx->fd) == 0) {
				return 0; // goto Connected 
			} else {
				stm_machine_goto_error(stm, -1);
			}
		} else {
			conn_ctx->fd = WSASocketA(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
			if (conn_ctx->fd != INVALID_SOCKET) {
				struct sockaddr_in local_addr = {0};
				local_addr.sin_family = AF_INET;
				local_addr.sin_addr.s_addr = htonl(INADDR_ANY);
				local_addr.sin_port = htons(0);
				if (bind(conn_ctx->fd, (SOCKADDR*)&local_addr, sizeof(local_addr)) == 0) {
					if (tcp_conn_prepare_socket(stm, conn_ctx->fd) == 0) {
						conn_ctx->arr_server = array_copy(conn_arg->arr_server);
						while (conn_ctx->arr_server->nelts > 0) {
							int rc = start_connect(stm);
							if (rc != 0) {
								if (WSAGetLastError() == WSA_IO_PENDING) {
									if (conn_arg->connect_timeout > 0) {
										conn_ctx->connect_timeout = conn_arg->connect_timeout;
										conn_ctx->connect_timer_id = stm_dispatcher_schedule_timer(stm->host_dispatcher, stm->stm_id, conn_ctx->connect_timeout);
									}
									return 1; // goto Connecting 
								}
							} else {
								return 0; // goto Connected 
							}
						}
					}
				}
			}
			if (conn_ctx->fd != INVALID_SOCKET) {
				closesocket(conn_ctx->fd);
				conn_ctx->fd = INVALID_SOCKET;
			}
			stm_machine_goto_error(stm, -1);
		}
	}
	return 0;
}

static void on_tcp_start_exit(stm_machine_t *stm, const stm_state_t *state)
{
	
}

static void on_tcp_connecting_init(stm_machine_t *stm, const stm_state_t *state)
{
	//tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
}

static unsigned int on_tcp_connecting_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	if (e->type == EVT_MSG_TCP_CONN_CONNECT_RESULT) {
		int error = e->param1;
		if (error) {
			while (conn_ctx->arr_server->nelts > 0) {
				int rc = start_connect(stm);
				if (rc == 0 || WSAGetLastError() == WSA_IO_PENDING) {
					return 0; // no jmp
				}
			}
			return 2; // goto Exit
		} else {
			int param1 = conn_ctx->current_endpt->server_ip;
			int param2 = conn_ctx->current_endpt->server_port;
			stm_event_t *te = stm_event_create_base(STM_TCP_CONN_EVT_MSG_CONNECT, stm->stm_id, stm->report_stm_id, param1, param2, 0, 0);
			stm_dispatcher_post_event(stm->host_dispatcher, te);
			stm_event_dec_ref(te);
			setsockopt(conn_ctx->fd, SOL_SOCKET, SO_UPDATE_CONNECT_CONTEXT, NULL, 0);
			return 1; // goto Connected
		}
	} else if (e->type == STM_SYS_EVT_MSG_TIMER) {
		conn_ctx->connect_timer_id = 0;
		CancelIo((HANDLE)conn_ctx->fd);
	}
	return 0;
}

static void on_tcp_connecting_exit(stm_machine_t *stm, const stm_state_t *state)
{
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	if (conn_ctx->connect_timer_id) {
		stm_dispatcher_cancel_timer(stm->host_dispatcher, conn_ctx->connect_timer_id);
		conn_ctx->connect_timer_id = 0;
	}
}

static void on_tcp_connected_init(stm_machine_t *stm, const stm_state_t *state)
{
	
}

static unsigned int on_tcp_connected_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	if (e->type == STM_TCP_CONN_EVT_REQ_RX) {
		int rc = tcp_conn_handle_rx_cmd(stm, e);
		if (rc) {
			return 2; // goto Exit
		} else {
			return 1; // goto R
		}
	} else if (e->type == STM_TCP_CONN_EVT_REQ_TX) {
		int rc = tcp_conn_handle_tx_cmd(stm, e);
		if (rc) {
			return 2; // goto Exit
		} else {
			return 1; // goto T
		}
	} else if (e->type == STM_TCP_CONN_EVT_MSG_DETECT_BROKEN) {
		char buf[32];
		int n = recv(conn_ctx->fd, buf, sizeof(buf), MSG_PEEK);
		if (n == 0 || (n == -1 && WSAGetLastError() != WSAEWOULDBLOCK)) {
			return 1; // Goto Exit
		}
	} else if (e->type == EVT_MSG_TCP_CONN_LATE_CLOSE) {
		return 1; // goto Exit
	} else if (e->type == STM_TCP_CONN_EVT_MSG_CLOSE) {
		return 1; // goto Exit
	}
	return 0;
}

static void on_tcp_connected_exit(stm_machine_t *stm, const stm_state_t *state)
{
}

static void on_tcp_r_init(stm_machine_t *stm, const stm_state_t *state)
{

}

static unsigned int on_tcp_r_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	if (e->type == EVT_MSG_TCP_CONN_RECV_RESULT) {
		int rc = tcp_conn_handle_rx_result(stm, e);
		if (rc) {
			stm_event_t *te = stm_event_create_base(EVT_MSG_TCP_CONN_LATE_CLOSE, stm->stm_id, stm->stm_id, 0, 0, 0, 0);
			stm_machine_process_at_dispatch_level(stm, conn_ctx->connected_level, te);
			stm_event_dec_ref(te);
			return 1; // goto Connected
		} else {
			if (conn_ctx->pending_rx == 0)
				return 1; // goto Connected
		}
	} else if (e->type == STM_TCP_CONN_EVT_REQ_TX) {
		int rc = tcp_conn_handle_tx_cmd(stm, e);
		if (rc) {
			stm_event_t *te = stm_event_create_base(EVT_MSG_TCP_CONN_LATE_CLOSE, stm->stm_id, stm->stm_id, 0, 0, 0, 0);
			stm_machine_process_at_dispatch_level(stm, conn_ctx->connected_level, te);
			stm_event_dec_ref(te);
			return 2; // goto Connected
		} else {
			if (conn_ctx->pending_tx)
				return 1; // goto RT
		}
	}
	return 0;
}
static void on_tcp_r_exit(stm_machine_t *stm, const stm_state_t *state)
{

}

static void on_tcp_t_init(stm_machine_t *stm, const stm_state_t *state)
{

}

static unsigned int on_tcp_t_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);

	if (e->type == EVT_MSG_TCP_CONN_SEND_RESULT) {
		int rc = tcp_conn_handle_tx_result(stm, e);
		if (rc) {
			stm_event_t *te = stm_event_create_base(EVT_MSG_TCP_CONN_LATE_CLOSE, stm->stm_id, stm->stm_id, 0, 0, 0, 0);
			stm_machine_process_at_dispatch_level(stm, conn_ctx->connected_level, te);
			stm_event_dec_ref(te);
			return 1; // goto Connected
		} else {
			if (conn_ctx->pending_tx == 0)
				return 1; // goto Connected
		}
	} else if (e->type == STM_TCP_CONN_EVT_REQ_RX) {
		int rc = tcp_conn_handle_rx_cmd(stm, e);
		if (rc) {
			stm_event_t *te = stm_event_create_base(EVT_MSG_TCP_CONN_LATE_CLOSE, stm->stm_id, stm->stm_id, 0, 0, 0, 0);
			stm_machine_process_at_dispatch_level(stm, conn_ctx->connected_level, te);
			stm_event_dec_ref(te);
			return 1; // goto Connected
		} else {
			if (conn_ctx->pending_rx)
				return 2; // goto RT
		}
	}

	return 0;
}

static void on_tcp_t_exit(stm_machine_t *stm, const stm_state_t *state)
{

}

static void on_tcp_rt_init(stm_machine_t *stm, const stm_state_t *state)
{

}
static unsigned int on_tcp_rt_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	if (e->type == EVT_MSG_TCP_CONN_RECV_RESULT) {
		int rc = tcp_conn_handle_rx_result(stm, e);
		if (rc) {
			stm_event_t *te = stm_event_create_base(EVT_MSG_TCP_CONN_LATE_CLOSE, stm->stm_id, stm->stm_id, 0, 0, 0, 0);
			stm_machine_process_at_dispatch_level(stm, conn_ctx->connected_level, te);
			stm_event_dec_ref(te);
			shutdown(conn_ctx->fd, SD_RECEIVE);
			CancelIo((HANDLE)conn_ctx->fd);
			return 1; // goto T
		} else {
			if (conn_ctx->pending_rx == 0)
				return 1; // goto T
		}
	} else if (e->type == EVT_MSG_TCP_CONN_SEND_RESULT) {
		int rc = tcp_conn_handle_tx_result(stm, e);
		if (rc) {
			stm_event_t *te = stm_event_create_base(EVT_MSG_TCP_CONN_LATE_CLOSE, stm->stm_id, stm->stm_id, 0, 0, 0, 0);
			stm_machine_process_at_dispatch_level(stm, conn_ctx->connected_level, te);
			stm_event_dec_ref(te);
			shutdown(conn_ctx->fd, SD_SEND);
			CancelIo((HANDLE)conn_ctx->fd);
			return 1; // goto R
		} else {
			if (conn_ctx->pending_tx == 0)
				return 1; // goto R
		}
	}
	return 0;
}
static void on_tcp_rt_exit(stm_machine_t *stm, const stm_state_t *state)
{
}

static void on_tcp_doing_init(stm_machine_t *stm, const stm_state_t *state)
{
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	conn_ctx->connected_level = stm_machine_get_current_dispatch_level(stm);
	stm_machine_inc_dispatch_level(stm);
}
static unsigned int on_tcp_doing_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	if (e->type == STM_TCP_CONN_EVT_MSG_CLOSE) {
		stm_event_t *te = stm_event_create_base(EVT_MSG_TCP_CONN_LATE_CLOSE, stm->stm_id, stm->stm_id, 0, 0, 0, 0);
		CancelIo((HANDLE)conn_ctx->fd);
		stm_machine_process_at_dispatch_level(stm, conn_ctx->connected_level, te);
		stm_event_dec_ref(te);
	}
	return 0;
}
static void on_tcp_doing_exit(stm_machine_t *stm, const stm_state_t *state)
{
	stm_machine_dec_dispatch_level(stm);
}

static unsigned int TCP_CONNECTION_on_init(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	stm_tcp_conn_initial_arg_t *conn_arg = (stm_tcp_conn_initial_arg_t *)stm_event_get_context(e);
	int rc = 0;

	conn_ctx->cache_rx_buf = NULL;
	conn_ctx->fd = INVALID_SOCKET;
	conn_ctx->arr_server = NULL;
	conn_ctx->rx.type = IOCP_OP_NONE;
	conn_ctx->tx.type = IOCP_OP_NONE;
	conn_ctx->connected_level = -1;
	conn_ctx->acceptor_stm_id = 0;
	conn_ctx->pending_rx = 0;
	conn_ctx->pending_tx = 0;
	conn_ctx->connect_timer_id = 0;
	conn_ctx->connect_timeout = 0;

	stm_machine_defer_process_high_priority(stm, e);

	return 0;
}
static void TCP_CONNECTION_on_exit(stm_machine_t *stm)
{
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);

	if (conn_ctx->acceptor_stm_id) {
		stm_machine_t *acceptor_stm = stm_machine_id2ptr(conn_ctx->acceptor_stm_id);
		stm_event_t *te = stm_event_create_base(EVT_MSG_TCP_AP_CONNECTION_EXIT, stm->stm_id, conn_ctx->acceptor_stm_id, 0, 0, 0, 0);
		stm_dispatcher_post_event(acceptor_stm->host_dispatcher, te);
		stm_event_dec_ref(te);
	}
}
static void TCP_CONNECTION_on_destroy(stm_machine_t *stm)
{
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	
	iocp_op_clear(&conn_ctx->rx);
	iocp_op_clear(&conn_ctx->tx);
	if (conn_ctx->fd != INVALID_SOCKET) {
		closesocket(conn_ctx->fd);
		conn_ctx->fd = INVALID_SOCKET;
	}
	if (conn_ctx->cache_rx_buf) {
		free(conn_ctx->cache_rx_buf);
		conn_ctx->cache_rx_buf = NULL;
	}
	if (conn_ctx->arr_server) {
		array_free(conn_ctx->arr_server);
		conn_ctx->arr_server = NULL;
	}

	printf("tcp destroy! 0x%08x\n", conn_ctx);
}
static void TCP_CONNECTION_on_state_trans(stm_machine_t *stm, const stm_state_t *src_st, const stm_state_t *dst_st)
{
	tcp_conn_ctx_t *conn_ctx = (tcp_conn_ctx_t *)stm_machine_get_context(stm);
	//printf("tcp connection (src,dst) : %s, %s 0x%08x\n", src_st->name, dst_st->name, conn_ctx);
	//.....
}

BEGIN_STM_STATE(TCP_CONNECTION)
	STM_STATE_ENTRY(STM_TCP_CONN_START, "START", STM_STATE_ROOT, on_tcp_start_init, on_tcp_start_event, on_tcp_start_exit)
	STM_STATE_ENTRY(STM_TCP_CONN_CONNECTED, "CONNECTED", STM_STATE_ROOT, on_tcp_connected_init, on_tcp_connected_event, on_tcp_connected_exit)
	STM_STATE_ENTRY(STM_TCP_CONN_DOING, "DOING", STM_STATE_ROOT, on_tcp_doing_init, on_tcp_doing_event, on_tcp_doing_exit)
	STM_STATE_ENTRY(STM_TCP_CONN_CONNECTING, "CONNECTING", STM_TCP_CONN_DOING, on_tcp_connecting_init, on_tcp_connecting_event, on_tcp_connecting_exit)
	STM_STATE_ENTRY(STM_TCP_CONN_R, "R", STM_TCP_CONN_DOING, on_tcp_r_init, on_tcp_r_event, on_tcp_r_exit)
	STM_STATE_ENTRY(STM_TCP_CONN_T, "T", STM_TCP_CONN_DOING, on_tcp_t_init, on_tcp_t_event, on_tcp_t_exit)
	STM_STATE_ENTRY(STM_TCP_CONN_RT, "RT", STM_TCP_CONN_DOING, on_tcp_rt_init, on_tcp_rt_event, on_tcp_rt_exit)
END_STM_STATE()

BEGIN_STM_STATE_RULE(TCP_CONNECTION)
	STM_STATE_RULE(STM_STATE_INIT, STM_TCP_CONN_EVT_MSG_INITIAL_ARG, STM_CMD_JMP_IF(STM_TCP_CONN_START, 0))
	STM_STATE_RULE(STM_TCP_CONN_START, STM_TCP_CONN_EVT_MSG_INITIAL_ARG, STM_CMD_JMP_IF(STM_TCP_CONN_CONNECTED, 0))
	STM_STATE_RULE(STM_TCP_CONN_START, STM_TCP_CONN_EVT_MSG_INITIAL_ARG, STM_CMD_JMP_IF(STM_TCP_CONN_CONNECTING, 1))
	STM_STATE_RULE(STM_TCP_CONN_CONNECTING, EVT_MSG_TCP_CONN_CONNECT_RESULT, STM_CMD_NO_JMP_IF(0))
	STM_STATE_RULE(STM_TCP_CONN_CONNECTING, STM_SYS_EVT_MSG_TIMER, STM_CMD_NO_JMP_IF(0))
	STM_STATE_RULE(STM_TCP_CONN_CONNECTING, EVT_MSG_TCP_CONN_CONNECT_RESULT, STM_CMD_JMP_IF(STM_TCP_CONN_CONNECTED, 1))
	STM_STATE_RULE(STM_TCP_CONN_CONNECTING, EVT_MSG_TCP_CONN_CONNECT_RESULT, STM_CMD_JMP_IF(STM_STATE_EXIT, 2))
	STM_STATE_RULE(STM_TCP_CONN_DOING, STM_TCP_CONN_EVT_MSG_CLOSE, STM_CMD_NO_JMP_IF(0))
	STM_STATE_RULE(STM_TCP_CONN_CONNECTED, STM_TCP_CONN_EVT_REQ_RX, STM_CMD_JMP_IF(STM_TCP_CONN_R, 1))
	STM_STATE_RULE(STM_TCP_CONN_CONNECTED, STM_TCP_CONN_EVT_REQ_RX, STM_CMD_JMP_IF(STM_STATE_EXIT, 2))
	STM_STATE_RULE(STM_TCP_CONN_CONNECTED, STM_TCP_CONN_EVT_REQ_TX, STM_CMD_JMP_IF(STM_TCP_CONN_T, 1))
	STM_STATE_RULE(STM_TCP_CONN_CONNECTED, STM_TCP_CONN_EVT_REQ_TX, STM_CMD_JMP_IF(STM_STATE_EXIT, 2))
	STM_STATE_RULE(STM_TCP_CONN_CONNECTED, EVT_MSG_TCP_CONN_LATE_CLOSE, STM_CMD_JMP_IF(STM_STATE_EXIT, 1))
	STM_STATE_RULE(STM_TCP_CONN_CONNECTED, STM_TCP_CONN_EVT_MSG_CLOSE, STM_CMD_JMP_IF(STM_STATE_EXIT, 1))
	STM_STATE_RULE(STM_TCP_CONN_CONNECTED, STM_TCP_CONN_EVT_MSG_DETECT_BROKEN, STM_CMD_JMP_IF(STM_STATE_EXIT, 1))
	STM_STATE_RULE(STM_TCP_CONN_R, EVT_MSG_TCP_CONN_RECV_RESULT, STM_CMD_JMP_IF(STM_TCP_CONN_CONNECTED, 1))
	STM_STATE_RULE(STM_TCP_CONN_R, STM_TCP_CONN_EVT_REQ_TX, STM_CMD_JMP_IF(STM_TCP_CONN_RT, 1))
	STM_STATE_RULE(STM_TCP_CONN_R, STM_TCP_CONN_EVT_REQ_TX, STM_CMD_JMP_IF(STM_TCP_CONN_CONNECTED, 2))
	STM_STATE_RULE(STM_TCP_CONN_T, EVT_MSG_TCP_CONN_SEND_RESULT, STM_CMD_JMP_IF(STM_TCP_CONN_CONNECTED, 1))
	STM_STATE_RULE(STM_TCP_CONN_T, STM_TCP_CONN_EVT_REQ_RX, STM_CMD_JMP_IF(STM_TCP_CONN_CONNECTED, 1))
	STM_STATE_RULE(STM_TCP_CONN_T, STM_TCP_CONN_EVT_REQ_RX, STM_CMD_JMP_IF(STM_TCP_CONN_RT, 2))
	STM_STATE_RULE(STM_TCP_CONN_RT, EVT_MSG_TCP_CONN_RECV_RESULT, STM_CMD_JMP_IF(STM_TCP_CONN_T, 1))
	STM_STATE_RULE(STM_TCP_CONN_RT, EVT_MSG_TCP_CONN_SEND_RESULT, STM_CMD_JMP_IF(STM_TCP_CONN_R, 1))
END_STM_STATE_RULE()

DEFINE_STM_MACHINE(TCP_CONNECTION, STM_CAT_SYSTEM, STM_CLS_ID_TCP_CONNECTION, sizeof(tcp_conn_ctx_t), TCP_CONNECTION_on_init, TCP_CONNECTION_on_exit, TCP_CONNECTION_on_destroy, TCP_CONNECTION_on_state_trans)

// acceptor state: start, run, 

typedef struct tcp_ap_ctx_t
{
	SOCKET fd;
	array_header_t *arr_accept;
	int actual_concurrent;
	LONG *active_connection;
	LONG max_connection;
	int pause_timer_id;
}tcp_ap_ctx_t;

static int init_ap_ctx(stm_machine_t *stm, unsigned long ip, unsigned short port)
{
	tcp_ap_ctx_t *ap_ctx = (tcp_ap_ctx_t *)stm_machine_get_context(stm);
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	int rc = 0;
	ap_ctx->fd = WSASocketA(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
	if (ap_ctx->fd != INVALID_SOCKET) {
		if (CreateIoCompletionPort((HANDLE)ap_ctx->fd, iocp_disp->iocp_handle, (ULONG_PTR)stm, 0)) {
			struct sockaddr_in addr = {0};
			addr.sin_family = AF_INET;
			addr.sin_port = htons(port);
			addr.sin_addr.s_addr = ip;
			rc = bind(ap_ctx->fd, (struct sockaddr*)&addr, sizeof(addr));
			if (rc == 0) {
				rc = listen(ap_ctx->fd, 10);
			}
		} else {
			rc = -1;
		}
		if (rc != 0) {
			closesocket(ap_ctx->fd);
			ap_ctx->fd = INVALID_SOCKET;
		}
	} else {
		rc = -1;
	}
	return rc;
}

static void term_ap_ctx(stm_machine_t *stm)
{
	tcp_ap_ctx_t *ap_ctx = (tcp_ap_ctx_t *)stm_machine_get_context(stm);
	if (ap_ctx->fd != INVALID_SOCKET) {
		closesocket(ap_ctx->fd);
		ap_ctx->fd = INVALID_SOCKET;
	}
}


static int start_accept(stm_machine_t *stm, int id)
{
	tcp_ap_ctx_t *ap_ctx = (tcp_ap_ctx_t *)stm_machine_get_context(stm);
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	iocp_op_t *op = &ARRAY_IDX(ap_ctx->arr_accept, id, iocp_op_t);
	int rc = -1;
	SOCKET fd;

	fd = WSASocketA(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
	if (fd != INVALID_SOCKET) {
		DWORD dwBytesTransfer;
		BOOL reuseaddr = 1;
		setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char*)&reuseaddr, sizeof(reuseaddr));
		iocp_op_prepare_for_ap(op, fd, id);
		if (AcceptEx(ap_ctx->fd, fd, op->ap.accept_buf, 0, ACCEPT_ADDR_LEN, ACCEPT_ADDR_LEN, &dwBytesTransfer, &op->ov)) {
			stm_tcp_ap_connection *ap_conn;
			stm_event_t *te;
			SOCKADDR *lpLocalSockaddr=NULL,
				*lpRemoteSockaddr=NULL;
			int LocalSockaddrLen=0,
				RemoteSockaddrLen=0;
			rc = 0;
			InterlockedIncrement(ap_ctx->active_connection);
			GetAcceptExSockaddrs(op->ap.accept_buf, 0, ACCEPT_ADDR_LEN, ACCEPT_ADDR_LEN, &lpLocalSockaddr, &LocalSockaddrLen, &lpRemoteSockaddr, &RemoteSockaddrLen);
			te = stm_event_create_base(STM_TCP_AP_EVT_MSG_CONN, stm->stm_id, stm->report_stm_id, 0, 0, sizeof(stm_tcp_ap_connection), (void**)&ap_conn);
			ap_conn->socket = fd;
			memcpy(&ap_conn->local_addr, lpLocalSockaddr, sizeof(SOCKADDR));
			memcpy(&ap_conn->remote_addr, lpRemoteSockaddr, sizeof(SOCKADDR));
			op->ap.fd = INVALID_SOCKET;
			setsockopt(fd, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, (char*)&ap_ctx->fd, sizeof(SOCKET));
			stm_dispatcher_post_event(stm->report_dispatcher, te);
			stm_event_dec_ref(te);
		} else {
			if (WSAGetLastError() != WSA_IO_PENDING) {
				iocp_op_clear(op);
			}
		}
	}

	return rc;
}

static int start_accept_all(stm_machine_t *stm, int concurrent)
{
	tcp_ap_ctx_t *ap_ctx = (tcp_ap_ctx_t *)stm_machine_get_context(stm);
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	int rc = 0;
	int i;
	int error = 0;
	int clients = 0;

	if (concurrent != -1) {
		ap_ctx->arr_accept = array_make(concurrent, sizeof(iocp_op_t));
	} else {
		concurrent = ap_ctx->arr_accept->nelts;
	}

	for (i = 0; i < concurrent; ++i) {
		iocp_op_t *op = &ARRAY_IDX(ap_ctx->arr_accept, i, iocp_op_t);
		op->type = IOCP_OP_NONE;
		while (*ap_ctx->active_connection < ap_ctx->max_connection && start_accept(stm, i) == 0) {
			clients++;
		}
		if (WSAGetLastError() == WSA_IO_PENDING) {
			ap_ctx->actual_concurrent ++;
		} else {
			error ++;
		}
	}

	if (clients == 0 && error == concurrent)
		rc = -1;

	return rc;
}

static void on_tcp_ap_start_init(stm_machine_t *stm, const stm_state_t *state)
{

}
static unsigned int on_tcp_ap_start_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	tcp_ap_ctx_t *ap_ctx = (tcp_ap_ctx_t *)stm_machine_get_context(stm);
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);

	if (e->type == STM_TCP_AP_EVT_MSG_INITIAL_ARG) {
		stm_tcp_ap_initial_arg_t *initial_arg = (stm_tcp_ap_initial_arg_t *)stm_event_get_context(e);
		int rc;
		ap_ctx->active_connection = initial_arg->active_connection;
		ap_ctx->max_connection = initial_arg->max_connection;
		rc = init_ap_ctx(stm, initial_arg->server_ip, initial_arg->server_port);
		if (rc == 0) {
			int concurrent = initial_arg->concurrent_count ? initial_arg->concurrent_count : 5;
			concurrent = min(concurrent, ap_ctx->max_connection);
			rc = start_accept_all(stm, initial_arg->concurrent_count);
			if (rc == 0) {
				if (ap_ctx->actual_concurrent == 0)
					return 1; // goto Pause
			}
		}
		if (rc != 0) {
			stm_machine_goto_error(stm, rc);
		}
	} else if (e->type == EVT_MSG_TCP_AP_ACCEPT_RESULT) {
		int error = e->param1;
		int id = e->param2;
		iocp_op_t *op = &ARRAY_IDX(ap_ctx->arr_accept, id, iocp_op_t);
		ap_ctx->actual_concurrent--;
		if (error == 0) {
			int rc = 0;
			stm_tcp_ap_connection *ap_conn;
			stm_event_t *te;
			SOCKADDR *lpLocalSockaddr=NULL,
				*lpRemoteSockaddr=NULL;
			int LocalSockaddrLen=0,
				RemoteSockaddrLen=0;
			InterlockedIncrement(ap_ctx->active_connection);
			setsockopt(op->ap.fd, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, (char*)&ap_ctx->fd, sizeof(SOCKET));
			GetAcceptExSockaddrs(op->ap.accept_buf, 0, ACCEPT_ADDR_LEN, ACCEPT_ADDR_LEN, &lpLocalSockaddr, &LocalSockaddrLen, &lpRemoteSockaddr, &RemoteSockaddrLen);
			te = stm_event_create_base(STM_TCP_AP_EVT_MSG_CONN, stm->stm_id, stm->report_stm_id, (int)op->ap.fd, 0, sizeof(stm_tcp_ap_connection), (void**)&ap_conn);
			ap_conn->socket = op->ap.fd;
			memcpy(&ap_conn->local_addr, lpLocalSockaddr, sizeof(SOCKADDR));
			memcpy(&ap_conn->remote_addr, lpRemoteSockaddr, sizeof(SOCKADDR));
			op->ap.fd = INVALID_SOCKET;
			stm_dispatcher_post_event(stm->report_dispatcher, te);
			stm_event_dec_ref(te);
			while (*ap_ctx->active_connection < ap_ctx->max_connection && (rc = start_accept(stm, id)) == 0) {
				;
			}
			if (rc != 0) {
				if (WSAGetLastError() == WSA_IO_PENDING) {
					ap_ctx->actual_concurrent++;
					if (ap_ctx->actual_concurrent < ap_ctx->arr_accept->nelts) {
						int i;
						for (i = 0; i < ap_ctx->arr_accept->nelts; ++i) {
							iocp_op_t *op = &ARRAY_IDX(ap_ctx->arr_accept, i, iocp_op_t);
							if (op->type == IOCP_OP_ACCEPT && op->ap.fd == INVALID_SOCKET) {
								int rc = 0;
								while (*ap_ctx->active_connection < ap_ctx->max_connection && (rc = start_accept(stm, i)) == 0) {
									;
								}
								if (rc == -1 && WSAGetLastError() == WSA_IO_PENDING)
									ap_ctx->actual_concurrent++;
							}
						}
					}
				}
			}
		} else {
			iocp_op_clear(op);
		}
		if (ap_ctx->actual_concurrent == 0) 
			return 1; // goto Pause
	} else if (e->type == STM_TCP_AP_EVT_MSG_CLOSE) {
		CancelIo((HANDLE)ap_ctx->fd);
		return 1; // goto Stopping
	} else if (e->type == EVT_MSG_TCP_AP_CONNECTION_EXIT) {
		InterlockedDecrement(ap_ctx->active_connection);
	} else {

	}

	return 0;
}
static void on_tcp_ap_start_exit(stm_machine_t *stm, const stm_state_t *state)
{

}

static void on_tcp_ap_pause_init(stm_machine_t *stm, const stm_state_t *state)
{
	tcp_ap_ctx_t *ap_ctx = (tcp_ap_ctx_t *)stm_machine_get_context(stm);
	ap_ctx->pause_timer_id = stm_dispatcher_schedule_timer(stm->host_dispatcher, stm->stm_id, 500);
}
static unsigned int on_tcp_ap_pause_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	tcp_ap_ctx_t *ap_ctx = (tcp_ap_ctx_t *)stm_machine_get_context(stm);

	if (e->type == STM_SYS_EVT_MSG_TIMER) {
		int rc = start_accept_all(stm, -1);
		ap_ctx->pause_timer_id = 0;
		if (rc == -1) {
			stm_machine_goto_error(stm, rc);
		} else {
			if (ap_ctx->actual_concurrent > 0) {
				return 1; // goto Start
			} else {
				ap_ctx->pause_timer_id = stm_dispatcher_schedule_timer(stm->host_dispatcher, stm->stm_id, 500);
			}
		}
	} else if (e->type == EVT_MSG_TCP_AP_CONNECTION_EXIT) {
		InterlockedDecrement(ap_ctx->active_connection);
		if (ap_ctx->actual_concurrent < ap_ctx->arr_accept->nelts) {
			int i;
			for (i = 0; i < ap_ctx->arr_accept->nelts; ++i) {
				iocp_op_t *op = &ARRAY_IDX(ap_ctx->arr_accept, i, iocp_op_t);
				if (op->type == IOCP_OP_ACCEPT && op->ap.fd == INVALID_SOCKET) {
					int rc = 0;
					while (*ap_ctx->active_connection < ap_ctx->max_connection && (rc = start_accept(stm, i)) == 0) {
						;
					}
					if (rc == -1 && WSAGetLastError() == WSA_IO_PENDING)
						ap_ctx->actual_concurrent++;
				}
			}
		}
		if (ap_ctx->actual_concurrent != 0)
			return 1; // goto Start
	}

	return 0;
}
static void on_tcp_ap_pause_exit(stm_machine_t *stm, const stm_state_t *state)
{
	tcp_ap_ctx_t *ap_ctx = (tcp_ap_ctx_t *)stm_machine_get_context(stm);
	if (ap_ctx->pause_timer_id != 0) {
		stm_dispatcher_cancel_timer(stm->host_dispatcher, ap_ctx->pause_timer_id);
		ap_ctx->pause_timer_id = 0;
	}
}

static void on_tcp_ap_stopping_init(stm_machine_t *stm, const stm_state_t *state)
{
}
static unsigned int on_tcp_ap_stopping_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	tcp_ap_ctx_t *ap_ctx = (tcp_ap_ctx_t *)stm_machine_get_context(stm);
	if (e->type == EVT_MSG_TCP_AP_ACCEPT_RESULT) {
		int error = e->param1;
		int id = e->param2;
		iocp_op_t *op = &ARRAY_IDX(ap_ctx->arr_accept, id, iocp_op_t);
		ap_ctx->actual_concurrent--;
		if (error == 0) {
			stm_tcp_ap_connection *ap_conn;
			stm_event_t *te;
			SOCKADDR *lpLocalSockaddr=NULL,
				*lpRemoteSockaddr=NULL;
			int LocalSockaddrLen=0,
				RemoteSockaddrLen=0;
			InterlockedIncrement(ap_ctx->active_connection);
			setsockopt(op->ap.fd, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, (char*)&ap_ctx->fd, sizeof(SOCKET));
			GetAcceptExSockaddrs(op->ap.accept_buf, 0, ACCEPT_ADDR_LEN, ACCEPT_ADDR_LEN, &lpLocalSockaddr, &LocalSockaddrLen, &lpRemoteSockaddr, &RemoteSockaddrLen);
			te = stm_event_create_base(STM_TCP_AP_EVT_MSG_CONN, stm->stm_id, stm->report_stm_id, (int)op->ap.fd, 0, sizeof(stm_tcp_ap_connection), (void**)&ap_conn);
			ap_conn->socket = op->ap.fd;
			memcpy(&ap_conn->local_addr, lpLocalSockaddr, sizeof(SOCKADDR));
			memcpy(&ap_conn->remote_addr, lpRemoteSockaddr, sizeof(SOCKADDR));
			op->ap.fd = INVALID_SOCKET;
			stm_dispatcher_post_event(stm->report_dispatcher, te);
			stm_event_dec_ref(te);
		} else {
			iocp_op_clear(op);
		}
		if (ap_ctx->actual_concurrent == 0)
			return 1;
	} else if (e->type == EVT_MSG_TCP_AP_CONNECTION_EXIT) {
		InterlockedDecrement(ap_ctx->active_connection);
	}
	return 0;
}
static void on_tcp_ap_stopping_exit(stm_machine_t *stm, const stm_state_t *state)
{

}

static unsigned int TCP_ACCEPTOR_on_init(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	tcp_ap_ctx_t *ap_ctx = (tcp_ap_ctx_t *)stm_machine_get_context(stm);

	ap_ctx->fd = INVALID_SOCKET;
	ap_ctx->arr_accept = NULL;
	ap_ctx->actual_concurrent = 0;
	ap_ctx->active_connection = NULL;
	ap_ctx->max_connection = 0;
	ap_ctx->pause_timer_id = 0;

	stm_machine_defer_process_high_priority(stm, e);

	return 0;
}
static void TCP_ACCEPTOR_on_exit(stm_machine_t *stm)
{

}
static void TCP_ACCEPTOR_on_destroy(stm_machine_t *stm)
{
	tcp_ap_ctx_t *ap_ctx = (tcp_ap_ctx_t *)stm_machine_get_context(stm);

	if (ap_ctx->fd != INVALID_SOCKET) {
		closesocket(ap_ctx->fd);
		ap_ctx->fd = INVALID_SOCKET;
	}
	if (ap_ctx->arr_accept) {
		int i;
		for (i = 0; i < ap_ctx->arr_accept->nelts; ++i) {
			iocp_op_t *op = &ARRAY_IDX(ap_ctx->arr_accept, i, iocp_op_t);
			iocp_op_clear(op);
		}
		array_free(ap_ctx->arr_accept);
		ap_ctx->arr_accept = NULL;
	}

	printf("tcp ap destroy!\n");
}
static void TCP_ACCEPTOR_on_state_trans(stm_machine_t *stm, const stm_state_t *src_state, const stm_state_t *dst_state)
{
	printf("acceptor (src,dst) : %s, %s\n", src_state->name, dst_state->name);
}

BEGIN_STM_STATE(TCP_ACCEPTOR)
	STM_STATE_ENTRY(STM_TCP_AP_START, "START", STM_STATE_ROOT, on_tcp_ap_start_init, on_tcp_ap_start_event, on_tcp_ap_start_exit)
	STM_STATE_ENTRY(STM_TCP_AP_PAUSE, "PAUSE", STM_STATE_ROOT, on_tcp_ap_pause_init, on_tcp_ap_pause_event, on_tcp_ap_pause_exit)
	STM_STATE_ENTRY(STM_TCP_AP_STOPPING, "STOPPING", STM_STATE_ROOT, on_tcp_ap_stopping_init, on_tcp_ap_stopping_event, on_tcp_ap_stopping_exit)
END_STM_STATE()

BEGIN_STM_STATE_RULE(TCP_ACCEPTOR)
	STM_STATE_RULE(STM_STATE_INIT, STM_TCP_AP_EVT_MSG_INITIAL_ARG, STM_CMD_JMP_IF(STM_TCP_AP_START, 0))
	STM_STATE_RULE(STM_TCP_AP_START, STM_TCP_AP_EVT_MSG_INITIAL_ARG, STM_CMD_JMP_IF(STM_TCP_AP_PAUSE, 1))
	STM_STATE_RULE(STM_TCP_AP_START, EVT_MSG_TCP_AP_ACCEPT_RESULT, STM_CMD_JMP_IF(STM_TCP_AP_PAUSE, 1))
	STM_STATE_RULE(STM_TCP_AP_START, EVT_MSG_TCP_AP_CONNECTION_EXIT, STM_CMD_NO_JMP_IF(0))
	STM_STATE_RULE(STM_TCP_AP_START, STM_TCP_AP_EVT_MSG_CLOSE, STM_CMD_JMP_IF(STM_TCP_AP_STOPPING, 1))
	STM_STATE_RULE(STM_TCP_AP_PAUSE, STM_TCP_AP_EVT_MSG_CLOSE, STM_CMD_JMP_IF(STM_STATE_EXIT, 0))
	STM_STATE_RULE(STM_TCP_AP_PAUSE, STM_SYS_EVT_MSG_TIMER, STM_CMD_JMP_IF(STM_TCP_AP_START, 1))
	STM_STATE_RULE(STM_TCP_AP_PAUSE, EVT_MSG_TCP_AP_CONNECTION_EXIT, STM_CMD_JMP_IF(STM_TCP_AP_START, 1))
	STM_STATE_RULE(STM_TCP_AP_STOPPING, EVT_MSG_TCP_AP_ACCEPT_RESULT, STM_CMD_JMP_IF(STM_STATE_EXIT, 1))
END_STM_STATE_RULE()

DEFINE_STM_MACHINE(TCP_ACCEPTOR, STM_CAT_SYSTEM, STM_CLS_ID_TCP_ACCEPTOR, sizeof(tcp_ap_ctx_t), TCP_ACCEPTOR_on_init, TCP_ACCEPTOR_on_exit, TCP_ACCEPTOR_on_destroy, TCP_ACCEPTOR_on_state_trans)

typedef struct udp_conn_ctx_t
{
	SOCKET fd;
	struct {
		iocp_op_t ops[MAX_UDP_CONCURENT];
		int free_ops[MAX_UDP_CONCURENT];
		int free_ops_sp;
		int concurrent;
	}tx;
	struct {
		iocp_op_t ops[MAX_UDP_CONCURENT];
		int free_ops[MAX_UDP_CONCURENT];
		int free_ops_sp;
		int concurrent;
	}rx;
	int ready_level;
}udp_conn_ctx_t;

static int init_udp_ctx_fd(stm_machine_t *stm, int fd)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	conn_ctx->fd = fd;
	if (CreateIoCompletionPort((HANDLE)fd, iocp_disp->iocp_handle, (ULONG_PTR)stm, 0)) {
		return 0;
	}
	return -1;
}

static int init_udp_ctx(stm_machine_t *stm, unsigned long ip, unsigned short port)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	iocp_dispatcher_t *iocp_disp = (iocp_dispatcher_t *)stm_dispatcher_get_context(stm->host_dispatcher);
	int fd;
	int rc = -1;

	fd = WSASocketA(AF_INET, SOCK_DGRAM, IPPROTO_UDP, 0, 0, WSA_FLAG_OVERLAPPED);
	if (fd != INVALID_SOCKET) {
		struct sockaddr_in addr = {0};
		addr.sin_family = AF_INET;
		addr.sin_addr.s_addr = ip;
		addr.sin_port = htons(port);
		if (bind(fd, (struct sockaddr*)&addr, sizeof(addr)) == 0) {
			if (CreateIoCompletionPort((HANDLE)fd, iocp_disp->iocp_handle, (ULONG_PTR)stm, 0)) {
				rc = 0;
			}
		}
	}
	
	if (rc == -1) {
		if (fd != INVALID_SOCKET) {
			closesocket(fd);
		}
	} else {
		conn_ctx->fd = fd;
	}

	return rc;
}

static int udp_conn_handle_rx_cmd(stm_machine_t *stm, stm_event_t *e)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	int rc = 0;
	int max_pkt_size = e->param1;
	int user_data = e->param2;
	if (conn_ctx->rx.free_ops_sp) {
		int id = conn_ctx->rx.free_ops[--conn_ctx->rx.free_ops_sp];
		iocp_op_t *op = &conn_ctx->rx.ops[id];
		DWORD dwTransferBytes = 0;
		iocp_op_prepare_for_udp_rx(op, id, max_pkt_size, user_data);
		op->udp_rx.wsabuf.buf = op->udp_rx.buf;
		op->udp_rx.wsabuf.len = max_pkt_size;
		op->udp_rx.dwFlags = 0;
		//printf("iocp_op_prepare_for_udp_rx id=%d, max_size=%d\n", id, max_pkt_size); 
		rc = WSARecvFrom(conn_ctx->fd, &op->udp_rx.wsabuf, 1, &dwTransferBytes, &op->udp_rx.dwFlags, (struct sockaddr*)&op->udp_rx.src_addr, &op->udp_rx.from_len, &op->ov, NULL);
		if (rc == SOCKET_ERROR && WSAGetLastError() != WSA_IO_PENDING) {
			stm_event_t *te = stm_event_create_base(STM_UDP_CONN_EVT_RES_RX, stm->stm_id, stm->report_stm_id, rc, user_data, 0, 0);
			stm_dispatcher_post_event(stm->report_dispatcher, te);
			stm_event_dec_ref(te);
			conn_ctx->rx.free_ops[conn_ctx->rx.free_ops_sp++] = id;
		} else {
			conn_ctx->rx.concurrent++;
		}
	} else {
		stm_event_t *te = stm_event_create_base(STM_UDP_CONN_EVT_RES_RX, stm->stm_id, stm->report_stm_id, rc = -1, user_data, 0, 0);
		stm_dispatcher_post_event(stm->report_dispatcher, te);
		stm_event_dec_ref(te);
		//printf("not enough free ops, iocp_op_prepare_for_udp_rx\n");
	}
	return rc;
}

static int udp_conn_handle_tx_cmd(stm_machine_t *stm, stm_event_t *e)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	stm_udp_conn_tx_t *arg = (stm_udp_conn_tx_t *)stm_event_get_context(e);
	int rc = 0;
	int user_data = e->param2;
	if (conn_ctx->tx.free_ops_sp) {
		int id = conn_ctx->tx.free_ops[--conn_ctx->tx.free_ops_sp];
		iocp_op_t *op = &conn_ctx->tx.ops[id];
		DWORD dwBytesTransfer = 0;
		struct sockaddr_in addr = {0};
		//printf("iocp_op_prepare_for_udp_tx id=%d, buf=%08x, offset=%d,size=%d, dst_port=%d\n", id, arg->buf, arg->offset, arg->size, arg->dst_port);
		iocp_op_prepare_for_udp_tx(op, id, arg->buf, arg->offset, arg->size, user_data);
		addr.sin_family = AF_INET;
		addr.sin_port = htons(arg->dst_port);
		addr.sin_addr.s_addr = arg->dst_ip;
		op->udp_tx.wsabuf.buf = op->udp_tx.buf;
		op->udp_tx.wsabuf.len = op->udp_tx.size;
		//printf("send to %s:%d\n", inet_ntoa(addr.sin_addr), arg->dst_port);
		rc = WSASendTo(conn_ctx->fd, &op->udp_tx.wsabuf, 1, &dwBytesTransfer, 0, (struct sockaddr*)&addr, sizeof(addr), &op->ov, NULL);
		if (rc == SOCKET_ERROR && WSAGetLastError() != WSA_IO_PENDING) {
			stm_event_t *te = stm_event_create_base(STM_UDP_CONN_EVT_RES_TX, stm->stm_id, stm->report_stm_id, rc, user_data, 0, 0);
			stm_dispatcher_post_event(stm->report_dispatcher, te);
			stm_event_dec_ref(te);
			conn_ctx->tx.free_ops[conn_ctx->tx.free_ops_sp++] = id;
		} else {
			conn_ctx->tx.concurrent++;
		}
	} else {
		stm_event_t *te = stm_event_create_base(STM_UDP_CONN_EVT_RES_TX, stm->stm_id, stm->report_stm_id, rc = -1, user_data, 0, 0);
		stm_dispatcher_post_event(stm->report_dispatcher, te);
		stm_event_dec_ref(te);
		//printf("not enough free ops, iocp_op_prepare_for_udp_tx\n");
	}
	return rc;
}

static int udp_conn_handle_rx_result(stm_machine_t *stm, stm_event_t *e)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	int error = e->param1;
	int id = e->param2;
	iocp_op_t *op = &conn_ctx->rx.ops[id];
	if (error == 0) {
		int size = op->udp_rx.offset;
		stm_udp_conn_rx_msg_t *rx_msg;
		int ctx_len = sizeof(stm_udp_conn_rx_msg_t) + size;
		stm_event_t *te = stm_event_create_base(STM_UDP_CONN_EVT_RES_RX, stm->stm_id, stm->report_stm_id, error, op->udp_rx.user_data, ctx_len, (void**)&rx_msg);
		rx_msg->buf = (char*)(rx_msg+1);
		rx_msg->offset = 0;
		rx_msg->size = size;
		memcpy(rx_msg->buf, op->udp_rx.buf, size);
		memcpy(&rx_msg->src_addr, &op->udp_rx.src_addr, sizeof(SOCKADDR));
		stm_dispatcher_post_event(stm->report_dispatcher, te);
		stm_event_dec_ref(te);
		//printf("udp rx, id=%d,size=%d, %s:%d\n", id, size, inet_ntoa(op->udp_rx.src_addr.sin_addr), ntohs(op->udp_rx.src_addr.sin_port));
	} else {
		stm_event_t *te = stm_event_create_base(STM_UDP_CONN_EVT_RES_RX, stm->stm_id, stm->report_stm_id, error, op->udp_rx.user_data, 0, 0);
		stm_dispatcher_post_event(stm->report_dispatcher, te);
		stm_event_dec_ref(te);
		//printf("udp rx, error = %d\n", WSAGetLastError());
	}
	conn_ctx->rx.concurrent--;
	conn_ctx->rx.free_ops[conn_ctx->rx.free_ops_sp++] = id;
	return 0;
}

static int udp_conn_handle_tx_result(stm_machine_t *stm, stm_event_t *e)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	int error = e->param1;
	int id = e->param2;
	iocp_op_t *op = &conn_ctx->tx.ops[id];
	stm_event_t *te = stm_event_create_base(STM_UDP_CONN_EVT_RES_TX, stm->stm_id, stm->report_stm_id, error, op->udp_tx.user_data, 0, 0);
	stm_dispatcher_post_event(stm->report_dispatcher, te);
	stm_event_dec_ref(te);
	//printf("udp tx, id=%d, transferBytes=%d\n", id, op->udp_tx.offset);
	conn_ctx->tx.concurrent--;
	conn_ctx->tx.free_ops[conn_ctx->tx.free_ops_sp++] = id;
	return 0;
}

static void on_udp_conn_start_init(stm_machine_t *stm, const stm_state_t *state)
{

}

static unsigned int on_udp_conn_start_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);

	if (e->type == STM_UDP_CONN_EVT_MSG_INITIAL_ARG) {
		stm_udp_conn_initial_arg_t *initial_arg = (stm_udp_conn_initial_arg_t *)stm_event_get_context(e);
		int rc;
		if (initial_arg->socket_fd == INVALID_SOCKET) {
			rc = init_udp_ctx(stm, initial_arg->local_ip, initial_arg->local_port);
		} else {
			rc = init_udp_ctx_fd(stm, initial_arg->socket_fd);
			if (rc != 0) {
				closesocket(initial_arg->socket_fd);
				initial_arg->socket_fd = INVALID_SOCKET;
			}
		}
		if (rc != 0) {
			stm_machine_goto_error(stm, rc);
		} else {
			DWORD dwBytesReturned = 0;
			BOOL bNewBehavior = FALSE;
			WSAIoctl(conn_ctx->fd, SIO_UDP_CONNRESET, &bNewBehavior, sizeof(bNewBehavior),
				NULL, 0, &dwBytesReturned, NULL, NULL);
			if (initial_arg->sock_rcv_buf_size) {
				setsockopt(conn_ctx->fd, SOL_SOCKET, SO_RCVBUF, (char*)&initial_arg->sock_rcv_buf_size, sizeof(int));
			}
			if (initial_arg->sock_snd_buf_size) {
				setsockopt(conn_ctx->fd, SOL_SOCKET, SO_SNDBUF, (char*)&initial_arg->sock_snd_buf_size, sizeof(int));
			}
			return 1; // goto ready
		}
	}
	return 0;
}
static void on_udp_conn_start_exit(stm_machine_t *stm, const stm_state_t *state)
{

}

static void on_udp_conn_ready_init(stm_machine_t *stm, const stm_state_t *state)
{

}
static unsigned int on_udp_conn_ready_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	if (e->type == STM_UDP_CONN_EVT_REQ_RX) {
		udp_conn_handle_rx_cmd(stm, e);
		if (conn_ctx->rx.concurrent != 0) {
			return 1; // goto R
		}
	} else if (e->type == STM_UDP_CONN_EVT_REQ_TX) {
		udp_conn_handle_tx_cmd(stm, e);
		if (conn_ctx->tx.concurrent != 0) {
			//printf("process req tx, io pending\n");
			return 1; // goto T
		} else {
			//printf("process req tx, fast\n");
		}
	} else if (e->type == EVT_MSG_UDP_CONN_RECV_RESULT) {
		assert(0);
	} else if (e->type == EVT_MSG_UDP_CONN_SEND_RESULT) {
		assert(0);
	}
	return 0;
}
static void on_udp_conn_ready_exit(stm_machine_t *stm, const stm_state_t *state)
{

}

static void on_udp_conn_doing_init(stm_machine_t *stm, const stm_state_t *state)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	conn_ctx->ready_level = stm_machine_get_current_dispatch_level(stm);
	stm_machine_inc_dispatch_level(stm);
}
static unsigned int on_udp_conn_doing_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	if (e->type == STM_UDP_CONN_EVT_MSG_CLOSE) {
		stm_event_t *te = stm_event_create_base(EVT_MSG_UDP_CONN_LATE_CLOSE, stm->stm_id, stm->stm_id, 0, 0, 0, 0);
		CancelIo((HANDLE)conn_ctx->fd);
		stm_machine_process_at_dispatch_level(stm, conn_ctx->ready_level, te);
		stm_event_dec_ref(te);
	}
	return 0;
}
static void on_udp_conn_doing_exit(stm_machine_t *stm, const stm_state_t *state)
{
	stm_machine_dec_dispatch_level(stm);
}

static void on_udp_conn_r_init(stm_machine_t *stm, const stm_state_t *state)
{

}
static unsigned int on_udp_conn_r_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	if (e->type == EVT_MSG_UDP_CONN_RECV_RESULT) {
		udp_conn_handle_rx_result(stm, e);
		if (conn_ctx->rx.concurrent == 0) {
			return 1; // goto Ready
		}
	} else if (e->type == STM_UDP_CONN_EVT_REQ_TX) {
		udp_conn_handle_tx_cmd(stm, e);
		if (conn_ctx->tx.concurrent != 0) {
			return 1; // goto RT
		}
	} else if (e->type == STM_UDP_CONN_EVT_REQ_RX) {
		udp_conn_handle_rx_cmd(stm, e);
	} else if (e->type == EVT_MSG_UDP_CONN_SEND_RESULT) {
		assert(0);
	}
	return 0; // no jump
}
static void on_udp_conn_r_exit(stm_machine_t *stm, const stm_state_t *state)
{

}

static void on_udp_conn_t_init(stm_machine_t *stm, const stm_state_t *state)
{

}
static unsigned int on_udp_conn_t_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	if (e->type == EVT_MSG_UDP_CONN_SEND_RESULT) {
		udp_conn_handle_tx_result(stm, e);
		if (conn_ctx->tx.concurrent == 0) {
			return 1; // goto Ready
		}
	} else if (e->type == STM_UDP_CONN_EVT_REQ_RX) {
		udp_conn_handle_rx_cmd(stm, e);
		if (conn_ctx->rx.concurrent != 0) {
			return 1; // goto RT
		}
	} else if (e->type == STM_UDP_CONN_EVT_REQ_TX) {
		udp_conn_handle_tx_cmd(stm, e);
	} else if (e->type == EVT_MSG_UDP_CONN_RECV_RESULT) {
		assert(0);
	}
	return 0; // no jump
}
static void on_udp_conn_t_exit(stm_machine_t *stm, const stm_state_t *state)
{

}

static void on_udp_conn_rt_init(stm_machine_t *stm, const stm_state_t *state)
{

}
static unsigned int on_udp_conn_rt_event(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	if (e->type == EVT_MSG_UDP_CONN_RECV_RESULT) {
		udp_conn_handle_rx_result(stm, e);
		if (conn_ctx->rx.concurrent == 0) {
			return 1; // goto T
		}
	} else if (e->type == EVT_MSG_UDP_CONN_SEND_RESULT) {
		udp_conn_handle_tx_result(stm, e);
		if (conn_ctx->tx.concurrent == 0) {
			return 1; // goto R
		}
	} else if (e->type == STM_UDP_CONN_EVT_REQ_TX) {
		udp_conn_handle_tx_cmd(stm, e);
	} else if (e->type == STM_UDP_CONN_EVT_REQ_RX) {
		udp_conn_handle_rx_cmd(stm, e);
	}
	return 0; // no jump
}
static void on_udp_conn_rt_exit(stm_machine_t *stm, const stm_state_t *state)
{

}

static unsigned int UDP_CONNECTION_on_init(stm_machine_t *stm, const stm_state_t *state, stm_event_t *e)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	int i;

	conn_ctx->fd = INVALID_SOCKET;
	conn_ctx->rx.concurrent = 0;
	conn_ctx->tx.concurrent = 0;
	for (i = 0; i < MAX_UDP_CONCURENT; ++i) {
		conn_ctx->rx.free_ops[i] = i;
		conn_ctx->rx.ops[i].type = IOCP_OP_NONE;
		conn_ctx->tx.free_ops[i] = i;
		conn_ctx->tx.ops[i].type = IOCP_OP_NONE;
	}
	conn_ctx->rx.free_ops_sp = i;
	conn_ctx->tx.free_ops_sp = i;

	stm_machine_defer_process_high_priority(stm, e);

	return 0;
}
static void UDP_CONNECTION_on_exit(stm_machine_t *stm)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	int i;

	for (i = 0; i < MAX_UDP_CONCURENT; ++i) {
		iocp_op_clear(&conn_ctx->rx.ops[i]);
		conn_ctx->rx.free_ops[i] = i;
		iocp_op_clear(&conn_ctx->tx.ops[i]);
		conn_ctx->tx.free_ops[i] = i;
	}
	conn_ctx->rx.free_ops_sp = MAX_UDP_CONCURENT;
	conn_ctx->tx.free_ops_sp = MAX_UDP_CONCURENT;

	printf("udp rx:%d tx:%d\n", conn_ctx->rx.concurrent, conn_ctx->tx.concurrent);
}
static void UDP_CONNECTION_on_destroy(stm_machine_t *stm)
{
	udp_conn_ctx_t *conn_ctx = (udp_conn_ctx_t *)stm_machine_get_context(stm);
	if (conn_ctx->fd != INVALID_SOCKET) {
		closesocket(conn_ctx->fd);
		conn_ctx->fd = INVALID_SOCKET;
	}
	printf("udp destroy!\n");
}
static void UDP_CONNECTION_on_state_trans(stm_machine_t *stm, const stm_state_t *src_state, const stm_state_t *dst_state)
{
	//printf("udp connection(%08x) (src,dst) : %d, %d\n", stm, src_state, dst_state);
}

BEGIN_STM_STATE(UDP_CONNECTION)
	STM_STATE_ENTRY(STM_UDP_CONN_START, "START", STM_STATE_ROOT, on_udp_conn_start_init, on_udp_conn_start_event, on_udp_conn_start_exit)
	STM_STATE_ENTRY(STM_UDP_CONN_READY, "READY", STM_STATE_ROOT, on_udp_conn_ready_init, on_udp_conn_ready_event, on_udp_conn_ready_exit)
	STM_STATE_ENTRY(STM_UDP_CONN_DOING, "DOING", STM_STATE_ROOT, on_udp_conn_doing_init, on_udp_conn_doing_event, on_udp_conn_doing_exit)
	STM_STATE_ENTRY(STM_UDP_CONN_R, "R", STM_UDP_CONN_DOING, on_udp_conn_r_init, on_udp_conn_r_event, on_udp_conn_r_exit)
	STM_STATE_ENTRY(STM_UDP_CONN_T, "T", STM_UDP_CONN_DOING, on_udp_conn_t_init, on_udp_conn_t_event, on_udp_conn_t_exit)
	STM_STATE_ENTRY(STM_UDP_CONN_RT, "RT", STM_UDP_CONN_DOING, on_udp_conn_rt_init, on_udp_conn_rt_event, on_udp_conn_rt_exit)
END_STM_STATE()

BEGIN_STM_STATE_RULE(UDP_CONNECTION)
	STM_STATE_RULE(STM_STATE_INIT, STM_UDP_CONN_EVT_MSG_INITIAL_ARG, STM_CMD_JMP_IF(STM_UDP_CONN_START, 0))
	STM_STATE_RULE(STM_UDP_CONN_START, STM_UDP_CONN_EVT_MSG_INITIAL_ARG, STM_CMD_JMP_IF(STM_UDP_CONN_READY, 1))
	STM_STATE_RULE(STM_UDP_CONN_READY, STM_UDP_CONN_EVT_REQ_RX, STM_CMD_JMP_IF(STM_UDP_CONN_R, 1))
	STM_STATE_RULE(STM_UDP_CONN_READY, STM_UDP_CONN_EVT_REQ_TX, STM_CMD_JMP_IF(STM_UDP_CONN_T, 1))
	STM_STATE_RULE(STM_UDP_CONN_READY, EVT_MSG_UDP_CONN_LATE_CLOSE, STM_CMD_JMP_IF(STM_STATE_EXIT, 0))
	STM_STATE_RULE(STM_UDP_CONN_READY, STM_UDP_CONN_EVT_MSG_CLOSE, STM_CMD_JMP_IF(STM_STATE_EXIT, 0))
	STM_STATE_RULE(STM_UDP_CONN_READY, EVT_MSG_UDP_CONN_SEND_RESULT, STM_CMD_NO_JMP_IF(0))
	STM_STATE_RULE(STM_UDP_CONN_READY, EVT_MSG_UDP_CONN_RECV_RESULT, STM_CMD_NO_JMP_IF(0))
	STM_STATE_RULE(STM_UDP_CONN_DOING, STM_UDP_CONN_EVT_MSG_CLOSE, STM_CMD_NO_JMP_IF(0))
	STM_STATE_RULE(STM_UDP_CONN_R, EVT_MSG_UDP_CONN_RECV_RESULT, STM_CMD_JMP_IF(STM_UDP_CONN_READY, 1))
	STM_STATE_RULE(STM_UDP_CONN_R, STM_UDP_CONN_EVT_REQ_TX, STM_CMD_JMP_IF(STM_UDP_CONN_RT, 1))
	STM_STATE_RULE(STM_UDP_CONN_R, STM_UDP_CONN_EVT_REQ_RX, STM_CMD_NO_JMP_IF(0))
	STM_STATE_RULE(STM_UDP_CONN_R, EVT_MSG_UDP_CONN_SEND_RESULT, STM_CMD_NO_JMP_IF(0))
	STM_STATE_RULE(STM_UDP_CONN_T, EVT_MSG_UDP_CONN_RECV_RESULT, STM_CMD_JMP_IF(STM_UDP_CONN_READY, 1))
	STM_STATE_RULE(STM_UDP_CONN_T, EVT_MSG_UDP_CONN_SEND_RESULT, STM_CMD_JMP_IF(STM_UDP_CONN_READY, 1))
	STM_STATE_RULE(STM_UDP_CONN_T, STM_UDP_CONN_EVT_REQ_RX, STM_CMD_JMP_IF(STM_UDP_CONN_RT, 1))
	STM_STATE_RULE(STM_UDP_CONN_T, STM_UDP_CONN_EVT_REQ_TX, STM_CMD_NO_JMP_IF(0))
	STM_STATE_RULE(STM_UDP_CONN_RT, EVT_MSG_UDP_CONN_RECV_RESULT, STM_CMD_JMP_IF(STM_UDP_CONN_T, 1))
	STM_STATE_RULE(STM_UDP_CONN_RT, EVT_MSG_UDP_CONN_SEND_RESULT, STM_CMD_JMP_IF(STM_UDP_CONN_R, 1))
	STM_STATE_RULE(STM_UDP_CONN_RT, STM_UDP_CONN_EVT_REQ_RX, STM_CMD_NO_JMP_IF(0))
	STM_STATE_RULE(STM_UDP_CONN_RT, STM_UDP_CONN_EVT_REQ_TX, STM_CMD_NO_JMP_IF(0))
END_STM_STATE_RULE()

DEFINE_STM_MACHINE(UDP_CONNECTION, STM_CAT_SYSTEM, STM_CLS_ID_UDP_CONNECTION, sizeof(udp_conn_ctx_t), UDP_CONNECTION_on_init, UDP_CONNECTION_on_exit, UDP_CONNECTION_on_destroy, UDP_CONNECTION_on_state_trans)


static stm_event_t *iocp_op_make_event(iocp_op_t *op, BOOL bRet, ULONG_PTR pKey, DWORD dwTransfer)
{
	stm_machine_t *stm = (stm_machine_t *)pKey;
	int param1 = 0, param2 = 0;
	op->dwLastError = GetLastError();
	switch (op->type) {
	case IOCP_OP_ACCEPT:
		param1 = !bRet;
		param2 = op->ap.id;
		return stm_event_create_base(EVT_MSG_TCP_AP_ACCEPT_RESULT, stm->stm_id, stm->stm_id, param1, param2, 0, 0);
	case IOCP_OP_CONNECT:
		param1 = !bRet;
		return stm_event_create_base(EVT_MSG_TCP_CONN_CONNECT_RESULT, stm->stm_id, stm->stm_id, param1, param2, 0, 0);
	case IOCP_OP_RX_ANY:
		param1 = !bRet;
		param2 = dwTransfer;
		op->any.offset += param2;
		return stm_event_create_base(EVT_MSG_TCP_CONN_RECV_RESULT, stm->stm_id, stm->stm_id, param1, param2, 0, 0);
	case IOCP_OP_RX_N:
		param1 = !bRet;
		param2 = dwTransfer;
		op->n.offset += param2;
		return stm_event_create_base(EVT_MSG_TCP_CONN_RECV_RESULT, stm->stm_id, stm->stm_id, param1, param2, 0, 0);
	case IOCP_OP_RX_UNTIL:
		param1 = !bRet;
		param2 = dwTransfer;
		op->t.offset += param2;
		return stm_event_create_base(EVT_MSG_TCP_CONN_RECV_RESULT, stm->stm_id, stm->stm_id, param1, param2, 0, 0);
	case IOCP_OP_TX:
		param1 = !bRet;
		param2 = dwTransfer;
		op->tx.offset += param2;
		return stm_event_create_base(EVT_MSG_TCP_CONN_SEND_RESULT, stm->stm_id, stm->stm_id, param1, param2, 0, 0);
	case IOCP_OP_TX_UDP:
		param1 = !bRet;
		param2 = op->udp_tx.id;
		op->udp_tx.offset = dwTransfer;
		return stm_event_create_base(EVT_MSG_UDP_CONN_SEND_RESULT, stm->stm_id, stm->stm_id, param1, param2, 0, 0);
	case IOCP_OP_RX_UDP:
		param1 = !bRet;
		param2 = op->udp_rx.id;
		op->udp_rx.offset = dwTransfer;
		{
			//printf("IOCP_RX_UDP, bRet = %d, dwTransfer=%d, %s:%d, lasterror:%d\n", bRet, dwTransfer, inet_ntoa(op->udp_rx.src_addr.sin_addr), ntohs(op->udp_rx.src_addr.sin_port), WSAGetLastError());
		}
		return stm_event_create_base(EVT_MSG_UDP_CONN_RECV_RESULT, stm->stm_id, stm->stm_id, param1, param2, 0, 0);
	default:
		break;
	}
	return NULL;
}

#pragma pack(push,1)
struct winthunk_t
{
	DWORD m_mov;
	DWORD m_this;
	BYTE m_jmp;
	DWORD   m_relproc;
};
#pragma pack(pop)

static int init_thunk(struct winthunk_t *thunk, DWORD_PTR proc, void *obj)
{
	thunk->m_mov = 0x042444C7;
	thunk->m_this = PtrToUlong(obj);
	thunk->m_jmp = 0xe9;
	thunk->m_relproc = (DWORD)((INT_PTR)proc - ((INT_PTR)obj+sizeof(struct winthunk_t)));
	FlushInstructionCache(GetCurrentProcess(), thunk, sizeof(struct winthunk_t));
	return 0;
}

typedef struct gui_dispatcher_timer_t
{
	struct winthunk_t thunk;
	UINT idEvent;
	int stm_id;
	stm_dispatcher_t *host_dispatcher;
}gui_dispatcher_timer_t;

typedef struct gui_dispatcher_t
{
	UINT message;
}gui_dispatcher_t;

static void CALLBACK TimerProc(HWND hwnd, UINT uMsg, UINT idEvent, DWORD dwTime)
{
	gui_dispatcher_timer_t *timer = (gui_dispatcher_timer_t*)hwnd;
	int stm_id = timer->stm_id;
	stm_event_t *e = stm_event_create_base(STM_SYS_EVT_MSG_TIMER, stm_id, stm_id, (int)timer, 0, 0, 0);
	KillTimer(NULL, timer->idEvent);
	stm_dispatcher_process(timer->host_dispatcher, e);
	stm_event_dec_ref(e);	
}

static void gui_dispatcher_on_destroy(stm_dispatcher_t *disp)
{
}

static int gui_dispatcher_post_event(stm_dispatcher_t *disp, stm_event_t *e)
{
	gui_dispatcher_t *gui_disp;
	int rc = -1;
	assert(disp);
	assert(e);
	gui_disp = (gui_dispatcher_t *)stm_dispatcher_get_context(disp);
	stm_event_inc_ref(e);
	if (PostThreadMessage(disp->thread_id, gui_disp->message, (WPARAM)e, 0)) {
		rc = 0;
	} else {
		stm_event_dec_ref(e);
	}
	return rc;
}

static int gui_dispatcher_schedule_timer(stm_dispatcher_t *disp, int stm_id, unsigned int interval)
{
	gui_dispatcher_timer_t *timer;
	assert(stm_id);
	assert(disp);
	assert(interval > 0);
	timer = (gui_dispatcher_timer_t*)VirtualAlloc(NULL, sizeof(gui_dispatcher_timer_t), MEM_COMMIT, PAGE_EXECUTE_READWRITE);
	if (timer) {
		gui_dispatcher_t *gui_disp;
		gui_disp = (gui_dispatcher_t *)stm_dispatcher_get_context(disp);		
		timer->stm_id = stm_id;
		timer->host_dispatcher = disp;
		init_thunk(&timer->thunk, (DWORD_PTR)&TimerProc, timer);
		timer->idEvent = SetTimer(NULL, 0, interval, (TIMERPROC)&timer->thunk);
		if (timer->idEvent == 0) {
			VirtualFree(timer, sizeof(gui_dispatcher_timer_t), MEM_RELEASE);
			timer = NULL;
		}
	}
	return (int)timer;
}

static void gui_dispatcher_cancel_timer(stm_dispatcher_t *disp, int timer_id)
{
	gui_dispatcher_timer_t *timer;
	assert(disp);
	assert(timer_id);
	timer = (gui_dispatcher_timer_t *)timer_id;
	KillTimer(NULL, timer->idEvent);
	VirtualFree(timer, sizeof(gui_dispatcher_timer_t), MEM_RELEASE);
}

static const struct stm_dispatcher_vtbl_t gui_dispatcher_vtbl = 
{
	"win32 gui dispatcher",
	&gui_dispatcher_on_destroy,
	&gui_dispatcher_post_event,
	&gui_dispatcher_schedule_timer,
	&gui_dispatcher_cancel_timer,
};

stm_dispatcher_t *stm_gui_dispatcher_create(UINT message)
{
	gui_dispatcher_t *gui_disp;
	stm_dispatcher_t *disp = stm_dispatcher_create_base(sizeof(gui_dispatcher_t), (void**)&gui_disp);
	if (disp) {
		disp->vtbl = &gui_dispatcher_vtbl;
		disp->thread_id = (unsigned int)GetCurrentThreadId();
		gui_disp->message = message;
	}
	return disp;
}

TOOLKIT_API BOOL stm_gui_dispatcher_pre_process_msg(stm_dispatcher_t *disp, MSG *pMsg)
{
	gui_dispatcher_t *gui_disp;
	assert(disp);
	assert(pMsg);
	gui_disp = (gui_dispatcher_t *)stm_dispatcher_get_context(disp);
	assert(GetCurrentThreadId() == disp->thread_id);
	if (pMsg->message == gui_disp->message) {
		stm_event_t *e = (stm_event_t*)pMsg->wParam;
		stm_dispatcher_process(disp, e);
		stm_event_dec_ref(e);
		return TRUE;
	}
	return FALSE;
}

TOOLKIT_API int stm_init()
{
	REGISTER_STM_MACHINE(TCP_ACCEPTOR);
	REGISTER_STM_MACHINE(TCP_CONNECTION);
	REGISTER_STM_MACHINE(UDP_CONNECTION);
	return 0;
}

TOOLKIT_API void stm_term()
{
	UNREGISTER_STM_MACHINE(TCP_ACCEPTOR);
	UNREGISTER_STM_MACHINE(TCP_CONNECTION);
	UNREGISTER_STM_MACHINE(UDP_CONNECTION);
}