pjlib/src/pj/ioqueue_epoll.c - pjproject - Gitiles

 /* $Header: /pjproject-0.3/pjlib/src/pj/ioqueue_epoll.c 4     10/29/05 10:27p Bennylp $ */
 /*
  * $Log: /pjproject-0.3/pjlib/src/pj/ioqueue_epoll.c $
  *
  * 4     10/29/05 10:27p Bennylp
  * Fixed misc warnings.
  *
  * 3     10/29/05 11:49a Bennylp
  * Fixed warnings.
  *
  * 2     10/29/05 11:31a Bennylp
  * Changed accept and lock.
  *
  * 1     10/17/05 10:49p Bennylp
  * Created.
  *
  */

 /*
  * ioqueue_epoll.c
  *
  * This is the implementation of IOQueue framework using /dev/epoll
  * API in _both_ Linux user-mode and kernel-mode.
  */

 #include <pj/ioqueue.h>
 #include <pj/os.h>
 #include <pj/lock.h>
 #include <pj/log.h>
 #include <pj/list.h>
 #include <pj/pool.h>
 #include <pj/string.h>
 #include <pj/assert.h>
 #include <pj/errno.h>
 #include <pj/sock.h>
 #include <pj/compat/socket.h>

 #if !defined(PJ_LINUX_KERNEL) || PJ_LINUX_KERNEL==0
     /*
      * Linux user mode
      */
 #   include <sys/epoll.h>
 #   include <errno.h>
 #   include <unistd.h>

 #   define epoll_data		data.ptr
 #   define epoll_data_type	void*
 #   define ioctl_val_type	unsigned long*
 #   define getsockopt_val_ptr	int*
 #   define os_getsockopt	getsockopt
 #   define os_ioctl		ioctl
 #   define os_read		read
 #   define os_close		close
 #   define os_epoll_create	epoll_create
 #   define os_epoll_ctl		epoll_ctl
 #   define os_epoll_wait	epoll_wait
 #else
     /*
      * Linux kernel mode.
      */
 #   include <linux/config.h>
 #   include <linux/version.h>
 #   if defined(MODVERSIONS)
 #	include <linux/modversions.h>
 #   endif
 #   include <linux/kernel.h>
 #   include <linux/poll.h>
 #   include <linux/eventpoll.h>
 #   include <linux/syscalls.h>
 #   include <linux/errno.h>
 #   include <linux/unistd.h>
 #   include <asm/ioctls.h>
     enum EPOLL_EVENTS
     {
 	EPOLLIN = 0x001,
 	EPOLLOUT = 0x004,
 	EPOLLERR = 0x008,
     };
 #   define os_epoll_create		sys_epoll_create
     static int os_epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)
     {
 	long rc;
 	mm_segment_t oldfs = get_fs();
 	set_fs(KERNEL_DS);
 	rc = sys_epoll_ctl(epfd, op, fd, event);
 	set_fs(oldfs);
 	if (rc) {
 	    errno = -rc;
 	    return -1;
 	} else {
 	    return 0;
 	}
     }
     static int os_epoll_wait(int epfd, struct epoll_event *events,
 			  int maxevents, int timeout)
     {
 	int count;
 	mm_segment_t oldfs = get_fs();
 	set_fs(KERNEL_DS);
 	count = sys_epoll_wait(epfd, events, maxevents, timeout);
 	set_fs(oldfs);
 	return count;
     }
 #   define os_close		sys_close
 #   define os_getsockopt	pj_sock_getsockopt
     static int os_read(int fd, void *buf, size_t len)
     {
 	long rc;
 	mm_segment_t oldfs = get_fs();
 	set_fs(KERNEL_DS);
 	rc = sys_read(fd, buf, len);
 	set_fs(oldfs);
 	if (rc) {
 	    errno = -rc;
 	    return -1;
 	} else {
 	    return 0;
 	}
     }
 #   define socklen_t		unsigned
 #   define ioctl_val_type	unsigned long
     int ioctl(int fd, int opt, ioctl_val_type value);
     static int os_ioctl(int fd, int opt, ioctl_val_type value)
     {
 	int rc;
         mm_segment_t oldfs = get_fs();
 	set_fs(KERNEL_DS);
 	rc = ioctl(fd, opt, value);
 	set_fs(oldfs);
 	if (rc < 0) {
 	    errno = -rc;
 	    return rc;
 	} else
 	    return rc;
     }
 #   define getsockopt_val_ptr	char*

 #   define epoll_data		data
 #   define epoll_data_type	__u32
 #endif

 #define THIS_FILE   "ioq_epoll"

 #define PJ_IOQUEUE_IS_READ_OP(op)   ((op & PJ_IOQUEUE_OP_READ) || \
                                      (op & PJ_IOQUEUE_OP_RECV) || \
                                      (op & PJ_IOQUEUE_OP_RECV_FROM))
 #define PJ_IOQUEUE_IS_WRITE_OP(op)  ((op & PJ_IOQUEUE_OP_WRITE) || \
                                      (op & PJ_IOQUEUE_OP_SEND) || \
                                      (op & PJ_IOQUEUE_OP_SEND_TO))


 #if PJ_HAS_TCP
 #  define PJ_IOQUEUE_IS_ACCEPT_OP(op)	(op & PJ_IOQUEUE_OP_ACCEPT)
 #  define PJ_IOQUEUE_IS_CONNECT_OP(op)	(op & PJ_IOQUEUE_OP_CONNECT)
 #else
 #  define PJ_IOQUEUE_IS_ACCEPT_OP(op)	0
 #  define PJ_IOQUEUE_IS_CONNECT_OP(op)	0
 #endif


 //#define TRACE_(expr) PJ_LOG(3,expr)
 #define TRACE_(expr)


 /*
  * This describes each key.
  */
 struct pj_ioqueue_key_t
 {
     PJ_DECL_LIST_MEMBER(struct pj_ioqueue_key_t)
     pj_sock_t		    fd;
     pj_ioqueue_operation_e  op;
     void		   *user_data;
     pj_ioqueue_callback	    cb;

     void		   *rd_buf;
     unsigned                rd_flags;
     pj_size_t		    rd_buflen;
     void		   *wr_buf;
     pj_size_t		    wr_buflen;

     pj_sockaddr_t	   *rmt_addr;
     int			   *rmt_addrlen;

     pj_sockaddr_t	   *local_addr;
     int			   *local_addrlen;

     pj_sock_t		   *accept_fd;
 };

 /*
  * This describes the I/O queue.
  */
 struct pj_ioqueue_t
 {
     pj_lock_t          *lock;
     pj_bool_t           auto_delete_lock;
     unsigned		max, count;
     pj_ioqueue_key_t	hlist;
     int			epfd;
 };

 /*
  * pj_ioqueue_create()
  *
  * Create select ioqueue.
  */
 PJ_DEF(pj_status_t) pj_ioqueue_create( pj_pool_t *pool,
                                        pj_size_t max_fd,
                                        int max_threads,
                                        pj_ioqueue_t **p_ioqueue)
 {
     pj_ioqueue_t *ioque;
     pj_status_t rc;

     PJ_UNUSED_ARG(max_threads);

     if (max_fd > PJ_IOQUEUE_MAX_HANDLES) {
         pj_assert(!"max_fd too large");
 	return PJ_EINVAL;
     }

     ioque = pj_pool_alloc(pool, sizeof(pj_ioqueue_t));
     ioque->max = max_fd;
     ioque->count = 0;
     pj_list_init(&ioque->hlist);

     rc = pj_lock_create_recursive_mutex(pool, "ioq%p", &ioque->lock);
     if (rc != PJ_SUCCESS)
 	return rc;

     ioque->auto_delete_lock = PJ_TRUE;
     ioque->epfd = os_epoll_create(max_fd);
     if (ioque->epfd < 0) {
 	return PJ_RETURN_OS_ERROR(pj_get_native_os_error());
     }

     PJ_LOG(4, ("pjlib", "select() I/O Queue created (%p)", ioque));

     *p_ioqueue = ioque;
     return PJ_SUCCESS;
 }

 /*
  * pj_ioqueue_destroy()
  *
  * Destroy ioqueue.
  */
 PJ_DEF(pj_status_t) pj_ioqueue_destroy(pj_ioqueue_t *ioque)
 {
     PJ_ASSERT_RETURN(ioque, PJ_EINVAL);
     PJ_ASSERT_RETURN(ioque->epfd > 0, PJ_EINVALIDOP);

     pj_lock_acquire(ioque->lock);
     os_close(ioque->epfd);
     ioque->epfd = 0;
     if (ioque->auto_delete_lock)
         pj_lock_destroy(ioque->lock);

     return PJ_SUCCESS;
 }

 /*
  * pj_ioqueue_set_lock()
  */
 PJ_DEF(pj_status_t) pj_ioqueue_set_lock( pj_ioqueue_t *ioque,
 					 pj_lock_t *lock,
 					 pj_bool_t auto_delete )
 {
     PJ_ASSERT_RETURN(ioque && lock, PJ_EINVAL);

     if (ioque->auto_delete_lock) {
         pj_lock_destroy(ioque->lock);
     }

     ioque->lock = lock;
     ioque->auto_delete_lock = auto_delete;

     return PJ_SUCCESS;
 }


 /*
  * pj_ioqueue_register_sock()
  *
  * Register a socket to ioqueue.
  */
 PJ_DEF(pj_status_t) pj_ioqueue_register_sock( pj_pool_t *pool,
 					      pj_ioqueue_t *ioque,
 					      pj_sock_t sock,
 					      void *user_data,
 					      const pj_ioqueue_callback *cb,
                                               pj_ioqueue_key_t **p_key)
 {
     pj_ioqueue_key_t *key = NULL;
     pj_uint32_t value;
     struct epoll_event ev;
     int status;
     pj_status_t rc = PJ_SUCCESS;

     PJ_ASSERT_RETURN(pool && ioque && sock != PJ_INVALID_SOCKET &&
                      cb && p_key, PJ_EINVAL);

     pj_lock_acquire(ioque->lock);

     if (ioque->count >= ioque->max) {
         rc = PJ_ETOOMANY;
 	TRACE_((THIS_FILE, "pj_ioqueue_register_sock error: too many files"));
 	goto on_return;
     }

     /* Set socket to nonblocking. */
     value = 1;
     if ((rc=os_ioctl(sock, FIONBIO, (ioctl_val_type)&value))) {
 	TRACE_((THIS_FILE, "pj_ioqueue_register_sock error: ioctl rc=%d",
                 rc));
         rc = pj_get_netos_error();
 	goto on_return;
     }

     /* Create key. */
     key = (pj_ioqueue_key_t*)pj_pool_zalloc(pool, sizeof(pj_ioqueue_key_t));
     key->fd = sock;
     key->user_data = user_data;
     pj_memcpy(&key->cb, cb, sizeof(pj_ioqueue_callback));

     /* os_epoll_ctl. */
     ev.events = EPOLLIN | EPOLLOUT | EPOLLERR;
     ev.epoll_data = (epoll_data_type)key;
     status = os_epoll_ctl(ioque->epfd, EPOLL_CTL_ADD, sock, &ev);
     if (status < 0) {
 	rc = pj_get_os_error();
 	TRACE_((THIS_FILE,
                 "pj_ioqueue_register_sock error: os_epoll_ctl rc=%d",
                 status));
 	goto on_return;
     }

     /* Register */
     pj_list_insert_before(&ioque->hlist, key);
     ++ioque->count;

 on_return:
     *p_key = key;
     pj_lock_release(ioque->lock);

     return rc;
 }

 /*
  * pj_ioqueue_unregister()
  *
  * Unregister handle from ioqueue.
  */
 PJ_DEF(pj_status_t) pj_ioqueue_unregister( pj_ioqueue_t *ioque,
 					   pj_ioqueue_key_t *key)
 {
     struct epoll_event ev;
     int status;

     PJ_ASSERT_RETURN(ioque && key, PJ_EINVAL);

     pj_lock_acquire(ioque->lock);

     pj_assert(ioque->count > 0);
     --ioque->count;
     pj_list_erase(key);

     ev.events = 0;
     ev.epoll_data = (epoll_data_type)key;
     status = os_epoll_ctl( ioque->epfd, EPOLL_CTL_DEL, key->fd, &ev);
     if (status != 0) {
 	pj_status_t rc = pj_get_os_error();
 	pj_lock_release(ioque->lock);
 	return rc;
     }

     pj_lock_release(ioque->lock);
     return PJ_SUCCESS;
 }

 /*
  * pj_ioqueue_get_user_data()
  *
  * Obtain value associated with a key.
  */
 PJ_DEF(void*) pj_ioqueue_get_user_data( pj_ioqueue_key_t *key )
 {
     PJ_ASSERT_RETURN(key != NULL, NULL);
     return key->user_data;
 }


 /*
  * pj_ioqueue_poll()
  *
  */
 PJ_DEF(int) pj_ioqueue_poll( pj_ioqueue_t *ioque, const pj_time_val *timeout)
 {
     int i, count, processed;
     struct epoll_event events[16];
     int msec;

     PJ_CHECK_STACK();

     msec = timeout ? PJ_TIME_VAL_MSEC(*timeout) : 9000;

     count = os_epoll_wait( ioque->epfd, events, PJ_ARRAY_SIZE(events), msec);
     if (count <= 0)
 	return count;

     /* Lock ioqueue. */
     pj_lock_acquire(ioque->lock);

     processed = 0;

     for (i=0; i<count; ++i) {
 	pj_ioqueue_key_t *h = (pj_ioqueue_key_t*)(epoll_data_type)
 				events[i].epoll_data;
 	pj_status_t rc;

 	/*
 	 * Check for completion of read operations.
 	 */
 	if ((events[i].events & EPOLLIN) && (PJ_IOQUEUE_IS_READ_OP(h->op))) {
 	    pj_ssize_t bytes_read = h->rd_buflen;

 	    if ((h->op & PJ_IOQUEUE_OP_RECV_FROM)) {
 	        rc = pj_sock_recvfrom( h->fd, h->rd_buf, &bytes_read, 0,
 				       h->rmt_addr, h->rmt_addrlen);
 	    } else if ((h->op & PJ_IOQUEUE_OP_RECV)) {
 	        rc = pj_sock_recv(h->fd, h->rd_buf, &bytes_read, 0);
 	    } else {
 		bytes_read = os_read( h->fd, h->rd_buf, bytes_read);
 		rc = (bytes_read >= 0) ? PJ_SUCCESS : pj_get_os_error();
 	    }

 	    if (rc != PJ_SUCCESS) {
 	        bytes_read = -rc;
 	    }

 	    h->op &= ~(PJ_IOQUEUE_OP_READ | PJ_IOQUEUE_OP_RECV |
 		       PJ_IOQUEUE_OP_RECV_FROM);

 	    /* Call callback. */
 	    (*h->cb.on_read_complete)(h, bytes_read);

 	    ++processed;
 	}
 	/*
 	 * Check for completion of accept() operation.
 	 */
 	else if ((events[i].events & EPOLLIN) &&
 		 (h->op & PJ_IOQUEUE_OP_ACCEPT))
 	{
 	    /* accept() must be the only operation specified on
 	     * server socket
 	     */
 	    pj_assert( h->op == PJ_IOQUEUE_OP_ACCEPT);

 	    rc = pj_sock_accept( h->fd, h->accept_fd,
 			         h->rmt_addr, h->rmt_addrlen);
 	    if (rc==PJ_SUCCESS && h->local_addr) {
 		rc = pj_sock_getsockname(*h->accept_fd, h->local_addr,
 				          h->local_addrlen);
 	    }

 	    h->op &= ~(PJ_IOQUEUE_OP_ACCEPT);

 	    /* Call callback. */
 	    (*h->cb.on_accept_complete)(h, *h->accept_fd, rc);

 	    ++processed;
 	}

 	/*
 	 * Check for completion of write operations.
 	 */
 	if ((events[i].events & EPOLLOUT) && PJ_IOQUEUE_IS_WRITE_OP(h->op)) {
 	    /* Completion of write(), send(), or sendto() operation. */

 	    /* Clear operation. */
 	    h->op &= ~(PJ_IOQUEUE_OP_WRITE | PJ_IOQUEUE_OP_SEND |
                        PJ_IOQUEUE_OP_SEND_TO);

 	    /* Call callback. */
 	    /* All data must have been sent? */
 	    (*h->cb.on_write_complete)(h, h->wr_buflen);

 	    ++processed;
 	}
 #if PJ_HAS_TCP
 	/*
 	 * Check for completion of connect() operation.
 	 */
 	else if ((events[i].events & EPOLLOUT) &&
 		 (h->op & PJ_IOQUEUE_OP_CONNECT))
 	{
 	    /* Completion of connect() operation */
 	    pj_ssize_t bytes_transfered;

 	    /* from connect(2):
 		* On Linux, use getsockopt to read the SO_ERROR option at
 		* level SOL_SOCKET to determine whether connect() completed
 		* successfully (if SO_ERROR is zero).
 		*/
 	    int value;
 	    socklen_t vallen = sizeof(value);
 	    int gs_rc = os_getsockopt(h->fd, SOL_SOCKET, SO_ERROR,
                                       (getsockopt_val_ptr)&value, &vallen);
 	    if (gs_rc != 0) {
 		/* Argh!! What to do now???
 		 * Just indicate that the socket is connected. The
 		 * application will get error as soon as it tries to use
 		 * the socket to send/receive.
 		 */
 		bytes_transfered = 0;
 	    } else {
                 bytes_transfered = value;
 	    }

 	    /* Clear operation. */
 	    h->op &= (~PJ_IOQUEUE_OP_CONNECT);

 	    /* Call callback. */
 	    (*h->cb.on_connect_complete)(h, bytes_transfered);

 	    ++processed;
 	}
 #endif /* PJ_HAS_TCP */

 	/*
 	 * Check for error condition.
 	 */
 	if (events[i].events & EPOLLERR) {
 	    if (h->op & PJ_IOQUEUE_OP_CONNECT) {
 		h->op &= ~PJ_IOQUEUE_OP_CONNECT;

 		/* Call callback. */
 		(*h->cb.on_connect_complete)(h, -1);

 		++processed;
 	    }
 	}
     }

     pj_lock_release(ioque->lock);

     return processed;
 }

 /*
  * pj_ioqueue_read()
  *
  * Start asynchronous read from the descriptor.
  */
 PJ_DEF(pj_status_t) pj_ioqueue_read( pj_ioqueue_t *ioque,
 			             pj_ioqueue_key_t *key,
 			             void *buffer,
 			             pj_size_t buflen)
 {
     PJ_ASSERT_RETURN(ioque && key && buffer, PJ_EINVAL);
     PJ_CHECK_STACK();

     /* For consistency with other ioqueue implementation, we would reject
      * if descriptor has already been submitted for reading before.
      */
     PJ_ASSERT_RETURN(((key->op & PJ_IOQUEUE_OP_READ) == 0 &&
                       (key->op & PJ_IOQUEUE_OP_RECV) == 0 &&
                       (key->op & PJ_IOQUEUE_OP_RECV_FROM) == 0),
                      PJ_EBUSY);

     pj_lock_acquire(ioque->lock);

     key->op |= PJ_IOQUEUE_OP_READ;
     key->rd_flags = 0;
     key->rd_buf = buffer;
     key->rd_buflen = buflen;

     pj_lock_release(ioque->lock);
     return PJ_EPENDING;
 }


 /*
  * pj_ioqueue_recv()
  *
  * Start asynchronous recv() from the socket.
  */
 PJ_DEF(pj_status_t) pj_ioqueue_recv(  pj_ioqueue_t *ioque,
 				      pj_ioqueue_key_t *key,
 				      void *buffer,
 				      pj_size_t buflen,
 				      unsigned flags )
 {
     PJ_ASSERT_RETURN(ioque && key && buffer, PJ_EINVAL);
     PJ_CHECK_STACK();

     /* For consistency with other ioqueue implementation, we would reject
      * if descriptor has already been submitted for reading before.
      */
     PJ_ASSERT_RETURN(((key->op & PJ_IOQUEUE_OP_READ) == 0 &&
                       (key->op & PJ_IOQUEUE_OP_RECV) == 0 &&
                       (key->op & PJ_IOQUEUE_OP_RECV_FROM) == 0),
                      PJ_EBUSY);

     pj_lock_acquire(ioque->lock);

     key->op |= PJ_IOQUEUE_OP_RECV;
     key->rd_buf = buffer;
     key->rd_buflen = buflen;
     key->rd_flags = flags;

     pj_lock_release(ioque->lock);
     return PJ_EPENDING;
 }

 /*
  * pj_ioqueue_recvfrom()
  *
  * Start asynchronous recvfrom() from the socket.
  */
 PJ_DEF(pj_status_t) pj_ioqueue_recvfrom( pj_ioqueue_t *ioque,
 				         pj_ioqueue_key_t *key,
 				         void *buffer,
 				         pj_size_t buflen,
                                          unsigned flags,
 				         pj_sockaddr_t *addr,
 				         int *addrlen)
 {
     PJ_ASSERT_RETURN(ioque && key && buffer, PJ_EINVAL);
     PJ_CHECK_STACK();

     /* For consistency with other ioqueue implementation, we would reject
      * if descriptor has already been submitted for reading before.
      */
     PJ_ASSERT_RETURN(((key->op & PJ_IOQUEUE_OP_READ) == 0 &&
                       (key->op & PJ_IOQUEUE_OP_RECV) == 0 &&
                       (key->op & PJ_IOQUEUE_OP_RECV_FROM) == 0),
                      PJ_EBUSY);

     pj_lock_acquire(ioque->lock);

     key->op |= PJ_IOQUEUE_OP_RECV_FROM;
     key->rd_buf = buffer;
     key->rd_buflen = buflen;
     key->rd_flags = flags;
     key->rmt_addr = addr;
     key->rmt_addrlen = addrlen;

     pj_lock_release(ioque->lock);
     return PJ_EPENDING;
 }

 /*
  * pj_ioqueue_write()
  *
  * Start asynchronous write() to the descriptor.
  */
 PJ_DEF(pj_status_t) pj_ioqueue_write( pj_ioqueue_t *ioque,
 			              pj_ioqueue_key_t *key,
 			              const void *data,
 			              pj_size_t datalen)
 {
     pj_status_t rc;
     pj_ssize_t sent;

     PJ_ASSERT_RETURN(ioque && key && data, PJ_EINVAL);
     PJ_CHECK_STACK();

     /* For consistency with other ioqueue implementation, we would reject
      * if descriptor has already been submitted for writing before.
      */
     PJ_ASSERT_RETURN(((key->op & PJ_IOQUEUE_OP_WRITE) == 0 &&
                       (key->op & PJ_IOQUEUE_OP_SEND) == 0 &&
                       (key->op & PJ_IOQUEUE_OP_SEND_TO) == 0),
                      PJ_EBUSY);

     sent = datalen;
     /* sent would be -1 after pj_sock_send() if it returns error. */
     rc = pj_sock_send(key->fd, data, &sent, 0);
     if (rc != PJ_SUCCESS && rc != PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK)) {
         return rc;
     }

     pj_lock_acquire(ioque->lock);

     key->op |= PJ_IOQUEUE_OP_WRITE;
     key->wr_buf = NULL;
     key->wr_buflen = datalen;

     pj_lock_release(ioque->lock);

     return PJ_EPENDING;
 }

 /*
  * pj_ioqueue_send()
  *
  * Start asynchronous send() to the descriptor.
  */
 PJ_DEF(pj_status_t) pj_ioqueue_send( pj_ioqueue_t *ioque,
 			             pj_ioqueue_key_t *key,
 			             const void *data,
 			             pj_size_t datalen,
                                      unsigned flags)
 {
     pj_status_t rc;
     pj_ssize_t sent;

     PJ_ASSERT_RETURN(ioque && key && data, PJ_EINVAL);
     PJ_CHECK_STACK();

     /* For consistency with other ioqueue implementation, we would reject
      * if descriptor has already been submitted for writing before.
      */
     PJ_ASSERT_RETURN(((key->op & PJ_IOQUEUE_OP_WRITE) == 0 &&
                       (key->op & PJ_IOQUEUE_OP_SEND) == 0 &&
                       (key->op & PJ_IOQUEUE_OP_SEND_TO) == 0),
                      PJ_EBUSY);

     sent = datalen;
     /* sent would be -1 after pj_sock_send() if it returns error. */
     rc = pj_sock_send(key->fd, data, &sent, flags);
     if (rc != PJ_SUCCESS && rc != PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK)) {
         return rc;
     }

     pj_lock_acquire(ioque->lock);

     key->op |= PJ_IOQUEUE_OP_SEND;
     key->wr_buf = NULL;
     key->wr_buflen = datalen;

     pj_lock_release(ioque->lock);

     return PJ_EPENDING;
 }


 /*
  * pj_ioqueue_sendto()
  *
  * Start asynchronous write() to the descriptor.
  */
 PJ_DEF(pj_status_t) pj_ioqueue_sendto( pj_ioqueue_t *ioque,
 			               pj_ioqueue_key_t *key,
 			               const void *data,
 			               pj_size_t datalen,
                                        unsigned flags,
 			               const pj_sockaddr_t *addr,
 			               int addrlen)
 {
     pj_status_t rc;
     pj_ssize_t sent;

     PJ_ASSERT_RETURN(ioque && key && data, PJ_EINVAL);
     PJ_CHECK_STACK();

     /* For consistency with other ioqueue implementation, we would reject
      * if descriptor has already been submitted for writing before.
      */
     PJ_ASSERT_RETURN(((key->op & PJ_IOQUEUE_OP_WRITE) == 0 &&
                       (key->op & PJ_IOQUEUE_OP_SEND) == 0 &&
                       (key->op & PJ_IOQUEUE_OP_SEND_TO) == 0),
                      PJ_EBUSY);

     sent = datalen;
     /* sent would be -1 after pj_sock_sendto() if it returns error. */
     rc = pj_sock_sendto(key->fd, data, &sent, flags, addr, addrlen);
     if (rc != PJ_SUCCESS && rc != PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK))  {
         return rc;
     }

     pj_lock_acquire(ioque->lock);

     key->op |= PJ_IOQUEUE_OP_SEND_TO;
     key->wr_buf = NULL;
     key->wr_buflen = datalen;

     pj_lock_release(ioque->lock);
     return PJ_EPENDING;
 }

 #if PJ_HAS_TCP
 /*
  * Initiate overlapped accept() operation.
  */
 PJ_DEF(int) pj_ioqueue_accept( pj_ioqueue_t *ioqueue,
 			       pj_ioqueue_key_t *key,
 			       pj_sock_t *new_sock,
 			       pj_sockaddr_t *local,
 			       pj_sockaddr_t *remote,
 			       int *addrlen)
 {
     /* check parameters. All must be specified! */
     pj_assert(ioqueue && key && new_sock);

     /* Server socket must have no other operation! */
     pj_assert(key->op == 0);

     pj_lock_acquire(ioqueue->lock);

     key->op = PJ_IOQUEUE_OP_ACCEPT;
     key->accept_fd = new_sock;
     key->rmt_addr = remote;
     key->rmt_addrlen = addrlen;
     key->local_addr = local;
     key->local_addrlen = addrlen;   /* use same addr. as rmt_addrlen */

     pj_lock_release(ioqueue->lock);
     return PJ_EPENDING;
 }

 /*
  * Initiate overlapped connect() operation (well, it's non-blocking actually,
  * since there's no overlapped version of connect()).
  */
 PJ_DEF(pj_status_t) pj_ioqueue_connect( pj_ioqueue_t *ioqueue,
 					pj_ioqueue_key_t *key,
 					const pj_sockaddr_t *addr,
 					int addrlen )
 {
     pj_status_t rc;

     /* check parameters. All must be specified! */
     PJ_ASSERT_RETURN(ioqueue && key && addr && addrlen, PJ_EINVAL);

     /* Connecting socket must have no other operation! */
     PJ_ASSERT_RETURN(key->op == 0, PJ_EBUSY);

     rc = pj_sock_connect(key->fd, addr, addrlen);
     if (rc == PJ_SUCCESS) {
 	/* Connected! */
 	return PJ_SUCCESS;
     } else {
 	if (rc == PJ_STATUS_FROM_OS(OSERR_EINPROGRESS) ||
             rc == PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK))
         {
 	    /* Pending! */
 	    pj_lock_acquire(ioqueue->lock);
 	    key->op = PJ_IOQUEUE_OP_CONNECT;
 	    pj_lock_release(ioqueue->lock);
 	    return PJ_EPENDING;
 	} else {
 	    /* Error! */
 	    return rc;
 	}
     }
 }
 #endif	/* PJ_HAS_TCP */
	/* $Header: /pjproject-0.3/pjlib/src/pj/ioqueue_epoll.c 4 10/29/05 10:27p Bennylp $ */
	/*
	* $Log: /pjproject-0.3/pjlib/src/pj/ioqueue_epoll.c $
	*
	* 4 10/29/05 10:27p Bennylp
	* Fixed misc warnings.
	*
	* 3 10/29/05 11:49a Bennylp
	* Fixed warnings.
	*
	* 2 10/29/05 11:31a Bennylp
	* Changed accept and lock.
	*
	* 1 10/17/05 10:49p Bennylp
	* Created.
	*
	*/

	/*
	* ioqueue_epoll.c
	*
	* This is the implementation of IOQueue framework using /dev/epoll
	* API in _both_ Linux user-mode and kernel-mode.
	*/

	#include <pj/ioqueue.h>
	#include <pj/os.h>
	#include <pj/lock.h>
	#include <pj/log.h>
	#include <pj/list.h>
	#include <pj/pool.h>
	#include <pj/string.h>
	#include <pj/assert.h>
	#include <pj/errno.h>
	#include <pj/sock.h>
	#include <pj/compat/socket.h>

	#if !defined(PJ_LINUX_KERNEL) \|\| PJ_LINUX_KERNEL==0
	/*
	* Linux user mode
	*/
	# include <sys/epoll.h>
	# include <errno.h>
	# include <unistd.h>

	# define epoll_data data.ptr
	# define epoll_data_type void*
	# define ioctl_val_type unsigned long*
	# define getsockopt_val_ptr int*
	# define os_getsockopt getsockopt
	# define os_ioctl ioctl
	# define os_read read
	# define os_close close
	# define os_epoll_create epoll_create
	# define os_epoll_ctl epoll_ctl
	# define os_epoll_wait epoll_wait
	#else
	/*
	* Linux kernel mode.
	*/
	# include <linux/config.h>
	# include <linux/version.h>
	# if defined(MODVERSIONS)
	# include <linux/modversions.h>
	# endif
	# include <linux/kernel.h>
	# include <linux/poll.h>
	# include <linux/eventpoll.h>
	# include <linux/syscalls.h>
	# include <linux/errno.h>
	# include <linux/unistd.h>
	# include <asm/ioctls.h>
	enum EPOLL_EVENTS
	{
	EPOLLIN = 0x001,
	EPOLLOUT = 0x004,
	EPOLLERR = 0x008,
	};
	# define os_epoll_create sys_epoll_create
	static int os_epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)
	{
	long rc;
	mm_segment_t oldfs = get_fs();
	set_fs(KERNEL_DS);
	rc = sys_epoll_ctl(epfd, op, fd, event);
	set_fs(oldfs);
	if (rc) {
	errno = -rc;
	return -1;
	} else {
	return 0;
	}
	}
	static int os_epoll_wait(int epfd, struct epoll_event *events,
	int maxevents, int timeout)
	{
	int count;
	mm_segment_t oldfs = get_fs();
	set_fs(KERNEL_DS);
	count = sys_epoll_wait(epfd, events, maxevents, timeout);
	set_fs(oldfs);
	return count;
	}
	# define os_close sys_close
	# define os_getsockopt pj_sock_getsockopt
	static int os_read(int fd, void *buf, size_t len)
	{
	long rc;
	mm_segment_t oldfs = get_fs();
	set_fs(KERNEL_DS);
	rc = sys_read(fd, buf, len);
	set_fs(oldfs);
	if (rc) {
	errno = -rc;
	return -1;
	} else {
	return 0;
	}
	}
	# define socklen_t unsigned
	# define ioctl_val_type unsigned long
	int ioctl(int fd, int opt, ioctl_val_type value);
	static int os_ioctl(int fd, int opt, ioctl_val_type value)
	{
	int rc;
	mm_segment_t oldfs = get_fs();
	set_fs(KERNEL_DS);
	rc = ioctl(fd, opt, value);
	set_fs(oldfs);
	if (rc < 0) {
	errno = -rc;
	return rc;
	} else
	return rc;
	}
	# define getsockopt_val_ptr char*

	# define epoll_data data
	# define epoll_data_type __u32
	#endif

	#define THIS_FILE "ioq_epoll"

	#define PJ_IOQUEUE_IS_READ_OP(op) ((op & PJ_IOQUEUE_OP_READ) \|\| \
	(op & PJ_IOQUEUE_OP_RECV) \|\| \
	(op & PJ_IOQUEUE_OP_RECV_FROM))
	#define PJ_IOQUEUE_IS_WRITE_OP(op) ((op & PJ_IOQUEUE_OP_WRITE) \|\| \
	(op & PJ_IOQUEUE_OP_SEND) \|\| \
	(op & PJ_IOQUEUE_OP_SEND_TO))


	#if PJ_HAS_TCP
	# define PJ_IOQUEUE_IS_ACCEPT_OP(op) (op & PJ_IOQUEUE_OP_ACCEPT)
	# define PJ_IOQUEUE_IS_CONNECT_OP(op) (op & PJ_IOQUEUE_OP_CONNECT)
	#else
	# define PJ_IOQUEUE_IS_ACCEPT_OP(op) 0
	# define PJ_IOQUEUE_IS_CONNECT_OP(op) 0
	#endif


	//#define TRACE_(expr) PJ_LOG(3,expr)
	#define TRACE_(expr)


	/*
	* This describes each key.
	*/
	struct pj_ioqueue_key_t
	{
	PJ_DECL_LIST_MEMBER(struct pj_ioqueue_key_t)
	pj_sock_t fd;
	pj_ioqueue_operation_e op;
	void *user_data;
	pj_ioqueue_callback cb;

	void *rd_buf;
	unsigned rd_flags;
	pj_size_t rd_buflen;
	void *wr_buf;
	pj_size_t wr_buflen;

	pj_sockaddr_t *rmt_addr;
	int *rmt_addrlen;

	pj_sockaddr_t *local_addr;
	int *local_addrlen;

	pj_sock_t *accept_fd;
	};

	/*
	* This describes the I/O queue.
	*/
	struct pj_ioqueue_t
	{
	pj_lock_t *lock;
	pj_bool_t auto_delete_lock;
	unsigned max, count;
	pj_ioqueue_key_t hlist;
	int epfd;
	};

	/*
	* pj_ioqueue_create()
	*
	* Create select ioqueue.
	*/
	PJ_DEF(pj_status_t) pj_ioqueue_create( pj_pool_t *pool,
	pj_size_t max_fd,
	int max_threads,
	pj_ioqueue_t **p_ioqueue)
	{
	pj_ioqueue_t *ioque;
	pj_status_t rc;

	PJ_UNUSED_ARG(max_threads);

	if (max_fd > PJ_IOQUEUE_MAX_HANDLES) {
	pj_assert(!"max_fd too large");
	return PJ_EINVAL;
	}

	ioque = pj_pool_alloc(pool, sizeof(pj_ioqueue_t));
	ioque->max = max_fd;
	ioque->count = 0;
	pj_list_init(&ioque->hlist);

	rc = pj_lock_create_recursive_mutex(pool, "ioq%p", &ioque->lock);
	if (rc != PJ_SUCCESS)
	return rc;

	ioque->auto_delete_lock = PJ_TRUE;
	ioque->epfd = os_epoll_create(max_fd);
	if (ioque->epfd < 0) {
	return PJ_RETURN_OS_ERROR(pj_get_native_os_error());
	}

	PJ_LOG(4, ("pjlib", "select() I/O Queue created (%p)", ioque));

	*p_ioqueue = ioque;
	return PJ_SUCCESS;
	}

	/*
	* pj_ioqueue_destroy()
	*
	* Destroy ioqueue.
	*/
	PJ_DEF(pj_status_t) pj_ioqueue_destroy(pj_ioqueue_t *ioque)
	{
	PJ_ASSERT_RETURN(ioque, PJ_EINVAL);
	PJ_ASSERT_RETURN(ioque->epfd > 0, PJ_EINVALIDOP);

	pj_lock_acquire(ioque->lock);
	os_close(ioque->epfd);
	ioque->epfd = 0;
	if (ioque->auto_delete_lock)
	pj_lock_destroy(ioque->lock);

	return PJ_SUCCESS;
	}

	/*
	* pj_ioqueue_set_lock()
	*/
	PJ_DEF(pj_status_t) pj_ioqueue_set_lock( pj_ioqueue_t *ioque,
	pj_lock_t *lock,
	pj_bool_t auto_delete )
	{
	PJ_ASSERT_RETURN(ioque && lock, PJ_EINVAL);

	if (ioque->auto_delete_lock) {
	pj_lock_destroy(ioque->lock);
	}

	ioque->lock = lock;
	ioque->auto_delete_lock = auto_delete;

	return PJ_SUCCESS;
	}


	/*
	* pj_ioqueue_register_sock()
	*
	* Register a socket to ioqueue.
	*/
	PJ_DEF(pj_status_t) pj_ioqueue_register_sock( pj_pool_t *pool,
	pj_ioqueue_t *ioque,
	pj_sock_t sock,
	void *user_data,
	const pj_ioqueue_callback *cb,
	pj_ioqueue_key_t **p_key)
	{
	pj_ioqueue_key_t *key = NULL;
	pj_uint32_t value;
	struct epoll_event ev;
	int status;
	pj_status_t rc = PJ_SUCCESS;

	PJ_ASSERT_RETURN(pool && ioque && sock != PJ_INVALID_SOCKET &&
	cb && p_key, PJ_EINVAL);

	pj_lock_acquire(ioque->lock);

	if (ioque->count >= ioque->max) {
	rc = PJ_ETOOMANY;
	TRACE_((THIS_FILE, "pj_ioqueue_register_sock error: too many files"));
	goto on_return;
	}

	/* Set socket to nonblocking. */
	value = 1;
	if ((rc=os_ioctl(sock, FIONBIO, (ioctl_val_type)&value))) {
	TRACE_((THIS_FILE, "pj_ioqueue_register_sock error: ioctl rc=%d",
	rc));
	rc = pj_get_netos_error();
	goto on_return;
	}

	/* Create key. */
	key = (pj_ioqueue_key_t*)pj_pool_zalloc(pool, sizeof(pj_ioqueue_key_t));
	key->fd = sock;
	key->user_data = user_data;
	pj_memcpy(&key->cb, cb, sizeof(pj_ioqueue_callback));

	/* os_epoll_ctl. */
	ev.events = EPOLLIN \| EPOLLOUT \| EPOLLERR;
	ev.epoll_data = (epoll_data_type)key;
	status = os_epoll_ctl(ioque->epfd, EPOLL_CTL_ADD, sock, &ev);
	if (status < 0) {
	rc = pj_get_os_error();
	TRACE_((THIS_FILE,
	"pj_ioqueue_register_sock error: os_epoll_ctl rc=%d",
	status));
	goto on_return;
	}

	/* Register */
	pj_list_insert_before(&ioque->hlist, key);
	++ioque->count;

	on_return:
	*p_key = key;
	pj_lock_release(ioque->lock);

	return rc;
	}

	/*
	* pj_ioqueue_unregister()
	*
	* Unregister handle from ioqueue.
	*/
	PJ_DEF(pj_status_t) pj_ioqueue_unregister( pj_ioqueue_t *ioque,
	pj_ioqueue_key_t *key)
	{
	struct epoll_event ev;
	int status;

	PJ_ASSERT_RETURN(ioque && key, PJ_EINVAL);

	pj_lock_acquire(ioque->lock);

	pj_assert(ioque->count > 0);
	--ioque->count;
	pj_list_erase(key);

	ev.events = 0;
	ev.epoll_data = (epoll_data_type)key;
	status = os_epoll_ctl( ioque->epfd, EPOLL_CTL_DEL, key->fd, &ev);
	if (status != 0) {
	pj_status_t rc = pj_get_os_error();
	pj_lock_release(ioque->lock);
	return rc;
	}

	pj_lock_release(ioque->lock);
	return PJ_SUCCESS;
	}

	/*
	* pj_ioqueue_get_user_data()
	*
	* Obtain value associated with a key.
	*/
	PJ_DEF(void) pj_ioqueue_get_user_data( pj_ioqueue_key_t key )
	{
	PJ_ASSERT_RETURN(key != NULL, NULL);
	return key->user_data;
	}


	/*
	* pj_ioqueue_poll()
	*
	*/
	PJ_DEF(int) pj_ioqueue_poll( pj_ioqueue_t ioque, const pj_time_val timeout)
	{
	int i, count, processed;
	struct epoll_event events[16];
	int msec;

	PJ_CHECK_STACK();

	msec = timeout ? PJ_TIME_VAL_MSEC(*timeout) : 9000;

	count = os_epoll_wait( ioque->epfd, events, PJ_ARRAY_SIZE(events), msec);
	if (count <= 0)
	return count;

	/* Lock ioqueue. */
	pj_lock_acquire(ioque->lock);

	processed = 0;

	for (i=0; i<count; ++i) {
	pj_ioqueue_key_t h = (pj_ioqueue_key_t)(epoll_data_type)
	events[i].epoll_data;
	pj_status_t rc;

	/*
	* Check for completion of read operations.
	*/
	if ((events[i].events & EPOLLIN) && (PJ_IOQUEUE_IS_READ_OP(h->op))) {
	pj_ssize_t bytes_read = h->rd_buflen;

	if ((h->op & PJ_IOQUEUE_OP_RECV_FROM)) {
	rc = pj_sock_recvfrom( h->fd, h->rd_buf, &bytes_read, 0,
	h->rmt_addr, h->rmt_addrlen);
	} else if ((h->op & PJ_IOQUEUE_OP_RECV)) {
	rc = pj_sock_recv(h->fd, h->rd_buf, &bytes_read, 0);
	} else {
	bytes_read = os_read( h->fd, h->rd_buf, bytes_read);
	rc = (bytes_read >= 0) ? PJ_SUCCESS : pj_get_os_error();
	}

	if (rc != PJ_SUCCESS) {
	bytes_read = -rc;
	}

	h->op &= ~(PJ_IOQUEUE_OP_READ \| PJ_IOQUEUE_OP_RECV \|
	PJ_IOQUEUE_OP_RECV_FROM);

	/* Call callback. */
	(*h->cb.on_read_complete)(h, bytes_read);

	++processed;
	}
	/*
	* Check for completion of accept() operation.
	*/
	else if ((events[i].events & EPOLLIN) &&
	(h->op & PJ_IOQUEUE_OP_ACCEPT))
	{
	/* accept() must be the only operation specified on
	* server socket
	*/
	pj_assert( h->op == PJ_IOQUEUE_OP_ACCEPT);

	rc = pj_sock_accept( h->fd, h->accept_fd,
	h->rmt_addr, h->rmt_addrlen);
	if (rc==PJ_SUCCESS && h->local_addr) {
	rc = pj_sock_getsockname(*h->accept_fd, h->local_addr,
	h->local_addrlen);
	}

	h->op &= ~(PJ_IOQUEUE_OP_ACCEPT);

	/* Call callback. */
	(h->cb.on_accept_complete)(h, h->accept_fd, rc);

	++processed;
	}

	/*
	* Check for completion of write operations.
	*/
	if ((events[i].events & EPOLLOUT) && PJ_IOQUEUE_IS_WRITE_OP(h->op)) {
	/* Completion of write(), send(), or sendto() operation. */

	/* Clear operation. */
	h->op &= ~(PJ_IOQUEUE_OP_WRITE \| PJ_IOQUEUE_OP_SEND \|
	PJ_IOQUEUE_OP_SEND_TO);

	/* Call callback. */
	/* All data must have been sent? */
	(*h->cb.on_write_complete)(h, h->wr_buflen);

	++processed;
	}
	#if PJ_HAS_TCP
	/*
	* Check for completion of connect() operation.
	*/
	else if ((events[i].events & EPOLLOUT) &&
	(h->op & PJ_IOQUEUE_OP_CONNECT))
	{
	/* Completion of connect() operation */
	pj_ssize_t bytes_transfered;

	/* from connect(2):
	* On Linux, use getsockopt to read the SO_ERROR option at
	* level SOL_SOCKET to determine whether connect() completed
	* successfully (if SO_ERROR is zero).
	*/
	int value;
	socklen_t vallen = sizeof(value);
	int gs_rc = os_getsockopt(h->fd, SOL_SOCKET, SO_ERROR,
	(getsockopt_val_ptr)&value, &vallen);
	if (gs_rc != 0) {
	/* Argh!! What to do now???
	* Just indicate that the socket is connected. The
	* application will get error as soon as it tries to use
	* the socket to send/receive.
	*/
	bytes_transfered = 0;
	} else {
	bytes_transfered = value;
	}

	/* Clear operation. */
	h->op &= (~PJ_IOQUEUE_OP_CONNECT);

	/* Call callback. */
	(*h->cb.on_connect_complete)(h, bytes_transfered);

	++processed;
	}
	#endif /* PJ_HAS_TCP */

	/*
	* Check for error condition.
	*/
	if (events[i].events & EPOLLERR) {
	if (h->op & PJ_IOQUEUE_OP_CONNECT) {
	h->op &= ~PJ_IOQUEUE_OP_CONNECT;

	/* Call callback. */
	(*h->cb.on_connect_complete)(h, -1);

	++processed;
	}
	}
	}

	pj_lock_release(ioque->lock);

	return processed;
	}

	/*
	* pj_ioqueue_read()
	*
	* Start asynchronous read from the descriptor.
	*/
	PJ_DEF(pj_status_t) pj_ioqueue_read( pj_ioqueue_t *ioque,
	pj_ioqueue_key_t *key,
	void *buffer,
	pj_size_t buflen)
	{
	PJ_ASSERT_RETURN(ioque && key && buffer, PJ_EINVAL);
	PJ_CHECK_STACK();

	/* For consistency with other ioqueue implementation, we would reject
	* if descriptor has already been submitted for reading before.
	*/
	PJ_ASSERT_RETURN(((key->op & PJ_IOQUEUE_OP_READ) == 0 &&
	(key->op & PJ_IOQUEUE_OP_RECV) == 0 &&
	(key->op & PJ_IOQUEUE_OP_RECV_FROM) == 0),
	PJ_EBUSY);

	pj_lock_acquire(ioque->lock);

	key->op \|= PJ_IOQUEUE_OP_READ;
	key->rd_flags = 0;
	key->rd_buf = buffer;
	key->rd_buflen = buflen;

	pj_lock_release(ioque->lock);
	return PJ_EPENDING;
	}


	/*
	* pj_ioqueue_recv()
	*
	* Start asynchronous recv() from the socket.
	*/
	PJ_DEF(pj_status_t) pj_ioqueue_recv( pj_ioqueue_t *ioque,
	pj_ioqueue_key_t *key,
	void *buffer,
	pj_size_t buflen,
	unsigned flags )
	{
	PJ_ASSERT_RETURN(ioque && key && buffer, PJ_EINVAL);
	PJ_CHECK_STACK();

	/* For consistency with other ioqueue implementation, we would reject
	* if descriptor has already been submitted for reading before.
	*/
	PJ_ASSERT_RETURN(((key->op & PJ_IOQUEUE_OP_READ) == 0 &&
	(key->op & PJ_IOQUEUE_OP_RECV) == 0 &&
	(key->op & PJ_IOQUEUE_OP_RECV_FROM) == 0),
	PJ_EBUSY);

	pj_lock_acquire(ioque->lock);

	key->op \|= PJ_IOQUEUE_OP_RECV;
	key->rd_buf = buffer;
	key->rd_buflen = buflen;
	key->rd_flags = flags;

	pj_lock_release(ioque->lock);
	return PJ_EPENDING;
	}

	/*
	* pj_ioqueue_recvfrom()
	*
	* Start asynchronous recvfrom() from the socket.
	*/
	PJ_DEF(pj_status_t) pj_ioqueue_recvfrom( pj_ioqueue_t *ioque,
	pj_ioqueue_key_t *key,
	void *buffer,
	pj_size_t buflen,
	unsigned flags,
	pj_sockaddr_t *addr,
	int *addrlen)
	{
	PJ_ASSERT_RETURN(ioque && key && buffer, PJ_EINVAL);
	PJ_CHECK_STACK();

	/* For consistency with other ioqueue implementation, we would reject
	* if descriptor has already been submitted for reading before.
	*/
	PJ_ASSERT_RETURN(((key->op & PJ_IOQUEUE_OP_READ) == 0 &&
	(key->op & PJ_IOQUEUE_OP_RECV) == 0 &&
	(key->op & PJ_IOQUEUE_OP_RECV_FROM) == 0),
	PJ_EBUSY);

	pj_lock_acquire(ioque->lock);

	key->op \|= PJ_IOQUEUE_OP_RECV_FROM;
	key->rd_buf = buffer;
	key->rd_buflen = buflen;
	key->rd_flags = flags;
	key->rmt_addr = addr;
	key->rmt_addrlen = addrlen;

	pj_lock_release(ioque->lock);
	return PJ_EPENDING;
	}

	/*
	* pj_ioqueue_write()
	*
	* Start asynchronous write() to the descriptor.
	*/
	PJ_DEF(pj_status_t) pj_ioqueue_write( pj_ioqueue_t *ioque,
	pj_ioqueue_key_t *key,
	const void *data,
	pj_size_t datalen)
	{
	pj_status_t rc;
	pj_ssize_t sent;

	PJ_ASSERT_RETURN(ioque && key && data, PJ_EINVAL);
	PJ_CHECK_STACK();

	/* For consistency with other ioqueue implementation, we would reject
	* if descriptor has already been submitted for writing before.
	*/
	PJ_ASSERT_RETURN(((key->op & PJ_IOQUEUE_OP_WRITE) == 0 &&
	(key->op & PJ_IOQUEUE_OP_SEND) == 0 &&
	(key->op & PJ_IOQUEUE_OP_SEND_TO) == 0),
	PJ_EBUSY);

	sent = datalen;
	/* sent would be -1 after pj_sock_send() if it returns error. */
	rc = pj_sock_send(key->fd, data, &sent, 0);
	if (rc != PJ_SUCCESS && rc != PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK)) {
	return rc;
	}

	pj_lock_acquire(ioque->lock);

	key->op \|= PJ_IOQUEUE_OP_WRITE;
	key->wr_buf = NULL;
	key->wr_buflen = datalen;

	pj_lock_release(ioque->lock);

	return PJ_EPENDING;
	}

	/*
	* pj_ioqueue_send()
	*
	* Start asynchronous send() to the descriptor.
	*/
	PJ_DEF(pj_status_t) pj_ioqueue_send( pj_ioqueue_t *ioque,
	pj_ioqueue_key_t *key,
	const void *data,
	pj_size_t datalen,
	unsigned flags)
	{
	pj_status_t rc;
	pj_ssize_t sent;

	PJ_ASSERT_RETURN(ioque && key && data, PJ_EINVAL);
	PJ_CHECK_STACK();

	/* For consistency with other ioqueue implementation, we would reject
	* if descriptor has already been submitted for writing before.
	*/
	PJ_ASSERT_RETURN(((key->op & PJ_IOQUEUE_OP_WRITE) == 0 &&
	(key->op & PJ_IOQUEUE_OP_SEND) == 0 &&
	(key->op & PJ_IOQUEUE_OP_SEND_TO) == 0),
	PJ_EBUSY);

	sent = datalen;
	/* sent would be -1 after pj_sock_send() if it returns error. */
	rc = pj_sock_send(key->fd, data, &sent, flags);
	if (rc != PJ_SUCCESS && rc != PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK)) {
	return rc;
	}

	pj_lock_acquire(ioque->lock);

	key->op \|= PJ_IOQUEUE_OP_SEND;
	key->wr_buf = NULL;
	key->wr_buflen = datalen;

	pj_lock_release(ioque->lock);

	return PJ_EPENDING;
	}


	/*
	* pj_ioqueue_sendto()
	*
	* Start asynchronous write() to the descriptor.
	*/
	PJ_DEF(pj_status_t) pj_ioqueue_sendto( pj_ioqueue_t *ioque,
	pj_ioqueue_key_t *key,
	const void *data,
	pj_size_t datalen,
	unsigned flags,
	const pj_sockaddr_t *addr,
	int addrlen)
	{
	pj_status_t rc;
	pj_ssize_t sent;

	PJ_ASSERT_RETURN(ioque && key && data, PJ_EINVAL);
	PJ_CHECK_STACK();

	/* For consistency with other ioqueue implementation, we would reject
	* if descriptor has already been submitted for writing before.
	*/
	PJ_ASSERT_RETURN(((key->op & PJ_IOQUEUE_OP_WRITE) == 0 &&
	(key->op & PJ_IOQUEUE_OP_SEND) == 0 &&
	(key->op & PJ_IOQUEUE_OP_SEND_TO) == 0),
	PJ_EBUSY);

	sent = datalen;
	/* sent would be -1 after pj_sock_sendto() if it returns error. */
	rc = pj_sock_sendto(key->fd, data, &sent, flags, addr, addrlen);
	if (rc != PJ_SUCCESS && rc != PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK)) {
	return rc;
	}

	pj_lock_acquire(ioque->lock);

	key->op \|= PJ_IOQUEUE_OP_SEND_TO;
	key->wr_buf = NULL;
	key->wr_buflen = datalen;

	pj_lock_release(ioque->lock);
	return PJ_EPENDING;
	}

	#if PJ_HAS_TCP
	/*
	* Initiate overlapped accept() operation.
	*/
	PJ_DEF(int) pj_ioqueue_accept( pj_ioqueue_t *ioqueue,
	pj_ioqueue_key_t *key,
	pj_sock_t *new_sock,
	pj_sockaddr_t *local,
	pj_sockaddr_t *remote,
	int *addrlen)
	{
	/* check parameters. All must be specified! */
	pj_assert(ioqueue && key && new_sock);

	/* Server socket must have no other operation! */
	pj_assert(key->op == 0);

	pj_lock_acquire(ioqueue->lock);

	key->op = PJ_IOQUEUE_OP_ACCEPT;
	key->accept_fd = new_sock;
	key->rmt_addr = remote;
	key->rmt_addrlen = addrlen;
	key->local_addr = local;
	key->local_addrlen = addrlen; /* use same addr. as rmt_addrlen */

	pj_lock_release(ioqueue->lock);
	return PJ_EPENDING;
	}

	/*
	* Initiate overlapped connect() operation (well, it's non-blocking actually,
	* since there's no overlapped version of connect()).
	*/
	PJ_DEF(pj_status_t) pj_ioqueue_connect( pj_ioqueue_t *ioqueue,
	pj_ioqueue_key_t *key,
	const pj_sockaddr_t *addr,
	int addrlen )
	{
	pj_status_t rc;

	/* check parameters. All must be specified! */
	PJ_ASSERT_RETURN(ioqueue && key && addr && addrlen, PJ_EINVAL);

	/* Connecting socket must have no other operation! */
	PJ_ASSERT_RETURN(key->op == 0, PJ_EBUSY);

	rc = pj_sock_connect(key->fd, addr, addrlen);
	if (rc == PJ_SUCCESS) {
	/* Connected! */
	return PJ_SUCCESS;
	} else {
	if (rc == PJ_STATUS_FROM_OS(OSERR_EINPROGRESS) \|\|
	rc == PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK))
	{
	/* Pending! */
	pj_lock_acquire(ioqueue->lock);
	key->op = PJ_IOQUEUE_OP_CONNECT;
	pj_lock_release(ioqueue->lock);
	return PJ_EPENDING;
	} else {
	/* Error! */
	return rc;
	}
	}
	}
	#endif /* PJ_HAS_TCP */