自学教程：C++ ACL_VSTREAM_SOCK函数代码示例

static void __rw_timer_fn(ACL_VSTREAM *stream){	char  myname[] = "__rw_timer_fn";	if (stream == NULL || ACL_VSTREAM_SOCK(stream) < 0)		acl_msg_fatal("%s(%d)->%s: input error, null stream",				__FILE__, __LINE__, myname);	if (acl_msg_verbose)		acl_msg_info("%s(%d)->%s: rw timeout, fd = %d",				__FILE__, __LINE__, myname,				ACL_VSTREAM_SOCK(stream));	acl_multi_server_disconnect(stream);}

const char* socket_stream::get_peer_ip() const{	if (stream_ == NULL)		return dummy_;	if (peer_ip_[0] != 0)		return peer_ip_;	char* ptr = ACL_VSTREAM_PEER(stream_);	if (ptr == NULL || *ptr == 0)	{		char  buf[64];		if (acl_getpeername(ACL_VSTREAM_SOCK(stream_),			buf, sizeof(buf)) == -1)		{			return dummy_;		}		acl_vstream_set_peer(stream_, buf);	}	return const_cast<socket_stream*> (this)->get_ip(		ACL_VSTREAM_PEER(stream_),		const_cast<socket_stream*> (this)->peer_ip_,		sizeof(peer_ip_));}

static void event_enable_write(ACL_EVENT *eventp, ACL_VSTREAM *stream,	int timeout, ACL_EVENT_NOTIFY_RDWR callback, void *context){	const char *myname = "event_enable_write";	EVENT_POLL_THR *event_thr = (EVENT_POLL_THR *) eventp;	ACL_EVENT_FDTABLE *fdp;	ACL_SOCKET sockfd;	sockfd = ACL_VSTREAM_SOCK(stream);	fdp = (ACL_EVENT_FDTABLE*) stream->fdp;	if (fdp == NULL) {		fdp = event_fdtable_alloc();		fdp->listener = 0;		fdp->stream = stream;		stream->fdp = (void *) fdp;	} else if (fdp->flag & EVENT_FDTABLE_FLAG_READ)		acl_msg_panic("%s(%d)->%s: fd %d: multiple I/O request",			__FILE__, __LINE__, myname, sockfd);	else {		fdp->listener = 0;		fdp->stream = stream;	}	if (fdp->w_callback != callback || fdp->w_context != context) {		fdp->w_callback = callback;		fdp->w_context = context;	}	if (timeout > 0) {		fdp->w_timeout = timeout * 1000000;		fdp->w_ttl = eventp->present + fdp->w_timeout;	} else {		fdp->w_ttl = 0;		fdp->w_timeout = 0;	}	if ((fdp->flag & EVENT_FDTABLE_FLAG_WRITE) != 0)		return;	stream->nrefer++;	fdp->flag = EVENT_FDTABLE_FLAG_WRITE | EVENT_FDTABLE_FLAG_EXPT;	THREAD_LOCK(&event_thr->event.tb_mutex);	fdp->fdidx = eventp->fdcnt;	eventp->fdtabs[eventp->fdcnt++] = fdp;	event_thr->fds[fdp->fdidx].fd = sockfd;	event_thr->fds[fdp->fdidx].events = POLLOUT | POLLHUP | POLLERR;	if (eventp->maxfd == ACL_SOCKET_INVALID || eventp->maxfd < sockfd)		eventp->maxfd = sockfd;	acl_fdmap_add(event_thr->fdmap, sockfd, fdp);	THREAD_UNLOCK(&event_thr->event.tb_mutex);	if (event_thr->event.blocked && event_thr->event.evdog	    && event_dog_client(event_thr->event.evdog) != stream)		event_dog_notify(event_thr->event.evdog);}

static void enable_listen(EVENT_KERNEL *ev, ACL_EVENT_FDTABLE *fdp){	const char *myname = "enable_listen";	ACL_SOCKET sock;	DWORD ReceiveLen = 0;	sock = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP,		0, 0, WSA_FLAG_OVERLAPPED);	memset(&fdp->event_read->overlapped, 0,		sizeof(fdp->event_read->overlapped));	fdp->stream->type |= ACL_VSTREAM_TYPE_LISTEN_IOCP;	fdp->stream->iocp_sock = sock;	if (AcceptEx(ACL_VSTREAM_SOCK(fdp->stream), sock,		fdp->event_read->myAddrBlock, 0,		ACCEPT_ADDRESS_LENGTH, ACCEPT_ADDRESS_LENGTH,		&ReceiveLen, &fdp->event_read->overlapped) == FALSE		&& acl_last_error() !=ERROR_IO_PENDING)	{		acl_msg_warn("%s(%d): AcceptEx error(%s)",			myname, __LINE__, acl_last_serror());	}}

static void event_enable_read(ACL_EVENT *eventp, ACL_VSTREAM *stream,	int timeout, ACL_EVENT_NOTIFY_RDWR callback, void *context){	const char *myname = "event_enable_read";	EVENT_KERNEL *ev = (EVENT_KERNEL *) eventp;	ACL_EVENT_FDTABLE *fdp = (ACL_EVENT_FDTABLE *) stream->fdp;	ACL_SOCKET sockfd = ACL_VSTREAM_SOCK(stream);	if (fdp == NULL) {		fdp = event_fdtable_alloc();		fdp->flag = EVENT_FDTABLE_FLAG_ADD_READ | EVENT_FDTABLE_FLAG_EXPT;		fdp->stream = stream;		acl_ring_append(&ev->fdp_delay_list, &fdp->delay_entry);		fdp->flag |= EVENT_FDTABLE_FLAG_DELAY_OPER;		stream->fdp = (void *) fdp;		/* 添加流关闭时的回调函数 */		acl_vstream_add_close_handle(stream, stream_on_close, eventp);	} else if ((fdp->flag & EVENT_FDTABLE_FLAG_ADD_READ)) {		goto END;	} else if ((fdp->flag & EVENT_FDTABLE_FLAG_DEL_READ)) {		/* 停止禁止读监听过程 */		acl_assert((fdp->flag & EVENT_FDTABLE_FLAG_READ));		/* 重新启用读监听过程, 因为之前的过程是正在拆除读监听过程但		 * 还没有正式拆除，所以只需要清除拆除标志位即可		 */		fdp->flag &= ~EVENT_FDTABLE_FLAG_DEL_READ;	} else if (!(fdp->flag & EVENT_FDTABLE_FLAG_READ)) {		fdp->flag |= EVENT_FDTABLE_FLAG_ADD_READ;		if (!(fdp->flag & EVENT_FDTABLE_FLAG_DELAY_OPER)) {			acl_ring_append(&ev->fdp_delay_list, &fdp->delay_entry);			fdp->flag |= EVENT_FDTABLE_FLAG_DELAY_OPER;		}	}END:	if (fdp->fdidx == -1) {		fdp->fdidx = eventp->fdcnt;		eventp->fdtabs[eventp->fdcnt++] = fdp;	}	if (eventp->maxfd != ACL_SOCKET_INVALID && eventp->maxfd < sockfd)		eventp->maxfd = sockfd;	if (fdp->r_callback != callback || fdp->r_context != context) {		fdp->r_callback = callback;		fdp->r_context = context;	}	if (timeout > 0) {		fdp->r_timeout = timeout * 1000000;		fdp->r_ttl = eventp->present + fdp->r_timeout;	} else {		fdp->r_ttl = 0;		fdp->r_timeout = 0;	}}

int socket_stream::sock_read(void *ctx, unsigned char *buf, size_t len){#ifdef HAS_POLARSSL	socket_stream* cli = (socket_stream*) ctx;	ACL_VSTREAM* stream = cli->get_vstream();	acl_assert(stream);	int   ret, timeout = 120;	if ((ret = acl_socket_read(ACL_VSTREAM_SOCK(stream), buf, len,		timeout, stream, NULL)) < 0)	{		int   errnum = acl_last_error();		if (ret == ACL_EINTR || ret == ACL_EWOULDBLOCK#if ACL_EWOULDBLOCK != ACL_EAGAIN			|| ret == ACL_EAGAIN#endif			)			return POLARSSL_ERR_NET_WANT_READ;		else if (errnum == ACL_ECONNRESET || errno == EPIPE)			return POLARSSL_ERR_NET_CONN_RESET;		else			return POLARSSL_ERR_NET_RECV_FAILED;	}	return ret;#else	(void) ctx;	(void) buf;	(void) len;	return -1;#endif}

const char* socket_stream::get_local_ip() const{	if (stream_ == NULL)		return dummy_;	// xxx: acl_vstream 中没有对此地址赋值	if (local_ip_[0] != 0)		return local_ip_;	char* ptr = ACL_VSTREAM_LOCAL(stream_);	if (ptr == NULL || *ptr == 0)	{		char  buf[256];		if (acl_getsockname(ACL_VSTREAM_SOCK(stream_),			buf, sizeof(buf)) == -1)		{			return dummy_;		}		acl_vstream_set_local(stream_, buf);	}	return const_cast<socket_stream*>(this)->get_ip(		ACL_VSTREAM_LOCAL(stream_),		const_cast<socket_stream*>(this)->local_ip_,		sizeof(local_ip_));}

static void event_enable_read(ACL_EVENT *eventp, ACL_VSTREAM *stream,	int timeout, ACL_EVENT_NOTIFY_RDWR callback, void *context){	ACL_EVENT_FDTABLE *fdp = read_enable(eventp, stream, timeout,			callback, context);	fdp->listener = acl_is_listening_socket(ACL_VSTREAM_SOCK(stream));}

int polarssl_io::sock_read(void *ctx, unsigned char *buf, size_t len){#ifdef HAS_POLARSSL	polarssl_io* io = (polarssl_io*) ctx;	int   ret, timeout = 120;	ACL_VSTREAM* vs = io->stream_->get_vstream();	ACL_SOCKET fd = ACL_VSTREAM_SOCK(vs);	ret = acl_socket_read(fd, buf, len, timeout, vs, NULL);	if (ret < 0)	{		int   errnum = acl_last_error();		if (ret == ACL_EINTR || ret == ACL_EWOULDBLOCK#if ACL_EWOULDBLOCK != ACL_EAGAIN			|| ret == ACL_EAGAIN#endif			)			return POLARSSL_ERR_NET_WANT_READ;		else if (errnum == ACL_ECONNRESET || errno == EPIPE)			return POLARSSL_ERR_NET_CONN_RESET;		else			return POLARSSL_ERR_NET_RECV_FAILED;	}	return ret;#else	(void) ctx;	(void) buf;	(void) len;	logger_error("HAS_POLARSSL not defined!");	return -1;#endif}

static int disable_write(EVENT_KERNEL *ev, ACL_EVENT_FDTABLE *fdp){	const char *myname = "disable_write";	ACL_VSTREAM *stream = fdp->stream;	ACL_SOCKET sockfd = ACL_VSTREAM_SOCK(stream);	int   err, ret = 0;	fdp->flag &= ~EVENT_FDTABLE_FLAG_DEL_WRITE;	fdp->flag &= ~EVENT_FDTABLE_FLAG_WRITE;	fdp->event_type &= ~(ACL_EVENT_WRITE | ACL_EVENT_CONNECT);	if ((fdp->flag & EVENT_FDTABLE_FLAG_READ)) {#if (ACL_EVENTS_KERNEL_STYLE == ACL_EVENTS_STYLE_KQUEUE)		EVENT_REG_DEL_WRITE(err, ev->event_fd, sockfd);#else		EVENT_REG_MOD_READ(err, ev->event_fd, sockfd, fdp);#endif	} else {#ifdef	USE_FDMAP		acl_fdmap_del(ev->fdmap, sockfd);#endif#ifdef	EVENT_REG_DEL_BOTH		EVENT_REG_DEL_BOTH(err, ev->event_fd, sockfd);#else		EVENT_REG_DEL_WRITE(err, ev->event_fd, sockfd);#endif		ret = 1;	}	if (err < 0) {		acl_msg_fatal("%s: %s: %s, err(%d), fd(%d), ret(%d)",			myname, EVENT_REG_DEL_TEXT, acl_last_serror(),			err, sockfd, ret);	}	return (ret);}

ACL_SOCKET aio_stream::get_socket() const{	ACL_VSTREAM* stream = get_vstream();	if (stream == NULL)		return ACL_SOCKET_INVALID;	return ACL_VSTREAM_SOCK(stream);}

int ssl_aio_stream::__sock_send(void *ctx, const unsigned char *buf, size_t len){#ifdef HAS_POLARSSL	ssl_aio_stream* cli = (ssl_aio_stream*) ctx;	ACL_VSTREAM* stream = cli->get_vstream();	acl_assert(stream);	int   ret, errnum;	if ((ret = acl_socket_write(ACL_VSTREAM_SOCK(stream), buf, len,		0, stream, NULL)) < 0)	{		errnum = acl_last_error();		if (errnum == ACL_EINTR)			return POLARSSL_ERR_NET_WANT_WRITE;		else if (errnum == ACL_EWOULDBLOCK)			return POLARSSL_ERR_NET_WANT_WRITE;#if ACL_EWOULDBLOCK != ACL_EAGAIN		else if (errnum == ACL_EAGAIN)			return POLARSSL_ERR_NET_WANT_WRITE;#endif		else if (errnum == ACL_ECONNRESET || errno == EPIPE)			return POLARSSL_ERR_NET_CONN_RESET;		else			return POLARSSL_ERR_NET_SEND_FAILED;	}	return ret;#else	(void) ctx;	(void) buf;	(void) len;	return -1;#endif}

const char* aio_stream::get_local(bool full /* = false */) const{	if (stream_ == NULL)		return dummy_;	ACL_VSTREAM* vs = acl_aio_vstream(stream_);	const char* ptr = ACL_VSTREAM_LOCAL(vs);	ACL_SOCKET fd = ACL_VSTREAM_SOCK(vs);	if (ptr == NULL || *ptr == 0)	{		char  buf[256];		if (acl_getsockname(fd, buf, sizeof(buf)) == -1)			return dummy_;		acl_vstream_set_local(vs, buf);	}	ptr = ACL_VSTREAM_LOCAL(vs);	if (full)		return ptr;	else if (local_ip_[0] != 0)		return local_ip_;	return const_cast<aio_stream*> (this)->get_ip(ptr,		const_cast<aio_stream*>(this)->local_ip_, sizeof(local_ip_));}

static void enable_write(EVENT_KERNEL *ev, ACL_EVENT_FDTABLE *fdp){	const char *myname = "enable_write";	ACL_SOCKET sockfd = ACL_VSTREAM_SOCK(fdp->stream);	int   err;	fdp->flag &= ~EVENT_FDTABLE_FLAG_ADD_WRITE;	fdp->flag |= EVENT_FDTABLE_FLAG_WRITE;	if ((fdp->flag & EVENT_FDTABLE_FLAG_READ)) {#if (ACL_EVENTS_KERNEL_STYLE == ACL_EVENTS_STYLE_KQUEUE)		EVENT_REG_ADD_WRITE(err, ev->event_fd, sockfd, fdp);#else		EVENT_REG_MOD_RDWR(err, ev->event_fd, sockfd, fdp);#endif	} else {		EVENT_REG_ADD_WRITE(err, ev->event_fd, sockfd, fdp);	}	if (err < 0) {		acl_msg_fatal("%s: %s: %s, err(%d), fd(%d)",			myname, EVENT_REG_ADD_TEXT,			acl_last_serror(), err, sockfd);	}}

static int vstream_server(void){	const char *myname = "vstream_server";	ACL_VSTREAM *sstream, *client;	char  ebuf[256];	sstream = acl_vstream_listen(addr, 128);	if (sstream == NULL) {		printf("%s(%d): listen on %s error(%s)/r/n",			myname, __LINE__, addr,			acl_last_strerror(ebuf, sizeof(ebuf)));		return (-1);	}	acl_tcp_defer_accept(ACL_VSTREAM_SOCK(sstream), 0);	printf("%s: listen %s ok/r/n", myname, addr);	while (1) {		client = acl_vstream_accept(sstream, NULL, 0);		if (client == NULL) {			printf("%s(%d): accept error(%s)/r/n",				myname, __LINE__,				acl_last_strerror(ebuf, sizeof(ebuf)));			return (-1);		}		printf("accept one/r/n");		if (1)			reply_client2(client);		else			reply_client(client);		acl_vstream_close(client);	}		return (0);}

static void event_disable_readwrite(ACL_EVENT *eventp, ACL_VSTREAM *stream){	const char *myname = "event_disable_readwrite";	EVENT_KERNEL *ev = (EVENT_KERNEL *) eventp;	ACL_SOCKET sockfd = ACL_VSTREAM_SOCK(stream);	ACL_EVENT_FDTABLE *fdp = (ACL_EVENT_FDTABLE *) stream->fdp;	int   err = 0;	if (fdp == NULL) {		return;	}	if (fdp->flag == 0 || fdp->fdidx < 0 || fdp->fdidx >= eventp->fdcnt) {		acl_msg_warn("%s(%d): sockfd(%d) no set, fdp no null",			myname, __LINE__, sockfd);		event_fdtable_free(fdp);		stream->fdp = NULL;		return;	}	if (eventp->maxfd == sockfd)		eventp->maxfd = ACL_SOCKET_INVALID;	if (fdp->fdidx < --eventp->fdcnt) {		eventp->fdtabs[fdp->fdidx] = eventp->fdtabs[eventp->fdcnt];		eventp->fdtabs[fdp->fdidx]->fdidx = fdp->fdidx;	}	fdp->fdidx = -1;#ifdef	EVENT_REG_DEL_BOTH	EVENT_REG_DEL_BOTH(err, ev->event_fd, sockfd);#else	if (fdp->flag & EVENT_FDTABLE_FLAG_READ) {		EVENT_REG_DEL_READ(err, ev->event_fd, sockfd);	}	if (fdp->flag & EVENT_FDTABLE_FLAG_WRITE) {		EVENT_REG_DEL_WRITE(err, ev->event_fd, sockfd);	}#endif	if (err < 0) {		acl_msg_fatal("%s: %s: %s", myname, EVENT_REG_DEL_TEXT,			acl_last_serror());	}#ifdef	USE_FDMAP	acl_fdmap_del(ev->fdmap, sockfd);#endif	if (fdp->fdidx_ready >= 0	    && fdp->fdidx_ready < eventp->fdcnt_ready	    && eventp->fdtabs_ready[fdp->fdidx_ready] == fdp)	{		eventp->fdtabs_ready[fdp->fdidx_ready] = NULL;	}	fdp->fdidx_ready = -1;	event_fdtable_free(fdp);	stream->fdp = NULL;}

void acl_event_disable_write(ACL_EVENT *eventp, ACL_VSTREAM *stream){	const char *myname = "acl_event_disable_write";	ACL_SOCKET sockfd = ACL_VSTREAM_SOCK(stream);	if (sockfd == ACL_SOCKET_INVALID)		acl_msg_fatal("%s(%d): sockfd(%d) invalid",			myname, __LINE__, sockfd);	eventp->disable_write_fn(eventp, stream);}

bool socket_stream::shutdown_readwrite(){	if (stream_ == NULL)	{		logger_error("stream_ null");		return false;	}	return acl_socket_shutdown(ACL_VSTREAM_SOCK(stream_), SHUT_RDWR) == 0;}

static void event_disable_write(ACL_EVENT *eventp, ACL_VSTREAM *stream){	const char *myname = "event_disable_write";	EVENT_SELECT *ev = (EVENT_SELECT *) eventp;	ACL_SOCKET sockfd = ACL_VSTREAM_SOCK(stream);	ACL_EVENT_FDTABLE *fdp = (ACL_EVENT_FDTABLE *) stream->fdp;	if (fdp == NULL) {		acl_msg_warn("%s(%d): fdp null", myname, __LINE__);		return;	}	if (fdp->fdidx < 0 || fdp->fdidx >= eventp->fdcnt) {		acl_msg_warn("%s(%d): sockfd(%d)'s fdidx invalid",			myname, __LINE__, sockfd);		return;	}	if (!(fdp->flag & EVENT_FDTABLE_FLAG_WRITE)) {		acl_msg_warn("%s(%d): sockfd(%d) not in wmask",			myname, __LINE__, sockfd);		return;	}	fdp->w_ttl = 0;	fdp->w_timeout = 0;	fdp->w_callback = NULL;	fdp->event_type &= ~(ACL_EVENT_WRITE | ACL_EVENT_CONNECT);	fdp->flag &= ~EVENT_FDTABLE_FLAG_WRITE;	if ((fdp->flag & EVENT_FDTABLE_FLAG_READ)) {		FD_CLR(sockfd, &ev->wmask);		return;	}	if (eventp->maxfd == sockfd)		eventp->maxfd = ACL_SOCKET_INVALID;	if (fdp->fdidx < --eventp->fdcnt) {		eventp->fdtabs[fdp->fdidx] = eventp->fdtabs[eventp->fdcnt];		eventp->fdtabs[fdp->fdidx]->fdidx = fdp->fdidx;	}	fdp->fdidx = -1;	if (fdp->fdidx_ready >= 0		&& fdp->fdidx_ready < eventp->ready_cnt		&& eventp->ready[fdp->fdidx_ready] == fdp)	{		eventp->ready[fdp->fdidx_ready] = NULL;	}	fdp->fdidx_ready = -1;	FD_CLR(sockfd, &ev->xmask);	FD_CLR(sockfd, &ev->wmask);}

static void event_enable_write(ACL_EVENT *eventp, ACL_VSTREAM *stream,	int timeout, ACL_EVENT_NOTIFY_RDWR callback, void *context){	EVENT_KERNEL *ev = (EVENT_KERNEL *) eventp;	ACL_EVENT_FDTABLE *fdp = (ACL_EVENT_FDTABLE *) stream->fdp;	ACL_SOCKET sockfd = ACL_VSTREAM_SOCK(stream);	if (fdp == NULL) {		fdp = event_fdtable_alloc();		fdp->flag = EVENT_FDTABLE_FLAG_ADD_WRITE | EVENT_FDTABLE_FLAG_EXPT;		fdp->stream = stream;		acl_ring_append(&ev->fdp_delay_list, &fdp->delay_entry);		fdp->flag |= EVENT_FDTABLE_FLAG_DELAY_OPER;		stream->fdp = (void *) fdp;		/* 添加流关闭时的回调函数 */		acl_vstream_add_close_handle(stream, stream_on_close, eventp);#ifdef	USE_FDMAP		acl_fdmap_add(ev->fdmap, sockfd, fdp);#endif	} else if ((fdp->flag & EVENT_FDTABLE_FLAG_ADD_WRITE)) {		goto END;	} else if ((fdp->flag & EVENT_FDTABLE_FLAG_DEL_WRITE)) {		acl_assert((fdp->flag & EVENT_FDTABLE_FLAG_WRITE));		fdp->flag &= ~EVENT_FDTABLE_FLAG_DEL_WRITE;	} else if (!(fdp->flag & EVENT_FDTABLE_FLAG_WRITE)) {		fdp->flag |= EVENT_FDTABLE_FLAG_ADD_WRITE;		if (!(fdp->flag & EVENT_FDTABLE_FLAG_DELAY_OPER)) {			acl_ring_append(&ev->fdp_delay_list, &fdp->delay_entry);			fdp->flag |= EVENT_FDTABLE_FLAG_DELAY_OPER;		}	}END:	if (fdp->fdidx == -1) {		fdp->fdidx = eventp->fdcnt;		eventp->fdtabs[eventp->fdcnt++] = fdp;	}	if (eventp->maxfd != ACL_SOCKET_INVALID && eventp->maxfd < sockfd)		eventp->maxfd = sockfd;	if (fdp->w_callback != callback || fdp->w_context != context) {		fdp->w_callback = callback;		fdp->w_context = context;	}	if (timeout > 0) {		fdp->w_timeout = timeout * 1000000;		fdp->w_ttl = eventp->present + fdp->w_timeout;	} else {		fdp->w_ttl = 0;		fdp->w_timeout = 0;	}}

ACL_SOCKET socket_stream::unbind_sock(){	if (stream_ == NULL)		return ACL_SOCKET_INVALID;	ACL_SOCKET sock = ACL_VSTREAM_SOCK(stream_);	stream_->fd.sock = ACL_SOCKET_INVALID;	eof_ = true;	opened_ = false;	return sock;}

static void stream_on_close(ACL_VSTREAM *stream, void *arg){	EVENT_POLL *ev = (EVENT_POLL*) arg;	ACL_EVENT_FDTABLE *fdp = (ACL_EVENT_FDTABLE*) stream->fdp;	ACL_SOCKET sockfd = ACL_VSTREAM_SOCK(stream);	int   ret = 0;	if (fdp == NULL)		return;	if ((fdp->flag & EVENT_FDTABLE_FLAG_READ)		&& (fdp->flag & EVENT_FDTABLE_FLAG_WRITE))	{		ret = 1;	} else if ((fdp->flag & EVENT_FDTABLE_FLAG_READ)) {		ret = 2;	} else if ((fdp->flag & EVENT_FDTABLE_FLAG_WRITE)) {		ret = 3;	}	if (ret) {		acl_fdmap_del(ev->fdmap, sockfd);	}	if (ev->event.maxfd == ACL_VSTREAM_SOCK(fdp->stream))		ev->event.maxfd = ACL_SOCKET_INVALID;	if (fdp->fdidx >= 0 && fdp->fdidx < --ev->event.fdcnt) {		ev->fds[fdp->fdidx] = ev->fds[ev->event.fdcnt];		ev->event.fdtabs[fdp->fdidx] = ev->event.fdtabs[ev->event.fdcnt];		ev->event.fdtabs[fdp->fdidx]->fdidx = fdp->fdidx;	}	fdp->fdidx = -1;	if (fdp->fdidx_ready >= 0		&& fdp->fdidx_ready < ev->event.ready_cnt		&& ev->event.ready[fdp->fdidx_ready] == fdp)	{		ev->event.ready[fdp->fdidx_ready] = NULL;	}	fdp->fdidx_ready = -1;	event_fdtable_free(fdp);	stream->fdp = NULL;}

/*----------------------------------------------------------------------------*/static void __read_notify_callback(int event_type, ACL_IOCTL *h_ioctl,	ACL_VSTREAM *cstream, void *context){	const char *myname = "__read_notify_callback";	ACL_APP_HANDLE *app;	int   ret;	app = (ACL_APP_HANDLE *) context;	switch (event_type) {	case ACL_EVENT_READ:		ret = __run_fn(cstream, __run_ctx);		if (ret < 0) {			if (__app_on_close != NULL)				__app_on_close(cstream, __run_ctx);			acl_vstream_close(cstream);		} else if (ret == 0) {			if (acl_msg_verbose)				acl_msg_info("enable(%s), fd=%d, h_ioctl(%p)",					myname, ACL_VSTREAM_SOCK(cstream),					(void*) h_ioctl);			acl_ioctl_enable_read(h_ioctl, cstream, acl_var_ioctl_rw_timeout,				__read_notify_callback, (void *) app);		}		break;	case ACL_EVENT_RW_TIMEOUT:		if (__app_on_timeout == NULL) {			if (__app_on_close != NULL)				__app_on_close(cstream, __run_ctx);			acl_vstream_close(cstream);		} else if (__app_on_timeout(cstream, __run_ctx) < 0) {			if (__app_on_close != NULL)				__app_on_close(cstream, __run_ctx);			acl_vstream_close(cstream);		} else {			acl_ioctl_enable_read(h_ioctl, cstream, acl_var_ioctl_rw_timeout,				__read_notify_callback, (void *) app);		}		break;	case ACL_EVENT_XCPT:		if (__app_on_close != NULL)			__app_on_close(cstream, __run_ctx);		acl_vstream_close(cstream);		break;	default:		acl_msg_fatal("%s, %s(%d): unknown event type(%d)",			__FILE__, myname, __LINE__, event_type);		/* not reached */		break;	}	if (acl_msg_verbose)		acl_msg_info("%s(%d): total alloc: %d",			myname, __LINE__, acl_mempool_total_allocated());}

static void event_disable_readwrite(ACL_EVENT *eventp, ACL_VSTREAM *stream){	const char *myname = "event_disable_readwrite";	EVENT_SELECT_THR *event_thr = (EVENT_SELECT_THR *) eventp;	ACL_EVENT_FDTABLE *fdp;	ACL_SOCKET sockfd;	sockfd = ACL_VSTREAM_SOCK(stream);	THREAD_LOCK(&event_thr->event.tb_mutex);	if (!FD_ISSET(sockfd, &event_thr->xmask)) {		acl_msg_error("%s(%d): sockfd(%d) not be set",			myname, __LINE__, sockfd);		THREAD_UNLOCK(&event_thr->event.tb_mutex);		return;	}	fdp = (ACL_EVENT_FDTABLE *) stream->fdp;	if (fdp == NULL) {		acl_msg_error("%s(%d): fdp null", myname, __LINE__);		THREAD_UNLOCK(&event_thr->event.tb_mutex);		return;	}	if (fdp->fdidx == -1)		acl_msg_fatal("%s(%d): fdidx(%d) invalid",			myname, __LINE__, fdp->fdidx);	FD_CLR(sockfd, &event_thr->xmask);	FD_CLR(sockfd, &event_thr->rmask);	FD_CLR(sockfd, &event_thr->wmask);	fdp->flag = 0;	if (eventp->maxfd == sockfd)		eventp->maxfd = ACL_SOCKET_INVALID;	if (eventp->fdtabs[fdp->fdidx] == fdp) {		if (fdp->fdidx < --eventp->fdcnt) {			eventp->fdtabs[fdp->fdidx] = eventp->fdtabs[eventp->fdcnt];			eventp->fdtabs[fdp->fdidx]->fdidx = fdp->fdidx;		}	} else		acl_msg_fatal("%s(%d): fdidx(%d)'s fdp invalid",			myname, __LINE__, fdp->fdidx);	if (fdp->fdidx_ready > 0	    && fdp->fdidx_ready < eventp->fdcnt_ready	    && eventp->fdtabs_ready[fdp->fdidx_ready] == fdp)	{		eventp->fdtabs_ready[fdp->fdidx_ready] = NULL;	}	event_fdtable_free(fdp);	stream->fdp = NULL;	stream->nrefer--;	THREAD_UNLOCK(&event_thr->event.tb_mutex);}

static void event_disable_read(ACL_EVENT *eventp, ACL_VSTREAM *stream){	const char *myname = "event_disable_read";	EVENT_POLL *ev = (EVENT_POLL *) eventp;	ACL_SOCKET sockfd = ACL_VSTREAM_SOCK(stream);	ACL_EVENT_FDTABLE *fdp = (ACL_EVENT_FDTABLE *) stream->fdp;	if (fdp == NULL) {		acl_msg_warn("%s(%d): fdp null", myname, __LINE__);		return;	}	if (fdp->fdidx < 0 || fdp->fdidx >= eventp->fdcnt) {		acl_msg_warn("%s(%d): sockfd(%d)'s fdidx(%d) invalid",			myname, __LINE__, sockfd, fdp->fdidx);		return;	}	if (!(fdp->flag & EVENT_FDTABLE_FLAG_READ)) {		acl_msg_warn("%s(%d): sockfd(%d) not be set",			myname, __LINE__, sockfd);		return;	}	fdp->r_ttl = 0;	fdp->r_timeout = 0;	fdp->r_callback = NULL;	fdp->event_type &= ~(ACL_EVENT_READ | ACL_EVENT_ACCEPT);	fdp->flag &= ~EVENT_FDTABLE_FLAG_READ;	if ((fdp->flag & EVENT_FDTABLE_FLAG_WRITE)) {		ev->fds[fdp->fdidx].events = POLLOUT | POLLHUP | POLLERR;		return;	}	if (eventp->maxfd == sockfd)		eventp->maxfd = ACL_SOCKET_INVALID;	if (fdp->fdidx < --eventp->fdcnt) {		ev->fds[fdp->fdidx] = ev->fds[eventp->fdcnt];		eventp->fdtabs[fdp->fdidx] = eventp->fdtabs[eventp->fdcnt];		eventp->fdtabs[fdp->fdidx]->fdidx = fdp->fdidx;	}	fdp->fdidx = -1;	if (fdp->fdidx_ready >= 0		&& fdp->fdidx_ready < eventp->ready_cnt		&& eventp->ready[fdp->fdidx_ready] == fdp)	{		eventp->ready[fdp->fdidx_ready] = NULL;	}	fdp->fdidx_ready = -1;	acl_fdmap_del(ev->fdmap, sockfd);}

static void event_disable_readwrite(ACL_EVENT *eventp, ACL_VSTREAM *stream){	const char *myname = "event_disable_readwrite";	EVENT_SELECT *ev = (EVENT_SELECT *) eventp;	ACL_EVENT_FDTABLE *fdp = (ACL_EVENT_FDTABLE *) stream->fdp;	ACL_SOCKET sockfd = ACL_VSTREAM_SOCK(stream);	if (fdp == NULL)		return;	if (fdp->flag == 0 || fdp->fdidx < 0 || fdp->fdidx >= eventp->fdcnt) {		acl_msg_warn("%s(%d): sockfd(%d) no set, fdp no null",			myname, __LINE__, sockfd);		event_fdtable_free(fdp);		stream->fdp = NULL;		return;	}	if (!FD_ISSET(sockfd, &ev->rmask) && !FD_ISSET(sockfd, &ev->wmask)) {		acl_msg_error("%s(%d): sockfd(%d) no set, fdp no null",			myname, __LINE__, sockfd);		event_fdtable_free(fdp);		stream->fdp = NULL;		return;	}	if (eventp->maxfd == sockfd)		eventp->maxfd = ACL_SOCKET_INVALID;	if (fdp->fdidx < --eventp->fdcnt) {		eventp->fdtabs[fdp->fdidx] = eventp->fdtabs[eventp->fdcnt];		eventp->fdtabs[fdp->fdidx]->fdidx = fdp->fdidx;	}	fdp->fdidx = -1;	if (FD_ISSET(sockfd, &ev->rmask)) {		FD_CLR(sockfd, &ev->rmask);	}	if (FD_ISSET(sockfd, &ev->wmask)) {		FD_CLR(sockfd, &ev->wmask);	}	FD_CLR(sockfd, &ev->xmask);	if (fdp->fdidx_ready >= 0		&& fdp->fdidx_ready < eventp->ready_cnt		&& eventp->ready[fdp->fdidx_ready] == fdp)	{		eventp->ready[fdp->fdidx_ready] = NULL;	}	fdp->fdidx_ready = -1;	event_fdtable_free(fdp);	stream->fdp = NULL;}

static void run_server(const char *addr){    ACL_VSTREAM *sstream = acl_vstream_listen(addr, 128);    acl_pthread_pool_t *pool;    CONN *conn;    acl_pthread_t tid;    acl_pthread_attr_t attr;    ACL_VSTREAM *client;    if (sstream == NULL) {        acl_msg_error("listen %s error(%s)", addr, acl_last_serror());        return;    }    pthread_attr_init(&attr);    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);    printf("listening on %s .../n", addr);    pool = acl_thread_pool_create(10, 10);    while (1) {        acl_mem_slice_delay_destroy();        if (acl_read_wait(ACL_VSTREAM_SOCK(sstream), 5) == -1)            continue;        client = acl_vstream_accept(sstream, NULL, 0);        if (client == NULL) {            acl_msg_error("accept error(%s)", acl_last_serror());            break;        }        acl_tcp_set_nodelay(ACL_VSTREAM_SOCK(client));        conn = acl_mycalloc(1, sizeof(CONN));        conn->stream = client;        if (0)            thread_run(conn);        else if (1)            acl_pthread_create(&tid, &attr, thread_main, conn);        else            acl_pthread_pool_add(pool, thread_run, conn);    }    acl_vstream_close(sstream);}

stream_hook* stream::setup_hook(stream_hook* hook){	if (stream_ == NULL)	{		logger_error("stream_ null");		return NULL;	}	stream_hook* old_hook = hook_;	if (stream_->type == ACL_VSTREAM_TYPE_FILE)	{		ACL_FSTREAM_RD_FN read_fn = stream_->fread_fn;		ACL_FSTREAM_WR_FN write_fn = stream_->fwrite_fn;		stream_->fread_fn = fread_hook;		stream_->fwrite_fn = fsend_hook;		acl_vstream_add_object(stream_, HOOK_KEY, this);		if (hook->open(stream_) == false)		{			// 如果打开失败，则恢复			stream_->fread_fn = read_fn;			stream_->fwrite_fn = write_fn;			acl_vstream_del_object(stream_, HOOK_KEY);			return hook;		}	}	else	{		ACL_VSTREAM_RD_FN read_fn = stream_->read_fn;		ACL_VSTREAM_WR_FN write_fn = stream_->write_fn;		stream_->read_fn = read_hook;		stream_->write_fn = send_hook;		acl_vstream_add_object(stream_, HOOK_KEY, this);		acl_tcp_set_nodelay(ACL_VSTREAM_SOCK(stream_));		if (hook->open(stream_) == false)		{			// 如果打开失败，则恢复			stream_->read_fn = read_fn;			stream_->write_fn = write_fn;			acl_vstream_del_object(stream_, HOOK_KEY);			return hook;		}	}	hook_ = hook;	return old_hook;}

void acl_event_enable_write(ACL_EVENT *eventp, ACL_VSTREAM *stream,	int write_timeout, ACL_EVENT_NOTIFY_RDWR callback, void *context){	const char *myname = "acl_event_enable_write";	ACL_SOCKET sockfd = ACL_VSTREAM_SOCK(stream);	if (sockfd == ACL_SOCKET_INVALID)		acl_msg_fatal("%s(%d): sockfd(%d) invalid",			myname, __LINE__, sockfd);	eventp->enable_write_fn(eventp, stream, write_timeout,			callback, context);}

static void event_enable_listen(ACL_EVENT *eventp, ACL_VSTREAM *stream,	int timeout, ACL_EVENT_NOTIFY_RDWR callback, void *context){	ACL_EVENT_FDTABLE *fdp = read_enable(eventp, stream, timeout,			callback, context);#if defined(ACL_MACOSX)	fdp->listener = 1;#else	fdp->listener = acl_is_listening_socket(ACL_VSTREAM_SOCK(stream));#endif}