myvtm/libtoolkit/ioqueue-unix.c

#include "precompile.h"
#include "ioqueue.h"
#include "timerqueue.h"
#include "memutil.h"
#include "refcnt.h"
#include "strutil.h"
#include "sockutil.h"
#include "bus.h"
#include "array.h"
#include "evtpoll.h"

#include <winpr/file.h>
#include <winpr/handle.h>
#include <winpr/synch.h>
#include <winpr/pipe.h>
#include <winpr/string.h>
#include <winpr/wlog.h>

#define TAG TOOLKIT_TAG("ioqueue")

#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 MT_TTL						10*60*1000 /* maintenance 10 minutes */
#define MT_INTERVAL					30*1000 /* maintenance 30 seconds */

/* The address specified in AcceptEx() must be 16 more than the size of
 * SOCKADDR (source: MSDN).
 */
#define ACCEPT_ADDR_LEN	(16+sizeof(SOCKADDR))

struct ioqueue_t {
	HANDLE iocp; //for build successfully temporary!!!!!!!!!!!!!!! TODO: delete it!!

	evtpoll_t* ep;
	
	void *user_data;
	/* timer */
	spinlock_t tm_queue_lock;
	timer_queue_t *tm_queue;
	/* msg handler */
	ioqueue_on_msg_callback msg_handlers[MAX_MSG][MAX_MSG_PRIORITY];
	LONG msg_cnt;
	/* connect */
	spinlock_t connect_list_lock;
	struct list_head connect_list;
	spinlock_t handler_list_lock;
	struct list_head handler_list;
	LONG stop;
};

typedef struct ioqueue_msg {
	int msg_type;
	int param1;
	int param2;
	HANDLE evt; /* for send message */
}ioqueue_msg;

#define HANDLE_TYPE_ACCEPTOR		        0x01
#define HANDLE_TYPE_TCPSOCK			    0x02
#define HANDLE_TYPE_UDPSOCK			    0x03
#define HANDLE_TYPE_FILE			            0x04
#define HANDLE_TYPE_PIPEACCEPTOR	    0x05

#define OV_ACCEPT			        0x01
#define OV_CONNECT			    0x02
#define OV_SENDSOME			    0x03
#define OV_SENDN			            0x04
#define OV_RECVSOME			    0x05
#define OV_RECVN			            0x06
#define OV_SENDTO			        0x07
#define OV_RECVFROM			    0x08
#define OV_READFILESOME		0x09
#define OV_WRITEFILESOME  	0x0a
#define OV_READFILEN		        0x0b
#define OV_WRITEFILEN		        0x0c
#define OV_RECVUNTIL		        0x0d
#define OV_CONNECTPIPE		    0x0e

typedef struct ioqueue_base_overlapped_t {
	OVERLAPPED ov;
	int type;
	void *user_data;
	struct list_head pending_entry;
	ioqueue_handle_context *handle_ctx;
}ioqueue_base_overlapped_t;

typedef struct ioqueue_accept_overlapped_t {
	ioqueue_base_overlapped_t base;
	SOCKET client;
	ioqueue_on_accept_callback on_accept_callback;
	char accept_buf[2*ACCEPT_ADDR_LEN];
}ioqueue_accept_overlapped_t;

typedef struct ioqueue_connect_overlapped_t {
	ioqueue_base_overlapped_t base;
	ioqueue_on_connect_callback on_connect_callback;
	struct list_head node;
	HANDLE hevt;
}ioqueue_connect_overlapped_t;

typedef struct ioqueue_sendsome_overlapped_t {
	ioqueue_base_overlapped_t base;
	ioqueue_on_send_callback on_send_callback;
	WSABUF wsabuf;
}ioqueue_sendsome_overlapped_t;

typedef struct ioqueue_sendn_overlapped_t {
	ioqueue_base_overlapped_t base;
	ioqueue_on_send_callback on_send_callback;
	WSABUF wsabuf;
	char *original_buf;
	unsigned int sended_bytes;
	unsigned int total_bytes;
}ioqueue_sendn_overlapped_t;

typedef struct ioqueue_recvsome_overlapped_t {
	ioqueue_base_overlapped_t base;
	ioqueue_on_recv_callback on_recv_callback;
	WSABUF wsabuf;
	DWORD dwFlags;
}ioqueue_recvsome_overlapped_t;

typedef struct ioqueue_recvn_overlapped_t {
	ioqueue_base_overlapped_t base;
	ioqueue_on_recv_callback on_recv_callback;
	WSABUF wsabuf;
	char *original_buf;
	unsigned int recved_bytes;
	unsigned int total_bytes;
	DWORD dwFlags;
}ioqueue_recvn_overlapped_t;

typedef struct ioqueue_recvuntil_overlapped_t {
	ioqueue_base_overlapped_t base;
	ioqueue_on_recvuntil_callback on_recvuntil_callback;
	WSABUF wsabuf;
	char *original_buf;
	char *delimer;
	unsigned int recved_bytes;
	unsigned int total_bytes;
	DWORD dwFlags;
}ioqueue_recvuntil_overlapped_t;

typedef struct ioqueue_sendto_overlapped_t {
	ioqueue_base_overlapped_t base;
	ioqueue_on_sendto_callback on_sendto_callback;
	WSABUF wsabuf;
}ioqueue_sendto_overlapped_t;

typedef struct ioqueue_recvfrom_overlapped_t {
	ioqueue_base_overlapped_t base;
	ioqueue_on_recvfrom_callback on_recvfrom_callback;
	WSABUF wsabuf;
	struct sockaddr_in peer;
	int addrlen;
	DWORD dwFlags;
}ioqueue_recvfrom_overlapped_t;

typedef struct ioqueue_readfilesome_overlapped_t {
	ioqueue_base_overlapped_t base;
	ioqueue_on_read_callback on_read_callback;
	char *buf;
	HANDLE hevt;
}ioqueue_readfilesome_overlapped_t;

typedef struct ioqueue_readfilen_overlapped_t {
	ioqueue_base_overlapped_t base;
	ioqueue_on_read_callback on_read_callback;
	char *buf;
	HANDLE hevt;
	unsigned int recved_bytes;
	unsigned int total_bytes;
}ioqueue_readfilen_overlapped_t;

typedef struct ioqueue_writefilesome_overlapped_t {
	ioqueue_base_overlapped_t base;
	ioqueue_on_write_callback on_write_callback;
	HANDLE hevt;
	char *buf;
}ioqueue_writefilesome_overlapped_t;

typedef struct ioqueue_writefilen_overlapped_t {
	ioqueue_base_overlapped_t base;
	ioqueue_on_write_callback on_write_callback;
	char *buf;
	HANDLE hevt;
	unsigned int sended_bytes;
	unsigned int total_bytes;
}ioqueue_writefilen_overlapped_t;

typedef struct ioqueue_connectpipe_overlapped_t {
	ioqueue_base_overlapped_t base;
	HANDLE client;
	HANDLE hevt;
	ioqueue_on_pipe_accept_callback on_accept_callback;
}ioqueue_connectpipe_overlapped_t;

static int is_os_gte_xp() /* is os version greater and equal than xp */
{
	static int yes = -1;
#ifdef _WIN32
	if (yes == -1) {
		OSVERSIONINFO ver;
		ver.dwOSVersionInfoSize = sizeof(OSVERSIONINFO);
		GetVersionEx(&ver);
		if (ver.dwMajorVersion > 5 || (ver.dwMajorVersion == 5 && ver.dwMinorVersion > 0)) {
			yes = 1;
		}
	}
#endif//_WIN32
	return yes;
}

static __inline LONG inc_msg_cnt(ioqueue_t *ioq) 
{ 
	return InterlockedIncrement(&ioq->msg_cnt); 
}

static __inline LONG dec_msg_cnt(ioqueue_t *ioq) 
{ 
	return InterlockedDecrement(&ioq->msg_cnt); 
}

static __inline void add_handler_list(ioqueue_handle_context *handle_ctx, ioqueue_t *ioq)
{
	spinlock_enter(&ioq->handler_list_lock, -1);
	list_add(&handle_ctx->node, &ioq->handler_list);
	spinlock_leave(&ioq->handler_list_lock);
}

static __inline void del_handler_list(ioqueue_handle_context *handle_ctx, ioqueue_t *ioq)
{
	if (handle_ctx->node.next) {
		spinlock_enter(&ioq->handler_list_lock, -1);
		list_del(&handle_ctx->node);
		handle_ctx->node.next = handle_ctx->node.prev = NULL;
		spinlock_leave(&ioq->handler_list_lock);
	}
}

static void ioqueue_handle_context_free(ioqueue_handle_context *handle_ctx)
{
	if (handle_ctx->type == HANDLE_TYPE_UDPSOCK 
		|| handle_ctx->type == HANDLE_TYPE_TCPSOCK 
		|| handle_ctx->type == HANDLE_TYPE_ACCEPTOR) {
		if (handle_ctx->u.sock != INVALID_SOCKET) {
			closesocket(handle_ctx->u.sock);
			handle_ctx->u.sock = INVALID_SOCKET;
		}
	} else if (handle_ctx->type == HANDLE_TYPE_FILE) {
		if (handle_ctx->u.file != INVALID_HANDLE_VALUE) {
			CloseHandle(handle_ctx->u.file);
			handle_ctx->u.file = INVALID_HANDLE_VALUE;
		}
	} else if (handle_ctx->type == HANDLE_TYPE_PIPEACCEPTOR) {
		if (handle_ctx->u.pipe_name) {
			free(handle_ctx->u.pipe_name);
			handle_ctx->u.pipe_name = NULL;
		}
	} else {
		assert(0);
		return;
	}
	del_handler_list(handle_ctx, handle_ctx->owner);
}

IMPLEMENT_REF_COUNT_MT(ioqueue_handle_context, ioqueue_handle_context, pending_ios, ioqueue_handle_context_free)

static __inline LONG inc_pending_io(ioqueue_handle_context *handle_ctx)
{
	return inc_ref(ioqueue_handle_context, handle_ctx);
}

static __inline LONG dec_pending_io(ioqueue_handle_context *handle_ctx)
{
	return dec_ref(ioqueue_handle_context, handle_ctx);
}

static SOCKET new_socket()
{
	SOCKET sock = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, 
		NULL, 0, WSA_FLAG_OVERLAPPED);
	if (sock != INVALID_SOCKET) {
		reuse_addr(sock);
	}
	return sock;
}

static void delete_socket(SOCKET sock)
{
	LINGER l;
	l.l_onoff = 1;
	l.l_linger = 0;
	setsockopt(sock, SOL_SOCKET, SO_LINGER, (char*)&l, sizeof(l));
	closesocket(sock);
}

TOOLKIT_API ioqueue_t *ioqueue_create()
{
	ioqueue_t *ioq = ZALLOC_T(ioqueue_t);
	if (!ioq)
		return NULL;

	ioq->ep = evtpoll_create();
	if (!ioq->ep) {
		goto on_error_0;
	}

	if (timer_heap_create(&ioq->tm_queue) != 0) {
		goto on_error_3;
	}
	spinlock_init(&ioq->tm_queue_lock);
	spinlock_init(&ioq->connect_list_lock);
	INIT_LIST_HEAD(&ioq->connect_list);
	spinlock_init(&ioq->handler_list_lock);
	INIT_LIST_HEAD(&ioq->handler_list);

	return ioq;

on_error_3:
	evtpoll_destroy(ioq->ep);
on_error_0:
	free(ioq);
	return NULL;
}

TOOLKIT_API void ioqueue_destroy(ioqueue_t *ioq)
{
	assert(ioq);
	assert(ioqueue_handler_empty(ioq));
	assert(ioqueue_msg_empty(ioq));
	timer_queue_destroy(ioq->tm_queue);
	evtpoll_destroy(ioq->ep);
	free(ioq);
}

TOOLKIT_API int ioqueue_handler_empty(ioqueue_t *ioq)
{
	int ret;
	assert(ioq);
	spinlock_enter(&ioq->handler_list_lock, -1);
	ret = list_empty(&ioq->handler_list);
	spinlock_leave(&ioq->handler_list_lock);
	return ret;
}

TOOLKIT_API int ioqueue_msg_empty(ioqueue_t *ioq)
{
	assert(ioq);
	return ioq->msg_cnt == 0;
}

TOOLKIT_API int ioqueue_msg_add_handler(ioqueue_t *ioq, int msg_type, int priority, ioqueue_on_msg_callback cb)
{
	assert(ioq);
	assert(cb);
	assert(msg_type >= 0 && msg_type <= MAX_MSG);
	assert(priority >= 0 && priority <= MAX_MSG_PRIORITY);
	ioq->msg_handlers[msg_type][priority] = cb;
	return 0;
}

TOOLKIT_API int ioqueue_msg_remove_handler(ioqueue_t *ioq, int msg_type, int priority)
{
	assert(ioq);
	assert(msg_type >= 0 && msg_type <= MAX_MSG);
	assert(priority >= 0 && priority <= MAX_MSG_PRIORITY);
	ioq->msg_handlers[msg_type][priority] = NULL;
	return 0;
}

static void dispatch_acceptor(int err, 
							  DWORD dwBytesTransfer, 
							  ioqueue_acceptor_t *acceptor, 
							  ioqueue_accept_overlapped_t *overlapped)
{
	ioqueue_t *ioq = acceptor->owner;

	if (err == 0) {
		/* only valid for winxp or later, ignore return value */
		setsockopt(overlapped->client, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, 
			(char*)&acceptor->u.sock, sizeof(SOCKET));
	} else {
		delete_socket(overlapped->client);
		overlapped->client = INVALID_SOCKET;
	}
	{
	SOCKET s = overlapped->client;
	int accepted = overlapped->on_accept_callback(acceptor, (ioqueue_overlapped_t*)overlapped, s, 
		overlapped->base.user_data, err);
	if (!err && !accepted && s != INVALID_SOCKET)
		delete_socket(s);
	}
}

static void dispatch_pipe_acceptor(int err, 
								   DWORD dwBytesTransfer, 
								   ioqueue_pipe_acceptor_t *acceptor, 
								   ioqueue_connectpipe_overlapped_t *overlapped)
{
	ioqueue_t *ioq = acceptor->owner;
	int accepted;

	CloseHandle(overlapped->hevt);
	overlapped->hevt = NULL;

	if (!err && overlapped->client == INVALID_HANDLE_VALUE)
		err = -1;

	if (err) {
		if (overlapped->client != INVALID_HANDLE_VALUE) {
			CloseHandle(overlapped->client);
			overlapped->client = INVALID_HANDLE_VALUE;
		}
	}

	accepted = overlapped->on_accept_callback(acceptor, 
		(ioqueue_overlapped_t*)overlapped, 
		overlapped->client,
		overlapped->base.user_data, err);
	if (!err && !accepted && overlapped->client != INVALID_HANDLE_VALUE) {
		CloseHandle(overlapped->client);
	}
}

static int pre_dispatch_network(BOOL* ret, DWORD* dwBytesTransfer, ioqueue_overlapped_t* io_ctx)
{
	int err = 0;
	ioqueue_base_overlapped_t* base_ov = (ioqueue_base_overlapped_t*)io_ctx;
	ioqueue_handle_context* handle_ctx = base_ov->handle_ctx;

	if (ret) *ret = FALSE;
	switch (handle_ctx->type) {
	case HANDLE_TYPE_ACCEPTOR:
	{
		SOCKET conn_socket;
		struct sockaddr_in addr;
		socklen_t addrlen;
		assert(((ioqueue_accept_overlapped_t*)io_ctx)->client == INVALID_SOCKET);
		addrlen = sizeof(addr);
		bzero(&addr, addrlen);
		conn_socket = accept(handle_ctx->u.sock, (struct sockaddr*) & addr, &addrlen);
		if (conn_socket == -1) {
			WLog_ERR(TAG, "accept connect socket failed: %d", errno);
			return -1;
		}
		WLog_INFO(TAG, "new connected socket fd(%d) arrived at listen socket: %d from %s:%d",
			conn_socket, handle_ctx->u.sock, inet_ntoa(addr.sin_addr), ntohs(addr.sin_port));

		((ioqueue_accept_overlapped_t*)io_ctx)->client = conn_socket;
		if (ret) *ret = TRUE;
		if (dwBytesTransfer) *dwBytesTransfer = 0;
		err = 1;
	}
	break;
	case HANDLE_TYPE_PIPEACCEPTOR:
		break;
	case HANDLE_TYPE_TCPSOCK:
	case HANDLE_TYPE_UDPSOCK:
	case HANDLE_TYPE_FILE:
		switch (base_ov->type) {
		case OV_CONNECT: //暂时没有看到该事件的使用
			break;
		case OV_SENDSOME: {

		}
						break;
		case OV_SENDN: {
			int n, nwrite;
			ioqueue_recvn_overlapped_t* overlapped = (ioqueue_recvn_overlapped_t*)io_ctx;
			ioqueue_tcpsock_t* tcpsock = overlapped->base.handle_ctx;
			n = 0;
			for (; n < overlapped->wsabuf.len; ) {
				do 
				{
					nwrite = write(tcpsock->u.sock, overlapped->wsabuf.buf + n, overlapped->wsabuf.len - n);

				} while (nwrite <0 && errno == EINTR);
				if (nwrite == 0) {
					WLog_WARN(TAG, "OV_SENDN: remote socket exit.");
					break;
				}
				else if (nwrite < 0) {
					if (errno == EAGAIN || errno == EWOULDBLOCK) {
						nwrite = 0;
					}
					else {
						WLog_ERR(TAG, "write error: %s(%d)", strerror(errno), errno);
					}
					break;
				}
				else {
					n += nwrite;
				}
			}
			WLog_INFO(TAG, "<OV_SENDN>: total_write: %d/%d, cur_send: %d", n, overlapped->wsabuf.len, nwrite);
			if (ret) {
				*ret = (n == 0 && nwrite <= 0) ? FALSE : TRUE;
			}
			if (dwBytesTransfer) *dwBytesTransfer = n;
			err = 1;
		}
					 break;
		case OV_RECVSOME: {

		}
						break;
		case OV_RECVUNTIL: {

		}
						 break;
		case OV_RECVN: {
			int n, nread;
			ioqueue_recvn_overlapped_t* overlapped = (ioqueue_recvn_overlapped_t*)io_ctx;
			ioqueue_tcpsock_t* tcpsock = overlapped->base.handle_ctx;
			n = 0;
			for (; n < overlapped->wsabuf.len;)
			{
				do  
				{
					nread = _recv(tcpsock->u.sock, overlapped->wsabuf.buf + n, overlapped->wsabuf.len - n, 0);
				} while (nread < 0 && errno == EINTR);
				if (nread > 0) {
					n += nread;
					continue;
				}
				if (nread == 0) {
					/*remote socket break*/
					WLog_WARN(TAG, "OV_RECVN: remote socket exit.");
					break;
				} else if (errno == EAGAIN || errno == EWOULDBLOCK) {
					nread = 0;
					break;
				} else {
					WLog_ERR(TAG, "read error: %s(%d)", strerror(errno), errno);
					break;
				}
			}
			WLog_INFO(TAG, "<OV_RECVN>: total_recv: %d/%d, cur_recv: %d", n, overlapped->wsabuf.len, nread);
			if (ret) {//TRUE means correct
				*ret = (n == 0 && nread <= 0) ? FALSE : TRUE;
			}
			if (dwBytesTransfer) *dwBytesTransfer = n;
			err = 1;
		}
					 break;
		case OV_SENDTO: {

		}
					  break;
		case OV_RECVFROM: {

		}
						break;
		case OV_READFILESOME: {

		}
							break;
		case OV_READFILEN: {

		}
						 break;
		case OV_WRITEFILESOME: {

		}
							 break;
		case OV_WRITEFILEN: {

		}
						  break;
		}
		break;
	default:
		assert(0);
		break;
	}
	return err;
}

static void dispatch_network(BOOL ret, DWORD dwBytesTransfer, ioqueue_overlapped_t *io_ctx)
{
	int err = ret ? 0 : -1; //-1 means error
	ioqueue_base_overlapped_t *base_ov = (ioqueue_base_overlapped_t*)io_ctx;
	ioqueue_handle_context *handle_ctx = base_ov->handle_ctx;

	fastlock_enter(handle_ctx->ov_pending_list_lock);
	list_del(&base_ov->pending_entry);
	fastlock_leave(handle_ctx->ov_pending_list_lock);
	dec_pending_io(handle_ctx);

	switch (handle_ctx->type) {
		case HANDLE_TYPE_ACCEPTOR:
			dispatch_acceptor(err, dwBytesTransfer, handle_ctx, (ioqueue_accept_overlapped_t*)io_ctx);
			break;
		case HANDLE_TYPE_PIPEACCEPTOR:
			dispatch_pipe_acceptor(err, dwBytesTransfer, handle_ctx, (ioqueue_connectpipe_overlapped_t*)io_ctx);
			break;
		case HANDLE_TYPE_TCPSOCK:
		case HANDLE_TYPE_UDPSOCK:
		case HANDLE_TYPE_FILE:
			switch (base_ov->type) {
				case OV_CONNECT: {
						ioqueue_connect_overlapped_t *overlapped = (ioqueue_connect_overlapped_t*)io_ctx;
						if (err == 0) {
							setsockopt(handle_ctx->u.sock, SOL_SOCKET, SO_UPDATE_CONNECT_CONTEXT, NULL, 0);
						}
						overlapped->on_connect_callback(handle_ctx, io_ctx, base_ov->user_data, err);
					}
					break;
				case OV_SENDSOME: {
						ioqueue_sendsome_overlapped_t *overlapped = (ioqueue_sendsome_overlapped_t*)io_ctx;
						overlapped->on_send_callback(handle_ctx, io_ctx, 
							overlapped->wsabuf.buf, dwBytesTransfer, base_ov->user_data, err);
					}
					break;
				case OV_SENDN: {
						ioqueue_sendn_overlapped_t *overlapped = (ioqueue_sendn_overlapped_t*)io_ctx;
						overlapped->sended_bytes += dwBytesTransfer;
						if (err == 0 && overlapped->sended_bytes < overlapped->total_bytes) {
							ioqueue_tcpsock_t* tcpsock = overlapped->base.handle_ctx;
							overlapped->wsabuf.buf += dwBytesTransfer;
							overlapped->wsabuf.len -= dwBytesTransfer;
							inc_pending_io(handle_ctx);
							overlapped->base.ov.Internal = 0;
							overlapped->base.ov.InternalHigh = 0;
							overlapped->base.ov.Offset = 0;
							overlapped->base.ov.OffsetHigh = 0;
							ioqueue_overlapped_set_mask(io_ctx, sizeof(ioqueue_sendn_overlapped_t));
							WLog_WARN(TAG, "OV_SENDN: Must be due with this situation %d < %d", overlapped->sended_bytes, overlapped->total_bytes);
							evtpoll_subscribe(ioqueue_tcpsock_get_owned_ioqueue(tcpsock)->ep, EV_WRITE, tcpsock->u.sock, NULL, io_ctx);
							//if (rc != 0 && WSAGetLastError() != WSA_IO_PENDING) {
							//	dec_pending_io(handle_ctx);
							//	overlapped->on_send_callback(handle_ctx, io_ctx, overlapped->original_buf, 
							//		overlapped->sended_bytes, base_ov->user_data, -1);
							//}
						} else {
							ioqueue_tcpsock_t* tcpsock = overlapped->base.handle_ctx;
							evtpoll_unsubscribe(ioqueue_tcpsock_get_owned_ioqueue(tcpsock)->ep, EV_WRITE, tcpsock->u.sock, 0);
							overlapped->on_send_callback(handle_ctx, io_ctx, overlapped->original_buf, 
								overlapped->sended_bytes, base_ov->user_data, err);
						}
					}
					break;
				case OV_RECVSOME: {
						ioqueue_recvsome_overlapped_t *overlapped = (ioqueue_recvsome_overlapped_t*)io_ctx;
						overlapped->on_recv_callback(handle_ctx, io_ctx, overlapped->wsabuf.buf, 
							dwBytesTransfer, base_ov->user_data, err);
					}
					break;
				case OV_RECVUNTIL: {
						ioqueue_recvuntil_overlapped_t *overlapped = (ioqueue_recvuntil_overlapped_t*)io_ctx;
						if (err == 0 && dwBytesTransfer) {
							const char *pos;
							overlapped->recved_bytes += dwBytesTransfer;
							pos = memstr(overlapped->original_buf, overlapped->recved_bytes, overlapped->delimer);
							if (pos) {
								free(overlapped->delimer);
								overlapped->on_recvuntil_callback(handle_ctx, io_ctx, overlapped->original_buf,
									overlapped->recved_bytes, (int)strlen(overlapped->delimer)+(int)(pos-overlapped->original_buf),
									base_ov->user_data, err);
							} else if (overlapped->recved_bytes < overlapped->total_bytes) {
								DWORD bytesRead;
								int rc;
								overlapped->wsabuf.buf += dwBytesTransfer;
								overlapped->wsabuf.len -= dwBytesTransfer;
								inc_pending_io(handle_ctx);
								overlapped->base.ov.Internal = 0;
								overlapped->base.ov.InternalHigh = 0;
								overlapped->base.ov.Offset = 0;
								overlapped->base.ov.OffsetHigh = 0;
								overlapped->dwFlags = 0;
								ioqueue_overlapped_set_mask(io_ctx, sizeof(ioqueue_recvuntil_overlapped_t));
								rc = WSARecv(handle_ctx->u.sock, &overlapped->wsabuf, 1, &bytesRead, &overlapped->dwFlags, 
									&overlapped->base.ov, NULL);
								if (rc != 0 && WSAGetLastError() != WSA_IO_PENDING) {
									dec_pending_io(handle_ctx);
									free(overlapped->delimer);
									overlapped->on_recvuntil_callback(handle_ctx, io_ctx, overlapped->original_buf,
										overlapped->recved_bytes, 0, base_ov->user_data, -1);
								}
							} else {
								free(overlapped->delimer);
								overlapped->on_recvuntil_callback(handle_ctx, io_ctx, overlapped->original_buf,
										overlapped->recved_bytes, 0, base_ov->user_data, -1);
							}
						} else {
							free(overlapped->delimer);
							overlapped->on_recvuntil_callback(handle_ctx, io_ctx, overlapped->original_buf,
								overlapped->recved_bytes, 0, base_ov->user_data, err);
						}
					}
					break;
				case OV_RECVN: {
						ioqueue_recvn_overlapped_t *overlapped = (ioqueue_recvn_overlapped_t*)io_ctx;
						overlapped->recved_bytes += dwBytesTransfer;
						if (err == 0 && overlapped->recved_bytes < overlapped->total_bytes) {
							ioqueue_tcpsock_t* tcpsock = overlapped->base.handle_ctx;
							overlapped->wsabuf.buf += dwBytesTransfer;
							overlapped->wsabuf.len -= dwBytesTransfer;
							inc_pending_io(handle_ctx);
							overlapped->base.ov.Internal = 0;
							overlapped->base.ov.InternalHigh = 0;
							overlapped->base.ov.Offset = 0;
							overlapped->base.ov.OffsetHigh = 0;
							overlapped->dwFlags = 0;
							ioqueue_overlapped_set_mask(io_ctx, sizeof(ioqueue_recvn_overlapped_t));
							WLog_WARN(TAG, "OV_RECVN: Must be due with this situation %d < %d", overlapped->recved_bytes, overlapped->total_bytes);
							evtpoll_subscribe(ioqueue_tcpsock_get_owned_ioqueue(tcpsock)->ep,  EV_READ, tcpsock->u.sock, io_ctx, NULL);
							//if (rc != 0 && WSAGetLastError() != WSA_IO_PENDING) {
							//	dec_pending_io(handle_ctx);
							//	overlapped->on_recv_callback(handle_ctx, io_ctx, overlapped->original_buf, 
							//		overlapped->recved_bytes, base_ov->user_data, -1);
							//}
						} else {
							overlapped->on_recv_callback(handle_ctx, io_ctx, overlapped->original_buf, 
								overlapped->recved_bytes, base_ov->user_data, err);
						}
					}
					break;
				case OV_SENDTO: {
						ioqueue_sendto_overlapped_t *overlapped = (ioqueue_sendto_overlapped_t*)io_ctx;
						overlapped->on_sendto_callback(handle_ctx, io_ctx, overlapped->wsabuf.buf,
							dwBytesTransfer, base_ov->user_data, err);
					}
					break;
				case OV_RECVFROM: {
						ioqueue_recvfrom_overlapped_t *overlapped = (ioqueue_recvfrom_overlapped_t*)io_ctx;
						overlapped->on_recvfrom_callback(handle_ctx, io_ctx, (struct sockaddr*)&overlapped->peer, 
							overlapped->addrlen, overlapped->wsabuf.buf, dwBytesTransfer, base_ov->user_data, err);
					}
					break;
				case OV_READFILESOME: {
						ioqueue_readfilesome_overlapped_t *overlapped = (ioqueue_readfilesome_overlapped_t*)io_ctx;
						CloseHandle(overlapped->hevt);
						overlapped->hevt = NULL;
						overlapped->on_read_callback(handle_ctx, io_ctx, overlapped->buf,
							dwBytesTransfer, base_ov->user_data, err);
					}
					break;
				case OV_READFILEN: {
						ioqueue_readfilen_overlapped_t *overlapped = (ioqueue_readfilen_overlapped_t*)io_ctx;
						CloseHandle(overlapped->hevt);
						overlapped->hevt = NULL;
						overlapped->recved_bytes += dwBytesTransfer;
						if (err == 0 && overlapped->recved_bytes < overlapped->total_bytes) {
							BOOL ret;
							DWORD left = overlapped->total_bytes - overlapped->recved_bytes;
							inc_pending_io(handle_ctx);
							overlapped->base.ov.Internal = 0;
							overlapped->base.ov.InternalHigh = 0;
							overlapped->base.ov.Offset += dwBytesTransfer;
							if (overlapped->base.ov.Offset < dwBytesTransfer)
								overlapped->base.ov.OffsetHigh += 1;
							ioqueue_overlapped_set_mask(io_ctx, sizeof(ioqueue_readfilen_overlapped_t));
							ret = ReadFile(handle_ctx->u.file, overlapped->buf+overlapped->recved_bytes, left, NULL, &overlapped->base.ov);
							if (!ret && GetLastError() != ERROR_IO_PENDING) {
								dec_pending_io(handle_ctx);
								overlapped->on_read_callback(handle_ctx, io_ctx, overlapped->buf, 
									overlapped->recved_bytes, base_ov->user_data, -1);
							}
						} else {
							overlapped->on_read_callback(handle_ctx, io_ctx, overlapped->buf, 
								overlapped->recved_bytes, base_ov->user_data, err);
						}
					}
					break;
				case OV_WRITEFILESOME: {
						ioqueue_writefilesome_overlapped_t *overlapped = (ioqueue_writefilesome_overlapped_t*)io_ctx;
						CloseHandle(overlapped->hevt);
						overlapped->hevt = NULL;
						overlapped->on_write_callback(handle_ctx, io_ctx, overlapped->buf, dwBytesTransfer, 
							base_ov->user_data, err);
					}
					break;
				case OV_WRITEFILEN: {
						ioqueue_writefilen_overlapped_t *overlapped = (ioqueue_writefilen_overlapped_t*)io_ctx;
						CloseHandle(overlapped->hevt);
						overlapped->hevt = NULL;
						overlapped->sended_bytes += dwBytesTransfer;
						if (err == 0 && overlapped->sended_bytes < overlapped->total_bytes) {
							BOOL ret;
							DWORD left = overlapped->total_bytes - overlapped->sended_bytes;
							inc_pending_io(handle_ctx);
							overlapped->base.ov.Internal = 0;
							overlapped->base.ov.InternalHigh = 0;
							overlapped->base.ov.Offset += dwBytesTransfer;
							if (overlapped->base.ov.Offset < dwBytesTransfer)
								overlapped->base.ov.OffsetHigh += 1;
							ioqueue_overlapped_set_mask(io_ctx, sizeof(ioqueue_writefilen_overlapped_t));
							ret = WriteFile(handle_ctx->u.file, overlapped->buf+overlapped->sended_bytes, left, NULL, &overlapped->base.ov);
							if (!ret && GetLastError() != ERROR_IO_PENDING) {
								dec_pending_io(handle_ctx);
								overlapped->on_write_callback(handle_ctx, io_ctx, overlapped->buf, 
									overlapped->sended_bytes, base_ov->user_data, -1);
							}
						} else {
							overlapped->on_write_callback(handle_ctx, io_ctx, overlapped->buf, 
								overlapped->sended_bytes, base_ov->user_data, err);
						}
					}
					break;
				default:
					assert(0);
					break;
			}
			break;
		default:
			assert(0);
			break;
	}
}

static void dispatch_msg(ioqueue_t *ioq, int msg_type, int param1, int param2, HANDLE evt)
{
	int chain = 1, i;
	for (i = 0; chain && i < MAX_MSG_PRIORITY; ++i) {
		ioqueue_on_msg_callback cb = ioq->msg_handlers[msg_type][i];
		if (cb) 
			chain = cb(msg_type, param1, param2);
	}
	if (evt) 
		SetEvent(evt);
}

TOOLKIT_API int ioqueue_post_message(ioqueue_t *ioq, int msg_type, int param1, int param2)
{
	ioqueue_msg *msg;
	assert(ioq);
	msg = MALLOC_T(ioqueue_msg);
	msg->msg_type = msg_type;
	msg->param1 = param1;
	msg->param2 = param2;
	msg->evt = NULL;
	inc_msg_cnt(ioq);
	if (!PostQueuedCompletionStatus(ioq->iocp, 0, (ULONG_PTR)msg, NULL)) {
		dec_msg_cnt(ioq);
		free(msg);
		return -1;
	}
	return 0;
}

TOOLKIT_API int ioqueue_send_message(ioqueue_t *ioq, int msg_type, int param1, int param2)
{
	ioqueue_msg msg = {msg_type, param1, param2};
	assert(ioq);
	msg.evt = CreateEventA(NULL, TRUE, FALSE, NULL);
	inc_msg_cnt(ioq);
	if (!PostQueuedCompletionStatus(ioq->iocp, 0, (ULONG_PTR)&msg, NULL)) {
		CloseHandle(msg.evt);
		dec_msg_cnt(ioq);
		return -1;
	}
	WaitForSingleObject(msg.evt, INFINITE);
	CloseHandle(msg.evt);
	dec_msg_cnt(ioq);
	return 0;
}

static int poll_all_events(ioqueue_t *ioq, HANDLE *hevts, ioqueue_connect_overlapped_t **ovs, int i)
{
	int count = 0;
	int left = i;
	int pos = 0;
	while (left > 0) {
		DWORD idx = WaitForMultipleObjects(left, &hevts[pos], FALSE, 0) - WAIT_OBJECT_0;
		if (idx <= (DWORD)left && idx >= 0) {
			WSANETWORKEVENTS net_events;
			ioqueue_connect_overlapped_t *triggered = ovs[idx+pos];
			list_del(&triggered->node);
			WSAEnumNetworkEvents(triggered->base.handle_ctx->u.sock, hevts[idx+pos], &net_events);
			WSAEventSelect(triggered->base.handle_ctx->u.sock, hevts[idx+pos], 0);
			CloseHandle(hevts[idx+pos]);
			triggered->on_connect_callback(triggered->base.handle_ctx, (ioqueue_overlapped_t*)triggered, 
				triggered->base.user_data, net_events.iErrorCode[FD_CONNECT_BIT] == 0 ? 0 : -1);
			left -= (int)idx + 1;
			pos += idx+1;
			count ++;
		} else {
			break;
		}
	}
	return count;
}

static int poll_connect_list(ioqueue_t *ioq)
{
	int count = 0, i = 0;
	HANDLE hevts[MAXIMUM_WAIT_OBJECTS];
	ioqueue_connect_overlapped_t *ovs[MAXIMUM_WAIT_OBJECTS];
	ioqueue_connect_overlapped_t *pos, *n;

	list_for_each_entry_safe(pos, n, &ioq->connect_list, ioqueue_connect_overlapped_t, node) {
		hevts[i] = pos->hevt;
		ovs[i] = pos;
		i++;
		if (i == MAXIMUM_WAIT_OBJECTS) {
			count += poll_all_events(ioq, hevts, ovs, i);
			i = 0;
		}
	}
	if (i > 0) {
		count += poll_all_events(ioq, hevts, ovs, i);
	}
	return count;
}

TOOLKIT_API void* ioqueue_set_user_data(ioqueue_t* ioq, void* user_data)
{
	void* old;
	assert(ioq);
	old = ioq->user_data;
	ioq->user_data = user_data;
	return old;
}

TOOLKIT_API void* ioqueue_get_user_data(ioqueue_t* ioq)
{
	assert(ioq);
	return ioq->user_data;
}

TOOLKIT_API int ioqueue_poll(ioqueue_t* q, int timeout)
{
	ioqueue_t *ioq = (ioqueue_t*)q;
	int count = 0, t = 0;
	int nfds = 0, i = 0;
	bus_daemon_t* deamon;
	struct epoll_event events[MAX_EPOLL_EVENT];
	deamon = (bus_daemon_t*)ioqueue_get_user_data(ioq);
	assert(deamon != NULL);
	/* network and msg, dispatch until no events */
	do 
	{
		BOOL ret;
		DWORD dwBytesTransfer = 0;
		t = 0;
		//有时会出现惊群的问题！！
		nfds = evtpoll_wait(ioq->ep, events, MAX_EPOLL_EVENT, t ? 0 : timeout);
		if (nfds < 0) { 
			break; 
		}
		for (i = 0; i < nfds; ++i) {
			struct epoll_event* pe;
			int n = 1;
			pe = events + i;
			WLog_INFO(TAG, "poll events[%d]::fd(0x%08X) OUT:%d, IN:%d", i, pe->data.fd
				, pe->events & EPOLLOUT ? 1: 0 ,pe->events & EPOLLIN ? 1: 0);

			while (n >= 0) {
				ioqueue_overlapped_t* io_ctx = NULL;
				n = evtpoll_deal(ioq->ep, pe, &io_ctx, 1);
				if (!n) {
					assert(io_ctx);
					pre_dispatch_network(&ret, &dwBytesTransfer, io_ctx);
					dispatch_network(ret, dwBytesTransfer, io_ctx);
					t++;
					count++;
				}
			}
		}
	} while (t > 0);

	/* win2k connect event */
	if (!is_os_gte_xp()) {
		spinlock_enter(&ioq->connect_list_lock, -1);
		poll_connect_list(ioq);
		spinlock_leave(&ioq->connect_list_lock);
	}

	/* timer heap */
	spinlock_enter(&ioq->tm_queue_lock, -1);
	count += timer_queue_poll(ioq->tm_queue, NULL); /* dispatch timer heap */
	spinlock_leave(&ioq->tm_queue_lock);

	if (ioq->stop == -1) {
		if (InterlockedCompareExchange(&ioq->stop, -2, -1) == -1) { /* close all handler */
			ioqueue_handle_context *pos, *n;
			spinlock_enter(&ioq->handler_list_lock, -1);
			list_for_each_entry_safe(pos, n, &ioq->handler_list, ioqueue_handle_context, node) {
				if (pos->type != HANDLE_TYPE_FILE) {
					closesocket(pos->u.sock);
					pos->u.sock = INVALID_SOCKET;
				} else {
					CloseHandle(pos->u.file);
					pos->u.file = INVALID_HANDLE_VALUE;
				}
			}
			spinlock_leave(&ioq->handler_list_lock);
		}
	}

	return count;
}

TOOLKIT_API void ioqueue_stop(ioqueue_t *ioq)
{
	assert(ioq);
	ioq->stop = -1;
}

/* timer */

TOOLKIT_API int ioqueue_timer_schedule(ioqueue_t *ioq, timer_entry *entry, unsigned int delay)
{
	int err;
	
	assert(ioq);
	assert(entry);

	if (ioq->stop)
		return -1;

	spinlock_enter(&ioq->tm_queue_lock, -1);
	err = timer_queue_schedule(ioq->tm_queue, entry, delay);
	spinlock_leave(&ioq->tm_queue_lock);
	return err;
}

TOOLKIT_API int ioqueue_timer_cancel(ioqueue_t *ioq, timer_entry *entry, int cancel)
{
	int err;
	assert(ioq);
	assert(entry);
	spinlock_enter(&ioq->tm_queue_lock, -1);
	err = timer_queue_cancel(ioq->tm_queue, entry, cancel);
	spinlock_leave(&ioq->tm_queue_lock);
	return err;
}

/* acceptor */

TOOLKIT_API int ioqueue_acceptor_create(ioqueue_t *ioq, 
							const char *ip, 
							unsigned short port, 
							ioqueue_acceptor_t* acceptor)
{
	WLog_INFO(TAG, "==> %s", __FUNCTION__);
	struct sockaddr_in service = {0};
	assert(ioq);
	assert(acceptor);
	assert(port);

	if (ioq->stop) {
		WLog_INFO(TAG, "<== %s", __FUNCTION__);
		return -1;
	}
	memset(acceptor, 0, sizeof(ioqueue_acceptor_t));
	/*Warning: only the front third params are effective !!*/
	acceptor->u.sock = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
	if (acceptor->u.sock == INVALID_SOCKET) {
		WLog_ERR(TAG, "accept socket create failed: %s", strerror(errno));
		goto on_error;
	}
	nonblock_sock(acceptor->u.sock);
	service.sin_family = AF_INET;
	service.sin_port = htons(port);
	service.sin_addr.s_addr = ip ? inet_addr(ip) : htonl(INADDR_ANY);
	if (bind(acceptor->u.sock, (struct sockaddr*) & service, sizeof(struct sockaddr)) != 0) {
		WLog_ERR(TAG, "bind sockect %d failed: %s", acceptor->u.sock, strerror(errno));
		goto on_error;
	}
	if (evtpoll_attach(ioq->ep, acceptor->u.sock)) {
		goto on_error;
	}
	acceptor->type = HANDLE_TYPE_ACCEPTOR;
	acceptor->owner = ioq;
	fastlock_init(acceptor->ov_pending_list_lock);
	INIT_LIST_HEAD(&acceptor->ov_pending_list);
	add_handler_list(acceptor, ioq);
	inc_ref(ioqueue_handle_context, acceptor);
	WLog_INFO(TAG, "<== normal %s", __FUNCTION__);
	return 0;

on_error:
	if (acceptor->u.sock != INVALID_SOCKET)
		closesocket(acceptor->u.sock);
	WLog_INFO(TAG, "<== error %s", __FUNCTION__);
	return -1;
}

TOOLKIT_API int ioqueue_acceptor_listen(ioqueue_acceptor_t* acceptor, int backlog)
{
	assert(acceptor);
	return listen(acceptor->u.sock, backlog);
}

TOOLKIT_API void ioqueue_acceptor_destroy(ioqueue_acceptor_t* acceptor)
{
	assert(acceptor);
	dec_ref(ioqueue_handle_context, acceptor);
}

TOOLKIT_API void ioqueue_acceptor_close(ioqueue_acceptor_t* acceptor)
{
	SOCKET s;
	assert(acceptor);
	s = acceptor->u.sock;
	if (s != INVALID_SOCKET) {
		acceptor->u.sock = INVALID_SOCKET;
		closesocket(s);
	}
}

TOOLKIT_API int ioqueue_acceptor_async_accept(ioqueue_acceptor_t* acceptor, 
								  ioqueue_overlapped_t *ov, 
								  ioqueue_on_accept_callback on_accept_callback,
								  void *user_data)
{
	ioqueue_t *ioq;
	ioqueue_accept_overlapped_t *overlapped;
	DWORD bytesTransfer;
	BOOL ret;

	assert(acceptor);
	assert(ov);
	assert(acceptor->type == HANDLE_TYPE_ACCEPTOR);
	assert(on_accept_callback);

	ioq = acceptor->owner;
	if (ioq->stop)
		return -1;
	overlapped = (ioqueue_accept_overlapped_t*)ov;
	memset(overlapped, 0, sizeof(ioqueue_accept_overlapped_t));
	ioqueue_overlapped_set_mask(ov, sizeof(ioqueue_accept_overlapped_t));
	overlapped->client = INVALID_SOCKET;
	fastlock_enter(acceptor->ov_pending_list_lock);
	list_add_tail(&overlapped->base.pending_entry, &acceptor->ov_pending_list);
	fastlock_leave(acceptor->ov_pending_list_lock);
	overlapped->base.type = OV_ACCEPT;
	overlapped->base.user_data = user_data;
	overlapped->base.handle_ctx = acceptor;
	inc_pending_io(acceptor);
	overlapped->on_accept_callback = on_accept_callback;
	ret = evtpoll_subscribe(ioq->ep, EV_ACCEPT, acceptor->u.sock, ov, NULL);
	if (ret == 0) {
		WLog_INFO(TAG, "acceptor subscribes accept event succ.");
		return 0;
	}
	WLog_ERR(TAG, "acceptor: epoll_ctl failed error: %d ", ret);
	fastlock_enter(acceptor->ov_pending_list_lock);
	list_del(&overlapped->base.pending_entry);
	fastlock_leave(acceptor->ov_pending_list_lock);
	dec_pending_io(acceptor);
	return -1;
}

TOOLKIT_API int ioqueue_acceptor_accept(ioqueue_acceptor_t* acceptor, SOCKET *s, struct sockaddr *addr, int *addrlen, int timeout)
{
	struct timeval tm;
	fd_set set;
	fd_set ex_set;
	int rc;

	FD_ZERO(&set);
	FD_ZERO(&ex_set);
	FD_SET(acceptor->u.sock, &set);
	FD_SET(acceptor->u.sock, &ex_set);

	tm.tv_sec = timeout / 1000;
	tm.tv_usec = 1000 * (timeout % 1000);

	rc = select(acceptor->u.sock+1, &set, NULL, &ex_set, &tm);
	if (rc > 0) {
		if (FD_ISSET(acceptor->u.sock, &ex_set))
			return -1;
		if (FD_ISSET(acceptor->u.sock, &set)) {
			SOCKET fd = accept(acceptor->u.sock, addr, addrlen);
			if (fd != INVALID_SOCKET) {
				*s = fd;
				return 0;
			}
		}
	}

	return -1;
}

TOOLKIT_API int ioqueue_acceptor_create_client(ioqueue_acceptor_t* acceptor, SOCKET s, ioqueue_tcpsock_t *tcpsock)
{
	ioqueue_t *ioq;
	assert(acceptor);
	assert(tcpsock);
	assert(s != INVALID_SOCKET);
	ioq = acceptor->owner;
	if (ioq->stop)
		return -1;
	memset(tcpsock, 0, sizeof(ioqueue_tcpsock_t));
	tcpsock->type = HANDLE_TYPE_TCPSOCK;
	tcpsock->u.sock = s;
	tcpsock->owner = ioq;
	tcpsock->user_data = NULL;
	fastlock_init(tcpsock->ov_pending_list_lock);
	INIT_LIST_HEAD(&tcpsock->ov_pending_list);
	if (0 != evtpoll_attach(ioq->ep, s)) {
		return -1;
	}
	add_handler_list(tcpsock, ioq);
	inc_ref(ioqueue_handle_context, tcpsock);
	WLog_DBG(TAG, "create a tcp session, fd=%d", s);
	return 0;
}

TOOLKIT_API SOCKET ioqueue_acceptor_get_raw_socket(ioqueue_acceptor_t* acceptor)
{
	assert(acceptor);
	assert(acceptor->type == HANDLE_TYPE_ACCEPTOR);
	assert(acceptor->u.sock != INVALID_SOCKET);
	return acceptor->u.sock;
}

TOOLKIT_API ioqueue_t* ioqueue_acceptor_get_owned_ioqueue(ioqueue_acceptor_t* acceptor)
{
	assert(acceptor);
	assert(acceptor->type == HANDLE_TYPE_ACCEPTOR);
	return acceptor->owner;
}

TOOLKIT_API void *ioqueue_acceptor_set_user_data(ioqueue_acceptor_t* acceptor, void *user_data)
{
	void *old;
	assert(acceptor);
	assert(acceptor->type == HANDLE_TYPE_ACCEPTOR);
	old = acceptor->user_data;
	acceptor->user_data = user_data;
	return old;
}

TOOLKIT_API void *ioqueue_acceptor_get_user_data(ioqueue_acceptor_t* acceptor)
{
	assert(acceptor);
	assert(acceptor->type == HANDLE_TYPE_ACCEPTOR);
	return acceptor->user_data;
}

TOOLKIT_API int ioqueue_acceptor_cancel(ioqueue_acceptor_t* acceptor)
{
	assert(acceptor);
	return CancelIo(acceptor->u.file) ? 0 : -1;
}

/* tcpsock */

TOOLKIT_API int ioqueue_tcpsock_create(ioqueue_t *ioq, ioqueue_tcpsock_t *tcpsock)
{
	SOCKET s;

	assert(ioq);
	assert(tcpsock);

	if (ioq->stop)
		return -1;

	s = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
	if (s == INVALID_SOCKET)
		return -1;
	if (ioqueue_tcpsock_create_from_handle(ioq, s, tcpsock) != 0) {
		closesocket(s);
		return -1;
	}
	return 0;
}

TOOLKIT_API int ioqueue_tcpsock_create_from_handle(ioqueue_t *ioq, SOCKET s, ioqueue_tcpsock_t *tcpsock)
{
	assert(ioq);
	assert(s != INVALID_SOCKET);
	assert(tcpsock);

	if (ioq->stop)
		return -1;

	memset(tcpsock, 0, sizeof(ioqueue_tcpsock_t));
	tcpsock->u.sock = s;
	reuse_addr(tcpsock->u.sock);
	nonblock_sock(tcpsock->u.sock);
	/* winxp or more we use ConnectEx, this funtion need bind at first */
	if (is_os_gte_xp()) {
		struct sockaddr_in local = {0};
		local.sin_family = AF_INET;
		local.sin_port = htons(0);
		local.sin_addr.s_addr = INADDR_ANY;
		if (bind(tcpsock->u.sock, (struct sockaddr*)&local, sizeof(struct sockaddr)) != 0)
			return -1;
	} else {
		/* for win2k we use connect, set socket to non-block mode */
		//u_long ul_onoff = 1;
		//if (ioctlsocket(tcpsock->u.sock, FIONBIO, &ul_onoff) != 0)
		//	goto on_error;
	}
	if (!CreateIoCompletionPort((HANDLE)tcpsock->u.sock, ioq->iocp, 0, 0))
		return -1;
	fastlock_init(tcpsock->ov_pending_list_lock);
	INIT_LIST_HEAD(&tcpsock->ov_pending_list);
	tcpsock->type = HANDLE_TYPE_TCPSOCK;
	tcpsock->owner = ioq;
	add_handler_list(tcpsock, ioq);
	inc_ref(ioqueue_handle_context, tcpsock);

	return 0;
}

TOOLKIT_API int ioqueue_tcpsock_async_connect(ioqueue_tcpsock_t *tcpsock, 
								  ioqueue_overlapped_t *ov,
								  const char *ip, 
								  unsigned short port, 
								  ioqueue_on_connect_callback on_connect_callback,
								  void* user_data)
{
	ioqueue_t *ioq;
	ioqueue_connect_overlapped_t *overlapped;
	struct sockaddr_in service;

	assert(tcpsock);
	assert(ov);
	assert(ip);
	assert(port);
	assert(on_connect_callback);

	ioq = tcpsock->owner;
	if (ioq->stop)
		return -1;

	ioq = tcpsock->owner;
	overlapped = (ioqueue_connect_overlapped_t*)ov;
	memset(overlapped, 0, sizeof(ioqueue_connect_overlapped_t));
	ioqueue_overlapped_set_mask(ov, sizeof(ioqueue_connect_overlapped_t));
	fastlock_enter(tcpsock->ov_pending_list_lock);
	list_add_tail(&overlapped->base.pending_entry, &tcpsock->ov_pending_list);
	fastlock_leave(tcpsock->ov_pending_list_lock);
	overlapped->base.type = OV_CONNECT;
	overlapped->base.handle_ctx = (ioqueue_handle_context*)tcpsock;
	overlapped->base.user_data = user_data;
	overlapped->on_connect_callback = on_connect_callback;
	inc_pending_io(tcpsock);

	if (is_os_gte_xp()) { /* use ConnectEx */
		DWORD dwBytes;
		BOOL ret;
		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
			);
//		LPFN_CONNECTEX   lpfnConnectEx;
		GUID GuidConnectEx = WSAID_CONNECTEX;
		if (WSAIoctl(tcpsock->u.sock, SIO_GET_EXTENSION_FUNCTION_POINTER, &GuidConnectEx, 
			sizeof(GuidConnectEx), &lpfnConnectEx, sizeof(lpfnConnectEx), &dwBytes, NULL, NULL) != 0) {
				fastlock_enter(tcpsock->ov_pending_list_lock);
				list_del(&overlapped->base.pending_entry);
				fastlock_leave(tcpsock->ov_pending_list_lock);
				dec_pending_io(tcpsock);
				return -1;
		}
		memset(&service, 0, sizeof(service));
		service.sin_family = AF_INET;
		service.sin_port = htons(port);
		service.sin_addr.s_addr = inet_addr(ip);
		{
			struct sockaddr_in local_addr = {0}; // bind to a INADDR_ANY and port 0 to let OS choose an local address
			local_addr.sin_family = AF_INET;
			local_addr.sin_addr.s_addr = htonl(INADDR_ANY);
			local_addr.sin_port = htons(0);
			ret = bind(tcpsock->u.sock, (SOCKADDR*)&local_addr, sizeof(local_addr)); // caution: ConnectEx need socket to be bounded at first
		}
		if (ret == 0) {
			ret = lpfnConnectEx(tcpsock->u.sock, (struct sockaddr*)&service, sizeof(service), NULL, 
				0, NULL, &overlapped->base.ov);
			if (ret || WSAGetLastError() == WSA_IO_PENDING)
				return 0;
		}
	} else { /* use non-blocking connect */
		overlapped->hevt = WSACreateEvent();
		if (WSAEventSelect(tcpsock->u.sock, overlapped->hevt, FD_CONNECT) == 0) {
			spinlock_enter(&ioq->connect_list_lock, -1);
			list_add_tail(&overlapped->node, &ioq->connect_list);
			spinlock_leave(&ioq->connect_list_lock);
			if (connect(tcpsock->u.sock, (struct sockaddr*)&service, sizeof(service)) == 0) {
				return 0;
			} else {
				spinlock_enter(&ioq->connect_list_lock, -1);
				list_del(&overlapped->node);
				spinlock_leave(&ioq->connect_list_lock);
			}
		}
		WSACloseEvent(overlapped->hevt);
	}
	fastlock_enter(tcpsock->ov_pending_list_lock);
	list_del(&overlapped->base.pending_entry);
	fastlock_leave(tcpsock->ov_pending_list_lock);
	dec_pending_io(tcpsock);

	return -1;
}

TOOLKIT_API int ioqueue_tcpsock_conect(ioqueue_tcpsock_t *tcpsock, 
						   const char *ip, 
						   unsigned short port, 
						   int timeout)
{
	fd_set wr_set;
	fd_set ex_set;
	struct timeval tm;

	assert(tcpsock);
	assert(ip);
	assert(port > 0);

	FD_ZERO(&wr_set);
	FD_ZERO(&ex_set);

	FD_SET(tcpsock->u.sock, &wr_set);
	FD_SET(tcpsock->u.sock, &ex_set);

	tm.tv_sec = timeout / 1000;
	tm.tv_usec = 1000 * (timeout % 1000);

	if (select(tcpsock->u.sock+1, NULL, &wr_set, &ex_set, &tm) > 0) {
		if (FD_ISSET(tcpsock->u.sock, &ex_set))
			return -1;
		if (FD_ISSET(tcpsock->u.sock, &wr_set))
			return 0;
	}

	return -1;
}


TOOLKIT_API int ioqueue_tcpsock_async_sendsome(ioqueue_tcpsock_t *tcpsock, 
								   ioqueue_overlapped_t *ov,
								   void *buf, 
								   unsigned int len, 
								   ioqueue_on_send_callback on_send_callback,
								   void *user_data)
{
	ioqueue_sendsome_overlapped_t *overlapped;
	DWORD bytesWritten;
	int rc;
	ioqueue_t *ioq;

	assert(tcpsock);
	assert(ov);
	assert(buf);
	assert(on_send_callback);

	ioq = ioqueue_tcpsock_get_owned_ioqueue(tcpsock);
	if (ioq->stop)
		return -1;

	overlapped = (ioqueue_sendsome_overlapped_t*)ov;
	memset(overlapped, 0, sizeof(ioqueue_sendsome_overlapped_t));
	ioqueue_overlapped_set_mask(ov, sizeof(ioqueue_sendsome_overlapped_t));
	fastlock_enter(tcpsock->ov_pending_list_lock);
	list_add_tail(&overlapped->base.pending_entry, &tcpsock->ov_pending_list);
	fastlock_leave(tcpsock->ov_pending_list_lock);
	overlapped->base.type = OV_SENDSOME;
	overlapped->base.handle_ctx = (ioqueue_handle_context*)tcpsock;
	overlapped->base.user_data = user_data;
	overlapped->on_send_callback = on_send_callback;
	overlapped->wsabuf.len = len;
	overlapped->wsabuf.buf = buf;
	inc_pending_io(tcpsock);
	rc = WSASend(tcpsock->u.sock, &overlapped->wsabuf, 1, &bytesWritten, 
		0, &overlapped->base.ov, NULL);
	if (rc == 0 || WSAGetLastError() == WSA_IO_PENDING)
		return 0;
	fastlock_enter(tcpsock->ov_pending_list_lock);
	list_del(&overlapped->base.pending_entry);
	fastlock_leave(tcpsock->ov_pending_list_lock);
	dec_pending_io(tcpsock);
	return -1;
}

TOOLKIT_API int ioqueue_tcpsock_async_sendn(ioqueue_tcpsock_t *tcpsock, 
								ioqueue_overlapped_t *ov,
								void *buf, 
								unsigned int len,
								ioqueue_on_send_callback on_send_callback,
								void* user_data)
{
	ioqueue_sendn_overlapped_t *overlapped;
	DWORD bytesWritten;
	int rc;
	ioqueue_t *ioq;

	assert(tcpsock);
	assert(ov);
	assert(buf);
	assert(on_send_callback);

	ioq = ioqueue_tcpsock_get_owned_ioqueue(tcpsock);
	if (ioq->stop)
		return -1;
	WLog_INFO(TAG, "fd(%d): ioqueue_tcpsock_async_sendn fired:  buf len: %d", tcpsock->u.sock, len);
	overlapped = (ioqueue_sendn_overlapped_t*)ov;
	memset(overlapped, 0, sizeof(ioqueue_sendn_overlapped_t));
	ioqueue_overlapped_set_mask(ov, sizeof(ioqueue_sendn_overlapped_t));
	fastlock_enter(tcpsock->ov_pending_list_lock);
	list_add_tail(&overlapped->base.pending_entry, &tcpsock->ov_pending_list);
	fastlock_leave(tcpsock->ov_pending_list_lock);
	overlapped->base.type = OV_SENDN;
	overlapped->base.handle_ctx = (ioqueue_handle_context*)tcpsock;
	overlapped->base.user_data = user_data;
	overlapped->on_send_callback = on_send_callback;
	overlapped->wsabuf.len = len;
	overlapped->wsabuf.buf = buf;
	overlapped->original_buf = buf;
	overlapped->sended_bytes = 0;
	overlapped->total_bytes = len;
	inc_pending_io(tcpsock);
	rc = evtpoll_subscribe(ioq->ep, EV_WRITE, tcpsock->u.sock, NULL, ov);
	if (!rc) {
		return 0;
	}
	WLog_ERR(TAG, "epoll_ctl for write failed return: %d", rc);
	fastlock_enter(tcpsock->ov_pending_list_lock);
	list_del(&overlapped->base.pending_entry);
	fastlock_leave(tcpsock->ov_pending_list_lock);
	dec_pending_io(tcpsock);
	return -1;
}

TOOLKIT_API int ioqueue_tcpsock_async_senduntil(ioqueue_tcpsock_t *tcpsock, 
									ioqueue_overlapped_t *ov,
									void *buf, 
									unsigned int len,
									const char *delimer,
									ioqueue_on_send_callback on_send_cb,
									void* user_data)
{
	const char *p;

	assert(tcpsock);
	assert(ov);
	assert(buf);
	assert(on_send_cb);
	assert(delimer);
	
	p = memstr(buf, len, delimer);
	if (!p)
		return -1;
	p += strlen(delimer);

	return ioqueue_tcpsock_async_sendn(tcpsock, ov, buf, p - (char*)buf, on_send_cb, user_data);
}

TOOLKIT_API int ioqueue_tcpsock_sendsome(ioqueue_tcpsock_t *tcpsock, 
							 void *buf, 
							 unsigned int len,
							 int timeout)
{
	assert(tcpsock);
	return send(tcpsock->u.sock, buf, len, 0);
}

TOOLKIT_API int ioqueue_tcpsock_sendn(ioqueue_tcpsock_t *tcpsock, 
						  void *buf, 
						  unsigned int len,
						  int timeout)
{
	return tsend_n(tcpsock->u.sock, buf, len, timeout);
}

TOOLKIT_API int ioqueue_tcpsock_senduntil(ioqueue_tcpsock_t *tcpsock, 
							  void *buf, 
							  unsigned int len, 
							  const char *delimer, 
							  int timeout)
{
	return tsend_until(tcpsock->u.sock, buf, len, delimer, timeout);
}

TOOLKIT_API int ioqueue_tcpsock_async_recvsome(ioqueue_tcpsock_t *tcpsock, 
								   ioqueue_overlapped_t *ov,
								   void *buf, 
								   unsigned int len, 
								   ioqueue_on_recv_callback on_recv_callback,
								   void *user_data)
{
	ioqueue_recvsome_overlapped_t *overlapped;
	DWORD bytesRead;
	int rc;
	ioqueue_t *ioq;

	assert(tcpsock);
	assert(ov);
	assert(buf);
	assert(on_recv_callback);

	ioq = tcpsock->owner;
	if (ioq->stop)
		return -1;

	overlapped = (ioqueue_recvsome_overlapped_t*)ov;
	memset(overlapped, 0, sizeof(ioqueue_recvsome_overlapped_t));
	ioqueue_overlapped_set_mask(ov, sizeof(ioqueue_recvsome_overlapped_t));
	fastlock_enter(tcpsock->ov_pending_list_lock);
	list_add_tail(&overlapped->base.pending_entry, &tcpsock->ov_pending_list);
	fastlock_leave(tcpsock->ov_pending_list_lock);
	overlapped->base.type = OV_RECVSOME;
	overlapped->base.handle_ctx = (ioqueue_handle_context*)tcpsock;
	overlapped->base.user_data = user_data;
	overlapped->on_recv_callback = on_recv_callback;
	overlapped->wsabuf.len = len;
	overlapped->wsabuf.buf = buf;
	overlapped->dwFlags = 0;
	inc_pending_io(tcpsock);
	rc = WSARecv(tcpsock->u.sock, &overlapped->wsabuf, 1, &bytesRead, &overlapped->dwFlags, 
			&overlapped->base.ov, NULL);
	if (rc == 0 || WSAGetLastError() == WSA_IO_PENDING)
		return 0;
	fastlock_enter(tcpsock->ov_pending_list_lock);
	list_del(&overlapped->base.pending_entry);
	fastlock_leave(tcpsock->ov_pending_list_lock);
	dec_pending_io(tcpsock);
	return -1;
}

TOOLKIT_API int ioqueue_tcpsock_async_recvn(ioqueue_tcpsock_t *tcpsock, 
								ioqueue_overlapped_t *ov,
								void *buf, 
								unsigned int len, 
								ioqueue_on_recv_callback on_recv_callback,
								void *user_data)
{
	ioqueue_recvn_overlapped_t *overlapped;
	DWORD bytesRead;
	int rc;
	ioqueue_t *ioq;

	assert(tcpsock);
	assert(ov);
	assert(buf);
	assert(on_recv_callback);

	ioq = tcpsock->owner;
	if (ioq->stop)
		return -1;

	overlapped = (ioqueue_recvn_overlapped_t*)ov;
	memset(overlapped, 0, sizeof(ioqueue_recvn_overlapped_t));
	ioqueue_overlapped_set_mask(ov, sizeof(ioqueue_recvn_overlapped_t));
	fastlock_enter(tcpsock->ov_pending_list_lock);
	list_add_tail(&overlapped->base.pending_entry, &tcpsock->ov_pending_list);
	fastlock_leave(tcpsock->ov_pending_list_lock);
	overlapped->base.type = OV_RECVN;
	overlapped->base.handle_ctx = (ioqueue_handle_context*)tcpsock;
	overlapped->base.user_data = user_data;
	overlapped->on_recv_callback = on_recv_callback;
	overlapped->wsabuf.len = len;
	overlapped->wsabuf.buf = buf;
	overlapped->original_buf = buf;
	overlapped->recved_bytes = 0;
	overlapped->total_bytes = len;
	overlapped->dwFlags = 0;
	inc_pending_io(tcpsock);
	rc = evtpoll_subscribe(ioq->ep, EV_READ, tcpsock->u.sock, ov, NULL);
	if (!rc) {
		return 0;
	}
	WLog_ERR(TAG, "epoll_ctl for read failed return: %d", rc);
	fastlock_enter(tcpsock->ov_pending_list_lock);
	list_del(&overlapped->base.pending_entry);
	fastlock_leave(tcpsock->ov_pending_list_lock);
	dec_pending_io(tcpsock);
	return -1;
}

TOOLKIT_API int ioqueue_tcpsock_async_recvuntil(ioqueue_tcpsock_t *tcpsock, 
									ioqueue_overlapped_t *ov,
									void *buf, 
									unsigned int len,
									const char *delimer,
									ioqueue_on_recvuntil_callback on_recvuntil_callback,
									void *user_data)
{
	ioqueue_recvuntil_overlapped_t *overlapped;
	DWORD bytesRead;
	int rc;
	ioqueue_t *ioq;

	assert(tcpsock);
	assert(ov);
	assert(buf);
	assert(delimer);
	assert(on_recvuntil_callback);

	ioq = tcpsock->owner;
	if (ioq->stop)
		return -1;

	overlapped = (ioqueue_recvuntil_overlapped_t*)ov;
	memset(overlapped, 0, sizeof(ioqueue_recvuntil_overlapped_t));
	ioqueue_overlapped_set_mask(ov, sizeof(ioqueue_recvuntil_overlapped_t));
	fastlock_enter(tcpsock->ov_pending_list_lock);
	list_add_tail(&overlapped->base.pending_entry, &tcpsock->ov_pending_list);
	fastlock_leave(tcpsock->ov_pending_list_lock);
	overlapped->base.type = OV_RECVUNTIL;
	overlapped->base.handle_ctx = (ioqueue_handle_context*)tcpsock;
	overlapped->base.user_data = user_data;
	overlapped->on_recvuntil_callback = on_recvuntil_callback;
	overlapped->wsabuf.len = len;
	overlapped->wsabuf.buf = buf;
	overlapped->original_buf = buf;
	overlapped->recved_bytes = 0;
	overlapped->total_bytes = len;
	overlapped->delimer = _strdup(delimer);
	overlapped->dwFlags = 0;
	inc_pending_io(tcpsock);
	rc = WSARecv(tcpsock->u.sock, &overlapped->wsabuf, 1, &bytesRead, &overlapped->dwFlags, 
			&overlapped->base.ov, NULL);
	if (rc == 0 || WSAGetLastError() == WSA_IO_PENDING)
		return 0;
	fastlock_enter(tcpsock->ov_pending_list_lock);
	list_del(&overlapped->base.pending_entry);
	fastlock_leave(tcpsock->ov_pending_list_lock);
	dec_pending_io(tcpsock);
	return -1;
}

TOOLKIT_API int ioqueue_tcpsock_recvsome(ioqueue_tcpsock_t *tcpsock, 
							 void *buf, 
							 unsigned int len, 
							 int timeout)
{
	return recv(tcpsock->u.sock, buf, len, 0);
}

TOOLKIT_API int ioqueue_tcpsock_recvn(ioqueue_tcpsock_t *tcpsock, 
						  void *buf, 
						  unsigned int len, 
						  int timeout)
{
	return trecv_n(tcpsock->u.sock, buf, len, timeout);
}

TOOLKIT_API int ioqueue_tcpsock_recvuntil(ioqueue_tcpsock_t *tcpsock, 
							  void *buf, 
							  unsigned int len, 
							  const char *delimer, 
							  unsigned int *header_len,
							  int timeout)
{
	return trecv_until(tcpsock->u.sock, buf, len, delimer, header_len, timeout);
}

TOOLKIT_API void ioqueue_tcpsock_close(ioqueue_tcpsock_t *tcpsock)
{
	SOCKET s;
	assert(tcpsock);
	s = tcpsock->u.sock;
	if (s != INVALID_SOCKET) {
		tcpsock->u.sock = INVALID_SOCKET;
		closesocket(s);
	}
}

TOOLKIT_API void ioqueue_tcpsock_destroy(ioqueue_tcpsock_t *tcpsock)
{
	assert(tcpsock);
	dec_ref(ioqueue_handle_context, tcpsock);
}

TOOLKIT_API int ioqueue_tcpsock_shutdown(ioqueue_tcpsock_t *tcpsock, int how)
{
	assert(tcpsock);
	return shutdown(tcpsock->u.sock, how);
}

TOOLKIT_API SOCKET ioqueue_tcpsock_get_raw_socket(ioqueue_tcpsock_t *tcpsock)
{
	assert(tcpsock);
	return tcpsock->u.sock;
}

TOOLKIT_API ioqueue_t* ioqueue_tcpsock_get_owned_ioqueue(ioqueue_tcpsock_t *tcpsock)
{
	assert(tcpsock);
	return tcpsock->owner;
}

TOOLKIT_API void *ioqueue_tcpsock_set_user_data(ioqueue_tcpsock_t *tcpsock, void *user_data)
{
	void *old;
	assert(tcpsock);
	old = tcpsock->user_data;
	tcpsock->user_data = user_data;
	return old;
}

TOOLKIT_API void *ioqueue_tcpsock_get_user_data(ioqueue_tcpsock_t *tcpsock)
{
	assert(tcpsock);
	return tcpsock->user_data;
}

TOOLKIT_API int ioqueue_tcpsock_cancel(ioqueue_tcpsock_t* tcpsock)
{
	assert(tcpsock);
	return CancelIo(tcpsock->u.file) ? 0 : -1;
}

/* udpsock */

TOOLKIT_API int ioqueue_udpsock_create(ioqueue_t *ioq, ioqueue_udpsock_t *udpsock)
{
	SOCKET s;
	
	assert(ioq);
	assert(udpsock);

	if (ioq->stop)
		return -1;

	s = WSASocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP, NULL, 0, WSA_FLAG_OVERLAPPED);
	if (s == INVALID_SOCKET) 
		return -1;
	if (ioqueue_udpsock_create_from_handle(ioq, s, udpsock) != 0) {
		closesocket(s);
		return -1;
	}
	return 0;
}

TOOLKIT_API int ioqueue_udpsock_create_from_handle(ioqueue_t *ioq, SOCKET s, ioqueue_udpsock_t *udpsock)
{
	assert(ioq);
	assert(udpsock);
	assert(s != INVALID_SOCKET);

	if (ioq->stop)
		return -1;

	memset(udpsock, 0, sizeof(ioqueue_udpsock_t));
	udpsock->u.sock = s;
	nonblock_sock(udpsock->u.sock);
	if (!CreateIoCompletionPort((HANDLE)udpsock->u.sock, ioq->iocp, 0, 0))
		return -1;
	fastlock_init(udpsock->ov_pending_list_lock);
	INIT_LIST_HEAD(&udpsock->ov_pending_list);
	udpsock->type = HANDLE_TYPE_UDPSOCK;
	udpsock->owner = ioq;
	add_handler_list(udpsock, ioq);
	inc_ref(ioqueue_handle_context, udpsock);

	return 0;
}

TOOLKIT_API void ioqueue_udpsock_close(ioqueue_udpsock_t *udpsock)
{
	SOCKET s;
	assert(udpsock);
	s = udpsock->u.sock;
	if (s != INVALID_SOCKET) {
		udpsock->u.sock = INVALID_SOCKET;
		closesocket(s);
	}
}

TOOLKIT_API void ioqueue_udpsock_destroy(ioqueue_udpsock_t *udpsock)
{
	assert(udpsock);
	dec_ref(ioqueue_handle_context, udpsock);
}

TOOLKIT_API int ioqueue_udpsock_async_sendto(ioqueue_udpsock_t* udpsock, 
								 ioqueue_overlapped_t *ov,
								 void *buf, 
								 int len, 
								 const struct sockaddr* to, 
								 int tolen,
								 ioqueue_on_sendto_callback on_sendto_callback,
								 void *user_data)
{
	ioqueue_sendto_overlapped_t *overlapped;
	int rc;
	DWORD bytesWritten;
	ioqueue_t *ioq;

	assert(udpsock);
	assert(ov);
	assert(buf);
	assert(to);
	assert(on_sendto_callback);
	assert(ov);

	ioq = udpsock->owner;
	if (ioq->stop)
		return -1;

	overlapped = (ioqueue_sendto_overlapped_t*)ov;
	memset(overlapped, 0, sizeof(ioqueue_sendto_overlapped_t));
	ioqueue_overlapped_set_mask(ov, sizeof(ioqueue_sendto_overlapped_t));
	fastlock_enter(udpsock->ov_pending_list_lock);
	list_add_tail(&overlapped->base.pending_entry, &udpsock->ov_pending_list);
	fastlock_leave(udpsock->ov_pending_list_lock);
	overlapped->base.type = OV_SENDTO;
	overlapped->base.user_data = user_data;
	overlapped->base.handle_ctx = (ioqueue_handle_context*)udpsock;
	overlapped->on_sendto_callback = on_sendto_callback;
	overlapped->wsabuf.len = len;
	overlapped->wsabuf.buf = buf;
	inc_pending_io(udpsock);
	rc = WSASendTo(udpsock->u.sock, &overlapped->wsabuf, 1, &bytesWritten, 0,
			to, tolen, &overlapped->base.ov, NULL);
	if (rc == 0 || WSAGetLastError() == WSA_IO_PENDING)
		return 0;
	fastlock_enter(udpsock->ov_pending_list_lock);
	list_del(&overlapped->base.pending_entry);
	fastlock_leave(udpsock->ov_pending_list_lock);
	dec_pending_io(udpsock);
	return -1;
}

TOOLKIT_API int ioqueue_udpsock_sendto(ioqueue_udpsock_t *udpsock, 
						   void *buf, 
						   int len, 
						   const struct sockaddr* to, 
						   int tolen,
						   int timeout)
{
	return tsendto(udpsock->u.sock, buf, len, to, tolen, timeout);
}

TOOLKIT_API int ioqueue_udpsock_async_recvfrom(ioqueue_udpsock_t* udpsock, 
								   ioqueue_overlapped_t *ov,
								   void* buf, 
								   int len, 
								   ioqueue_on_recvfrom_callback on_recvfrom_callback,
								   void *user_data)
{
	ioqueue_recvfrom_overlapped_t *overlapped;
	int rc;
	DWORD bytesRead;
	ioqueue_t *ioq;

	assert(udpsock);
	assert(ov);
	assert(buf);
	assert(on_recvfrom_callback);

	ioq = udpsock->owner;
	if (ioq->stop)
		return -1;

	overlapped = (ioqueue_recvfrom_overlapped_t*)ov;
	memset(overlapped, 0, sizeof(ioqueue_recvfrom_overlapped_t));
	ioqueue_overlapped_set_mask(ov, sizeof(ioqueue_recvfrom_overlapped_t));
	fastlock_enter(udpsock->ov_pending_list_lock);
	list_add_tail(&overlapped->base.pending_entry, &udpsock->ov_pending_list);
	fastlock_leave(udpsock->ov_pending_list_lock);
	overlapped->base.type = OV_RECVFROM;
	overlapped->base.user_data = user_data;
	overlapped->base.handle_ctx = (ioqueue_handle_context*)udpsock;
	overlapped->on_recvfrom_callback = on_recvfrom_callback;
	overlapped->wsabuf.len = len;
	overlapped->wsabuf.buf = buf;
	overlapped->dwFlags = 0;
	inc_pending_io(udpsock);
	rc = WSARecvFrom(udpsock->u.sock, &overlapped->wsabuf, 1, &bytesRead, &overlapped->dwFlags,
			(struct sockaddr*)&overlapped->peer, &overlapped->addrlen, &overlapped->base.ov, NULL);
	if (rc == 0 || WSAGetLastError() == WSA_IO_PENDING)
		return 0;
	fastlock_enter(udpsock->ov_pending_list_lock);
	list_del(&overlapped->base.pending_entry);
	fastlock_leave(udpsock->ov_pending_list_lock);
	dec_pending_io(udpsock);
	return -1;
}

TOOLKIT_API int ioqueue_udpsock_recvfrom(ioqueue_udpsock_t* udpsock, 
							 ioqueue_overlapped_t *overlapped,
							 void* buf, 
							 int len,
							 struct sockaddr *fromaddr,
							 int *addrlen,
							 int timeout)
{
	return trecvfrom(udpsock->u.sock, buf, len, fromaddr, addrlen, timeout);
}

TOOLKIT_API SOCKET ioqueue_udpsock_get_raw_socket(ioqueue_udpsock_t *udpsock)
{
	assert(udpsock);
	return udpsock->u.sock;
}

TOOLKIT_API ioqueue_t* ioqueue_udpsock_get_owned_ioqueue(ioqueue_udpsock_t *udpsock)
{
	assert(udpsock);
	return udpsock->owner;
}

TOOLKIT_API void *ioqueue_udpsock_set_user_data(ioqueue_udpsock_t *udpsock, void *user_data)
{
	void *old;
	assert(udpsock);
	old = udpsock->user_data;
	udpsock->user_data = user_data;
	return old;
}

TOOLKIT_API void *ioqueue_udpsock_get_user_data(ioqueue_udpsock_t *udpsock)
{
	assert(udpsock);
	return udpsock->user_data;
}

TOOLKIT_API int ioqueue_udpsock_cancel(ioqueue_udpsock_t *udpsock)
{
	assert(udpsock);
	return CancelIo(udpsock->u.file) ? 0 : -1;
}

/* file */

TOOLKIT_API int ioqueue_file_create(ioqueue_t *ioq,
						const char *path, 
						DWORD dwDesiredAccess,
						DWORD dwShareMode,
						DWORD dwCreationDisposition, 
						DWORD dwFlagsAndAttributes,
						ioqueue_file_t *file)
{
	HANDLE hFile;

	assert(ioq);
	assert(path);
	assert(file);

	if (ioq->stop)
		return -1;

	hFile = CreateFileA(path, dwDesiredAccess, dwShareMode, 
		NULL, dwCreationDisposition, 
		dwFlagsAndAttributes|FILE_FLAG_OVERLAPPED, NULL);

	return ioqueue_file_create_from_handle(ioq, hFile, file);
}

TOOLKIT_API int ioqueue_file_create_from_handle(ioqueue_t *ioq, HANDLE h, ioqueue_file_t *file)
{
	assert(ioq);
	assert(file);

	if (ioq->stop)
		return -1;

	memset(file, 0, sizeof(ioqueue_file_t));
	file->u.file = h;
	if (file->u.file == INVALID_HANDLE_VALUE)
		return -1;
	if (!CreateIoCompletionPort(file->u.file, ioq->iocp, 0, 0)) {
		CloseHandle(file->u.file);
		file->u.file = INVALID_HANDLE_VALUE;
		return -1;
	}
	fastlock_init(file->ov_pending_list_lock);
	INIT_LIST_HEAD(&file->ov_pending_list);
	file->type = HANDLE_TYPE_FILE;
	file->owner = ioq;
	add_handler_list(file, ioq);
	inc_ref(ioqueue_handle_context, file);
	return 0;
}

TOOLKIT_API void ioqueue_file_close(ioqueue_file_t* file)
{
	HANDLE s;
	assert(file);
	s = file->u.file;
	if (s != INVALID_HANDLE_VALUE) {
		file->u.file = INVALID_HANDLE_VALUE;
		CloseHandle(s);
	}
}

TOOLKIT_API void ioqueue_file_destroy(ioqueue_file_t* file)
{
	assert(file);
	dec_ref(ioqueue_handle_context, file);
}

TOOLKIT_API int ioqueue_file_async_readsome(ioqueue_file_t* file, 
								ioqueue_overlapped_t *ov,
								void *buf, 
								unsigned int len, 
								ioqueue_on_read_callback on_read_callback,
								void *user_data)
{
	return ioqueue_file_async_readsome_at(file, ov, buf, len, 0, 0, on_read_callback, user_data);
}

TOOLKIT_API int ioqueue_file_async_readn(ioqueue_file_t* file, 
							 ioqueue_overlapped_t *overlapped,
							 void *buf, 
							 unsigned int len, 
							 ioqueue_on_read_callback on_read_cb,
							 void *user_data)
{
	return ioqueue_file_async_readn_at(file, overlapped, buf, len, 0, 0, on_read_cb, user_data);
}

TOOLKIT_API int ioqueue_file_readsome(ioqueue_file_t *file, void *buf, unsigned int len)
{
	return ioqueue_file_readsome_at(file, buf, len, 0, 0);
}

TOOLKIT_API int ioqueue_file_readn(ioqueue_file_t *file, void *buf, unsigned int len)
{
	return ioqueue_file_readn_at(file, buf, len, 0, 0);
}

TOOLKIT_API int ioqueue_file_async_readsome_at(ioqueue_file_t* file, 
								   ioqueue_overlapped_t *ov,
								   void *buf, 
								   unsigned int len, 
								   DWORD posLow, 
								   DWORD posHigh, 
								   ioqueue_on_read_callback on_read_callback,
								   void *user_data)
{
	ioqueue_readfilesome_overlapped_t *overlapped;
	BOOL rc;
	ioqueue_t *ioq;

	assert(file);
	assert(ov);
	assert(buf);
	assert(on_read_callback); 

	ioq = file->owner;
	if (ioq->stop)
		return -1;

	overlapped = (ioqueue_readfilesome_overlapped_t*)ov;
	memset(overlapped, 0, sizeof(ioqueue_readfilesome_overlapped_t));
	ioqueue_overlapped_set_mask(ov, sizeof(ioqueue_readfilesome_overlapped_t));
	fastlock_enter(file->ov_pending_list_lock);
	list_add_tail(&overlapped->base.pending_entry, &file->ov_pending_list);
	fastlock_leave(file->ov_pending_list_lock);
	overlapped->hevt = CreateEventA(NULL, TRUE, FALSE, NULL);
	overlapped->base.type = OV_READFILESOME;
	overlapped->base.user_data = user_data;
	overlapped->base.handle_ctx = (ioqueue_handle_context*)file;
	overlapped->base.ov.Offset = posLow;
	overlapped->base.ov.OffsetHigh = posHigh;
	overlapped->on_read_callback = on_read_callback;
	overlapped->buf = buf;
	inc_pending_io(file);
	rc = ReadFile(file->u.file, buf, (DWORD)len, NULL, &overlapped->base.ov);
	if (rc || GetLastError() == ERROR_IO_PENDING)
		return 0;
	fastlock_enter(file->ov_pending_list_lock);
	list_del(&overlapped->base.pending_entry);
	fastlock_leave(file->ov_pending_list_lock);
	dec_pending_io(file);
	CloseHandle(overlapped->hevt);
	return -1;
}

TOOLKIT_API int ioqueue_file_async_readn_at(ioqueue_file_t* file, 
								ioqueue_overlapped_t *ov,
								void *buf, 
								unsigned int len, 
								DWORD posLow, 
								DWORD posHigh, 
								ioqueue_on_read_callback on_read_cb,
								void *user_data)
{
	ioqueue_readfilen_overlapped_t *overlapped;
	BOOL rc;
	ioqueue_t *ioq;

	assert(file);
	assert(ov);
	assert(buf);
	assert(on_read_cb); 

	ioq = file->owner;
	if (ioq->stop)
		return -1;

	overlapped = (ioqueue_readfilen_overlapped_t*)ov;
	memset(overlapped, 0, sizeof(ioqueue_readfilen_overlapped_t));
	ioqueue_overlapped_set_mask(ov, sizeof(ioqueue_readfilen_overlapped_t));
	fastlock_enter(file->ov_pending_list_lock);
	list_add_tail(&overlapped->base.pending_entry, &file->ov_pending_list);
	fastlock_leave(file->ov_pending_list_lock);
	overlapped->hevt = CreateEventA(NULL, TRUE, FALSE, NULL);
	overlapped->base.type = OV_READFILEN;
	overlapped->base.user_data = user_data;
	overlapped->base.handle_ctx = (ioqueue_handle_context*)file;
	overlapped->base.ov.Offset = posLow;
	overlapped->base.ov.OffsetHigh = posHigh;
	overlapped->on_read_callback = on_read_cb;
	overlapped->buf = buf;
	overlapped->recved_bytes = 0;
	overlapped->total_bytes = len;
	inc_pending_io(file);
	rc = ReadFile(file->u.file, buf, (DWORD)len, NULL, &overlapped->base.ov);
	if (rc || GetLastError() == ERROR_IO_PENDING)
		return 0;
	fastlock_enter(file->ov_pending_list_lock);
	list_del(&overlapped->base.pending_entry);
	fastlock_leave(file->ov_pending_list_lock);
	dec_pending_io(file);
	return -1;
}

TOOLKIT_API int ioqueue_file_readsome_at(ioqueue_file_t *file, 
							 void *buf, 
							 unsigned int len, 
							 DWORD posLow, 
							 DWORD posHigh)
{
	OVERLAPPED ov;
	BOOL ret;
	DWORD dwTransferBytes;
	int rc = -1;

	/* (MSDN)
	Even if you have passed the function a file handle associated with a completion port and 
	a valid OVERLAPPED structure, an application can prevent completion port notification. 
	This is done by specifying a valid event handle for the hEvent member of the OVERLAPPED  structure, 
	and setting its low-order bit. A valid event handle whose low-order bit is set keeps I/O completion 
	from being queued to the completion port.
	*/

	memset(&ov, 0, sizeof(ov));
	ov.Offset = posLow;
	ov.OffsetHigh = posHigh;
	ov.hEvent = CreateEventA(NULL, TRUE, FALSE, NULL);
	ov.hEvent = (HANDLE)((DWORD)ov.hEvent & 0x1);

	ret = ReadFile(file->u.file, buf, len, &dwTransferBytes, &ov);

	if (!ret && GetLastError() == ERROR_IO_PENDING) {
		ret = GetOverlappedResult(file->u.file, &ov, &dwTransferBytes, TRUE);
	}

	CloseHandle((HANDLE)((DWORD)ov.hEvent & ~1));

	if (ret && dwTransferBytes > 0)
		rc = dwTransferBytes;

	return rc;
}

TOOLKIT_API int ioqueue_file_readn_at(ioqueue_file_t *file, 
						  void *buf, 
						  unsigned int len, 
						  DWORD posLow, 
						  DWORD posHigh)
{
	OVERLAPPED ov;
	int rc = 0;
	DWORD left = len;
	DWORD offset = 0;

	memset(&ov, 0, sizeof(ov));
	ov.Offset = posLow;
	ov.OffsetHigh = posHigh;
	ov.hEvent = CreateEventA(NULL, TRUE, FALSE, NULL);
	ov.hEvent = (HANDLE)((DWORD)ov.hEvent & 0x1);

	while (left > 0) {
		BOOL ret;
		DWORD dwTransferBytes;
		ret = ReadFile(file->u.file, (char*)buf+offset, len, &dwTransferBytes, &ov);
		if (!ret && GetLastError() == ERROR_IO_PENDING) {
			ret = GetOverlappedResult(file->u.file, &ov, &dwTransferBytes, TRUE);
		}
		if (rc && dwTransferBytes) {
			offset += dwTransferBytes;
			left -= dwTransferBytes;
			ov.Internal = 0;
			ov.InternalHigh = 0;
			ov.Offset += dwTransferBytes;
			if (ov.Offset < dwTransferBytes)
				ov.OffsetHigh++;
		} else {
			rc = -1;
			break;
		}
	}

	CloseHandle((HANDLE)((DWORD)ov.hEvent & ~1));

	return rc;
}

TOOLKIT_API int ioqueue_file_async_writesome(ioqueue_file_t* file, 
								 ioqueue_overlapped_t *ov,
								 void* buf, 
								 unsigned int len, 
								 ioqueue_on_write_callback on_write_callback,
								 void *user_data)
{
	return ioqueue_file_async_writesome_at(file, ov, buf, len, 0, 0, on_write_callback, user_data);
}

TOOLKIT_API int ioqueue_file_async_writen(ioqueue_file_t* file, 
							  ioqueue_overlapped_t *overlapped,
							  void* buf, 
							  unsigned int len, 
							  ioqueue_on_write_callback on_write_cb,
							  void *user_data)
{
	return ioqueue_file_async_writen_at(file, overlapped, buf, len, 0, 0, on_write_cb, user_data);
}

TOOLKIT_API int ioqueue_file_writesome(ioqueue_file_t* file, const void *buf, unsigned int len)
{
	return ioqueue_file_writesome_at(file, buf, len, 0, 0);
}

TOOLKIT_API int ioqueue_file_writen(ioqueue_file_t* file, const void *buf, unsigned int len)
{
	return ioqueue_file_writen_at(file, buf, len, 0, 0);
}

TOOLKIT_API int ioqueue_file_async_writesome_at(ioqueue_file_t* file, 
									ioqueue_overlapped_t *ov,
									void* buf, 
									unsigned int len, 
									DWORD posLow, 
									DWORD posHigh, 
									ioqueue_on_write_callback on_write_callback,
									void *user_data)
{
	ioqueue_writefilesome_overlapped_t *overlapped;
	BOOL rc;
	ioqueue_t *ioq;

	assert(file);
	assert(ov);
	assert(buf);
	assert(on_write_callback);

	ioq = file->owner;
	if (ioq->stop)
		return -1;

	overlapped = (ioqueue_writefilesome_overlapped_t*)ov;
	memset(overlapped, 0, sizeof(ioqueue_writefilesome_overlapped_t));
	ioqueue_overlapped_set_mask(ov, sizeof(ioqueue_writefilesome_overlapped_t));
	fastlock_enter(file->ov_pending_list_lock);
	list_add_tail(&overlapped->base.pending_entry, &file->ov_pending_list);
	fastlock_leave(file->ov_pending_list_lock);
	overlapped->hevt = CreateEventA(NULL, TRUE, FALSE, NULL);
	overlapped->base.type = OV_WRITEFILESOME;
	overlapped->base.user_data = user_data;
	overlapped->base.handle_ctx = (ioqueue_handle_context*)file;
	overlapped->base.ov.Offset = posLow;
	overlapped->base.ov.OffsetHigh = posHigh;
	overlapped->on_write_callback = on_write_callback;
	overlapped->buf = buf;
	inc_pending_io(file);
	rc = WriteFile(file->u.file, buf, (DWORD)len, NULL, &overlapped->base.ov);
	if (rc || GetLastError() == ERROR_IO_PENDING)
		return 0;
	fastlock_enter(file->ov_pending_list_lock);
	list_del(&overlapped->base.pending_entry);
	fastlock_leave(file->ov_pending_list_lock);
	dec_pending_io(file);
	CloseHandle(overlapped->hevt);
	return -1;
}

TOOLKIT_API int ioqueue_file_async_writen_at(ioqueue_file_t* file, 
								 ioqueue_overlapped_t *ov,
								 void* buf, 
								 unsigned int len, 
								 DWORD posLow, 
								 DWORD posHigh, 
								 ioqueue_on_write_callback on_write_cb,
								 void *user_data)
{
	ioqueue_writefilen_overlapped_t *overlapped;
	BOOL rc;
	ioqueue_t *ioq;

	assert(file);
	assert(ov);
	assert(buf);
	assert(on_write_cb);

	ioq = file->owner;
	if (ioq->stop)
		return -1;

	overlapped = (ioqueue_writefilen_overlapped_t*)ov;
	memset(overlapped, 0, sizeof(ioqueue_writefilen_overlapped_t));
	ioqueue_overlapped_set_mask(ov, sizeof(ioqueue_writefilen_overlapped_t));
	fastlock_enter(file->ov_pending_list_lock);
	list_add_tail(&overlapped->base.pending_entry, &file->ov_pending_list);
	fastlock_leave(file->ov_pending_list_lock);
	overlapped->hevt = CreateEventA(NULL, TRUE, FALSE, NULL);
	overlapped->base.type = OV_WRITEFILEN;
	overlapped->base.user_data = user_data;
	overlapped->base.handle_ctx = (ioqueue_handle_context*)file;
	overlapped->base.ov.Offset = posLow;
	overlapped->base.ov.OffsetHigh = posHigh;
	overlapped->on_write_callback = on_write_cb;
	overlapped->buf = buf;
	overlapped->sended_bytes = 0;
	overlapped->total_bytes = len;
	inc_pending_io(file);
	rc = WriteFile(file->u.file, buf, (DWORD)len, NULL, &overlapped->base.ov);
	if (rc || GetLastError() == ERROR_IO_PENDING)
		return 0;
	fastlock_enter(file->ov_pending_list_lock);
	list_del(&overlapped->base.pending_entry);
	fastlock_leave(file->ov_pending_list_lock);
	dec_pending_io(file);
	CloseHandle(overlapped->hevt);
	return -1;
}

TOOLKIT_API int ioqueue_file_writesome_at(ioqueue_file_t* file,
							  const void *buf, 
							  unsigned int len, 
							  DWORD posLow, 
							  DWORD posHigh)
{
	OVERLAPPED ov;
	BOOL ret;
	DWORD dwTransferBytes;
	int rc = -1;

	memset(&ov, 0, sizeof(ov));
	ov.Offset = posLow;
	ov.OffsetHigh = posHigh;
	ov.hEvent = CreateEventA(NULL, TRUE, FALSE, NULL);
	ov.hEvent = (HANDLE)((DWORD)ov.hEvent & 0x1);

	ret = WriteFile(file->u.file, buf, len, &dwTransferBytes, &ov);

	if (!ret && GetLastError() == ERROR_IO_PENDING) {
		ret = GetOverlappedResult(file->u.file, &ov, &dwTransferBytes, TRUE);
	}

	CloseHandle((HANDLE)((DWORD)ov.hEvent & ~1));

	if (ret && dwTransferBytes > 0)
		rc = dwTransferBytes;

	return rc;
}

TOOLKIT_API int ioqueue_file_writen_at(ioqueue_file_t* file, 
						   const void *buf, 
						   unsigned int len, 
						   DWORD posLow, 
						   DWORD posHigh)
{
	OVERLAPPED ov;
	int rc = 0;
	DWORD offset = 0;
	DWORD left = len;

	memset(&ov, 0, sizeof(ov));
	ov.Offset = posLow;
	ov.OffsetHigh = posHigh;
	ov.hEvent = CreateEventA(NULL, TRUE, FALSE, NULL);
	ov.hEvent = (HANDLE)((DWORD)ov.hEvent & 0x1);

	while (left > 0) {
		BOOL ret;
		DWORD dwTransferBytes;
		ret = WriteFile(file->u.file, (char*)buf+offset, left, &dwTransferBytes, &ov);
		if (!ret && GetLastError() == ERROR_IO_PENDING) {
			ret = GetOverlappedResult(file->u.file, &ov, &dwTransferBytes, TRUE);
		}
		if (ret && dwTransferBytes > 0) {
			offset += dwTransferBytes;
			left -= dwTransferBytes;
			ov.Internal = 0;
			ov.InternalHigh = 0;
			ov.Offset += dwTransferBytes;
			if (ov.Offset < dwTransferBytes)
				ov.OffsetHigh ++;
		} else {
			rc = -1;
			break;
		}
	}

	CloseHandle((HANDLE)((DWORD)ov.hEvent & ~1));

	return rc;
}

TOOLKIT_API ioqueue_t* ioqueue_file_get_owned_ioqueue(ioqueue_file_t* file)
{
	assert(file);
	return file->owner;
}

TOOLKIT_API HANDLE ioqueue_file_get_raw_handle(ioqueue_file_t* file)
{
	assert(file);
	return file->u.file;
}

TOOLKIT_API void *ioqueue_file_set_user_data(ioqueue_file_t* file, void* user_data)
{
	void *old;
	assert(file);
	old = file->user_data;
	file->user_data = user_data;
	return old;
}

TOOLKIT_API void *ioqueue_file_get_user_data(ioqueue_file_t* file)
{
	assert(file);
	return file->user_data;
}

TOOLKIT_API int ioqueue_file_cancel(ioqueue_file_t* file)
{
	assert(file);
	return CancelIo(file->u.file) ? 0 : -1;
}

/* pipe acceptor */

TOOLKIT_API int ioqueue_pipe_acceptor_create(ioqueue_t *ioq, 
								 const char *name, 
								 ioqueue_pipe_acceptor_t *acceptor)
{
	assert(ioq);
	assert(name);
	assert(acceptor);

	memset(acceptor, 0, sizeof(ioqueue_pipe_acceptor_t));
	acceptor->u.pipe_name = strdup_printf("\\\\.\\pipe\\%s", name);
	acceptor->type = HANDLE_TYPE_PIPEACCEPTOR;
	acceptor->owner = ioq;
	fastlock_init(acceptor->ov_pending_list_lock);
	INIT_LIST_HEAD(&acceptor->ov_pending_list);
	add_handler_list(acceptor, ioq);
	inc_ref(ioqueue_handle_context, acceptor);

	return 0;
}

TOOLKIT_API void ioqueue_pipe_acceptor_destroy(ioqueue_pipe_acceptor_t *acceptor)
{
	assert(acceptor);
	dec_ref(ioqueue_handle_context, acceptor);
}

TOOLKIT_API ioqueue_t* ioqueue_pipe_acceptor_get_owned_ioqueue(ioqueue_pipe_acceptor_t *acceptor)
{
	assert(acceptor);
	assert(acceptor->type == HANDLE_TYPE_PIPEACCEPTOR);
	return acceptor->owner;
}

TOOLKIT_API void *ioqueue_pipe_acceptor_set_user_data(ioqueue_pipe_acceptor_t *acceptor, void *user_data)
{
	void *old;
	assert(acceptor);
	assert(acceptor->type == HANDLE_TYPE_PIPEACCEPTOR);
	old = acceptor->user_data;
	acceptor->user_data = user_data;
	return old;
}

TOOLKIT_API void *ioqueue_pipe_acceptor_get_user_data(ioqueue_pipe_acceptor_t *acceptor)
{
	assert(acceptor);
	assert(acceptor->type == HANDLE_TYPE_ACCEPTOR);
	return acceptor->user_data;
}

TOOLKIT_API int ioqueue_pipe_acceptor_async_accept(ioqueue_pipe_acceptor_t *acceptor,
									   ioqueue_overlapped_t *ov,
									   ioqueue_on_pipe_accept_callback on_accept_callback,
									   void *user_data)
{
	ioqueue_t *ioq;
	ioqueue_connectpipe_overlapped_t *overlapped;
	BOOL ret;

	assert(acceptor);
	assert(ov);
	assert(acceptor->type == HANDLE_TYPE_PIPEACCEPTOR);
	assert(on_accept_callback);

	ioq = acceptor->owner;
	if (ioq->stop)
		return -1;

	overlapped = (ioqueue_connectpipe_overlapped_t*)ov;
	memset(overlapped, 0, sizeof(ioqueue_connectpipe_overlapped_t));
	ioqueue_overlapped_set_mask(ov, sizeof(ioqueue_connectpipe_overlapped_t));
	overlapped->client = CreateNamedPipeA(acceptor->u.pipe_name, 
		PIPE_ACCESS_DUPLEX|FILE_FLAG_OVERLAPPED, PIPE_TYPE_BYTE, 
		PIPE_UNLIMITED_INSTANCES, 3072, 3072, NMPWAIT_WAIT_FOREVER, NULL);
	if (overlapped->client == INVALID_HANDLE_VALUE)
		return -1;
	if (!CreateIoCompletionPort(overlapped->client, ioq->iocp, 0, 0)) {
		CloseHandle(overlapped->client);
		return -1;
	}
	overlapped->hevt = CreateEventA(NULL, TRUE, FALSE, NULL); // must be use event, from MSDN
	overlapped->base.type = OV_CONNECTPIPE;
	overlapped->base.user_data = user_data;
	overlapped->base.handle_ctx = acceptor;
	overlapped->base.ov.hEvent = overlapped->hevt;
	fastlock_enter(acceptor->ov_pending_list_lock);
	list_add_tail(&overlapped->base.pending_entry, &acceptor->ov_pending_list);
	fastlock_leave(acceptor->ov_pending_list_lock);
	inc_pending_io(acceptor);
	overlapped->on_accept_callback = on_accept_callback;
	ret = ConnectNamedPipe(overlapped->client, &overlapped->base.ov);
	if (ret || GetLastError() == ERROR_IO_PENDING)
		return 0;
	fastlock_enter(acceptor->ov_pending_list_lock);
	list_del(&overlapped->base.pending_entry);
	fastlock_leave(acceptor->ov_pending_list_lock);
	dec_pending_io(acceptor);
	CloseHandle(overlapped->client);
	CloseHandle(overlapped->hevt);
	return -1;
}

TOOLKIT_API int ioqueue_pipe_acceptor_accept(ioqueue_pipe_acceptor_t *acceptor, HANDLE *p_pipe, int timeout)
{
	ioqueue_t *ioq;
	HANDLE pipe;
	OVERLAPPED ov;
	BOOL ret;

	assert(acceptor);
	assert(p_pipe);
	assert(acceptor->type == HANDLE_TYPE_PIPEACCEPTOR);

	ioq = acceptor->owner;
	if (ioq->stop)
		return -1;
	pipe = CreateNamedPipeA(acceptor->u.pipe_name, 
		PIPE_ACCESS_DUPLEX|FILE_FLAG_OVERLAPPED, PIPE_TYPE_BYTE, 
		PIPE_UNLIMITED_INSTANCES, 3072, 3072, (DWORD)timeout, NULL);
	if (pipe == INVALID_HANDLE_VALUE)
		return -1;
	memset(&ov, 0, sizeof(ov));
	ov.hEvent = CreateEventA(NULL, TRUE, FALSE, NULL);
	ret = ConnectNamedPipe(pipe, &ov);
	CloseHandle(ov.hEvent);
	if (ret && CreateIoCompletionPort(pipe, ioq->iocp, 0, 0)) {
		*p_pipe = pipe;
		return 0;
	} else {
		CloseHandle(pipe);
	}
	return -1;
}

TOOLKIT_API int ioqueue_pipe_acceptor_create_client(ioqueue_pipe_acceptor_t *acceptor, 
										HANDLE h, 
										ioqueue_file_t *pipe)
{
	ioqueue_t *ioq;
	assert(acceptor);
	assert(pipe);
	assert(h != INVALID_HANDLE_VALUE);
	ioq = acceptor->owner;
	if (ioq->stop)
		return -1;
	memset(pipe, 0, sizeof(ioqueue_tcpsock_t));
	pipe->type = HANDLE_TYPE_FILE;
	pipe->u.file = h;
	pipe->owner = ioq;
	INIT_LIST_HEAD(&pipe->ov_pending_list);
	add_handler_list(pipe, ioq);
	inc_ref(ioqueue_handle_context, pipe);
	return 0;
}

TOOLKIT_API int ioqueue_pipe_acceptor_cancel(ioqueue_pipe_acceptor_t *acceptor)
{
	//.....
	assert(0);



	return 0;
}

TOOLKIT_API int ioqueue_pipe_acceptor_close_pending_handle(ioqueue_pipe_acceptor_t *acceptor)
{
	assert(acceptor);

	fastlock_enter(acceptor->ov_pending_list_lock);
	{
		ioqueue_base_overlapped_t *pos;
		list_for_each_entry(pos, &acceptor->ov_pending_list, ioqueue_base_overlapped_t, pending_entry) {
			ioqueue_connectpipe_overlapped_t *overlapped = (ioqueue_connectpipe_overlapped_t *)pos;
			if (overlapped->client != INVALID_HANDLE_VALUE) {
				CloseHandle(overlapped->client);
				overlapped->client = INVALID_HANDLE_VALUE;
			}
		}
	}
	fastlock_leave(acceptor->ov_pending_list_lock);

	return 0;
}

#define IO_GET_TYPE(x)		((x) & 0x0000ff00)
#define IO_COMBINE_TYPE(x, y) ((((x) << 8) | (y)) & 0x0000ffff)

TOOLKIT_API int ioqueue_overlapped_get_mask(const ioqueue_overlapped_t* const io)
{
	assert(io);
	ioqueue_base_overlapped_t* base_ov = (ioqueue_base_overlapped_t*)io;
	return (int)base_ov->ov.Internal;
}

TOOLKIT_API void ioqueue_overlapped_set_mask(ioqueue_overlapped_t* io, int mask_value)
{
	assert(io);
	ioqueue_base_overlapped_t* base_ov = (ioqueue_base_overlapped_t*)io;
	base_ov->ov.Internal = mask_value;
}

/*at first, this function is only used evtpoll for dispatch, after then, bus_endpt seems involved*/
TOOLKIT_API uint32_t ioqueue_overlapped_get_type(const ioqueue_overlapped_t* const io)
{
	if (ioqueue_overlapped_get_mask(io) == sizeof(OVERLAPPED)) {
		return EV_BUS_ENDPOINT;
	}
	/*idx of acceptor*/
	int sub_type = 0;
	const ioqueue_base_overlapped_t* base_ov = (ioqueue_base_overlapped_t*)io;
	const ioqueue_handle_context* handle_ctx = base_ov->handle_ctx;
	const int type = handle_ctx->type;
	switch (type) {
	case HANDLE_TYPE_ACCEPTOR:
	{
		ioqueue_accept_overlapped_t* overlapped = (ioqueue_accept_overlapped_t*)io;
		sub_type = (int)overlapped->base.user_data;
	}
	break;
	case HANDLE_TYPE_PIPEACCEPTOR:
	{
		ioqueue_connectpipe_overlapped_t* overlapped = (ioqueue_connectpipe_overlapped_t*)io;
		sub_type = (int)overlapped->base.user_data;
	}
	break;
	case HANDLE_TYPE_TCPSOCK:
	case HANDLE_TYPE_UDPSOCK:
	case HANDLE_TYPE_FILE:
		sub_type = base_ov->type;
		break;
	default:
		break;
	}
	return IO_COMBINE_TYPE(type, sub_type);
}