Index: pjproject/main/pjlib/src/pj/ioqueue_select.c
===================================================================
--- pjproject/main/pjlib/src/pj/ioqueue_select.c	(revision 6)
+++ pjproject/main/pjlib/src/pj/ioqueue_select.c	(revision 11)
@@ -35,19 +35,29 @@
 #define THIS_FILE   "ioq_select"
 
-#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
+/*
+ * The select ioqueue relies on socket functions (pj_sock_xxx()) to return
+ * the correct error code.
+ */
+#if PJ_RETURN_OS_ERROR(100) != PJ_STATUS_FROM_OS(100)
+#   error "Error reporting must be enabled for this function to work!"
+#endif
+
+/**
+ * Get the number of descriptors in the set. This is defined in sock_select.c
+ * This function will only return the number of sockets set from PJ_FD_SET
+ * operation. When the set is modified by other means (such as by select()),
+ * the count will not be reflected here.
+ *
+ * That's why don't export this function in the header file, to avoid
+ * misunderstanding.
+ *
+ * @param fdsetp    The descriptor set.
+ *
+ * @return          Number of descriptors in the set.
+ */
+PJ_DECL(pj_size_t) PJ_FD_COUNT(const pj_fd_set_t *fdsetp);
+
+
+
 
 /*
@@ -61,4 +71,58 @@
 #endif
 
+struct generic_operation
+{
+    PJ_DECL_LIST_MEMBER(struct generic_operation);
+    pj_ioqueue_operation_e  op;
+};
+
+struct read_operation
+{
+    PJ_DECL_LIST_MEMBER(struct read_operation);
+    pj_ioqueue_operation_e  op;
+
+    void		   *buf;
+    pj_size_t		    size;
+    unsigned                flags;
+    pj_sockaddr_t	   *rmt_addr;
+    int			   *rmt_addrlen;
+};
+
+struct write_operation
+{
+    PJ_DECL_LIST_MEMBER(struct write_operation);
+    pj_ioqueue_operation_e  op;
+
+    char		   *buf;
+    pj_size_t		    size;
+    pj_ssize_t              written;
+    unsigned                flags;
+    pj_sockaddr_in	    rmt_addr;
+    int			    rmt_addrlen;
+};
+
+#if PJ_HAS_TCP
+struct accept_operation
+{
+    PJ_DECL_LIST_MEMBER(struct accept_operation);
+    pj_ioqueue_operation_e  op;
+
+    pj_sock_t              *accept_fd;
+    pj_sockaddr_t	   *local_addr;
+    pj_sockaddr_t	   *rmt_addr;
+    int			   *addrlen;
+};
+#endif
+
+union operation_key
+{
+    struct generic_operation generic;
+    struct read_operation    read;
+    struct write_operation   write;
+#if PJ_HAS_TCP
+    struct accept_operation  accept;
+#endif
+};
+
 /*
  * This describes each key.
@@ -66,23 +130,15 @@
 struct pj_ioqueue_key_t
 {
-    PJ_DECL_LIST_MEMBER(struct pj_ioqueue_key_t)
+    PJ_DECL_LIST_MEMBER(struct pj_ioqueue_key_t);
+    pj_ioqueue_t           *ioqueue;
     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;
+    int                     connecting;
+    struct read_operation   read_list;
+    struct write_operation  write_list;
+#if PJ_HAS_TCP
+    struct accept_operation accept_list;
+#endif
 };
 
@@ -95,5 +151,5 @@
     pj_bool_t           auto_delete_lock;
     unsigned		max, count;
-    pj_ioqueue_key_t	hlist;
+    pj_ioqueue_key_t	key_list;
     pj_fd_set_t		rfdset;
     pj_fd_set_t		wfdset;
@@ -110,36 +166,37 @@
 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_ioqueue_t *ioqueue;
     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_FD_ZERO(&ioque->rfdset);
-    PJ_FD_ZERO(&ioque->wfdset);
+    /* Check that arguments are valid. */
+    PJ_ASSERT_RETURN(pool != NULL && p_ioqueue != NULL && 
+                     max_fd > 0 && max_fd <= PJ_IOQUEUE_MAX_HANDLES, 
+                     PJ_EINVAL);
+
+    /* Check that size of pj_ioqueue_op_key_t is sufficient */
+    PJ_ASSERT_RETURN(sizeof(pj_ioqueue_op_key_t)-sizeof(void*) >=
+                     sizeof(union operation_key), PJ_EBUG);
+
+    ioqueue = pj_pool_alloc(pool, sizeof(pj_ioqueue_t));
+    ioqueue->max = max_fd;
+    ioqueue->count = 0;
+    PJ_FD_ZERO(&ioqueue->rfdset);
+    PJ_FD_ZERO(&ioqueue->wfdset);
 #if PJ_HAS_TCP
-    PJ_FD_ZERO(&ioque->xfdset);
-#endif
-    pj_list_init(&ioque->hlist);
-
-    rc = pj_lock_create_recursive_mutex(pool, "ioq%p", &ioque->lock);
+    PJ_FD_ZERO(&ioqueue->xfdset);
+#endif
+    pj_list_init(&ioqueue->key_list);
+
+    rc = pj_lock_create_recursive_mutex(pool, "ioq%p", &ioqueue->lock);
     if (rc != PJ_SUCCESS)
 	return rc;
 
-    ioque->auto_delete_lock = PJ_TRUE;
-
-    PJ_LOG(4, ("pjlib", "select() I/O Queue created (%p)", ioque));
-
-    *p_ioqueue = ioque;
+    ioqueue->auto_delete_lock = PJ_TRUE;
+
+    PJ_LOG(4, ("pjlib", "select() I/O Queue created (%p)", ioqueue));
+
+    *p_ioqueue = ioqueue;
     return PJ_SUCCESS;
 }
@@ -150,12 +207,14 @@
  * Destroy ioqueue.
  */
-PJ_DEF(pj_status_t) pj_ioqueue_destroy(pj_ioqueue_t *ioque)
+PJ_DEF(pj_status_t) pj_ioqueue_destroy(pj_ioqueue_t *ioqueue)
 {
     pj_status_t rc = PJ_SUCCESS;
 
-    PJ_ASSERT_RETURN(ioque, PJ_EINVAL);
-
-    if (ioque->auto_delete_lock)
-        rc = pj_lock_destroy(ioque->lock);
+    PJ_ASSERT_RETURN(ioqueue, PJ_EINVAL);
+
+    pj_lock_acquire(ioqueue->lock);
+
+    if (ioqueue->auto_delete_lock)
+        rc = pj_lock_destroy(ioqueue->lock);
 
     return rc;
@@ -164,24 +223,4 @@
 
 /*
- * 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()
  *
@@ -189,5 +228,5 @@
  */
 PJ_DEF(pj_status_t) pj_ioqueue_register_sock( pj_pool_t *pool,
-					      pj_ioqueue_t *ioque,
+					      pj_ioqueue_t *ioqueue,
 					      pj_sock_t sock,
 					      void *user_data,
@@ -199,10 +238,10 @@
     pj_status_t rc = PJ_SUCCESS;
     
-    PJ_ASSERT_RETURN(pool && ioque && sock != PJ_INVALID_SOCKET &&
+    PJ_ASSERT_RETURN(pool && ioqueue && sock != PJ_INVALID_SOCKET &&
                      cb && p_key, PJ_EINVAL);
 
-    pj_lock_acquire(ioque->lock);
-
-    if (ioque->count >= ioque->max) {
+    pj_lock_acquire(ioqueue->lock);
+
+    if (ioqueue->count >= ioqueue->max) {
         rc = PJ_ETOOMANY;
 	goto on_return;
@@ -212,5 +251,5 @@
     value = 1;
 #ifdef PJ_WIN32
-    if (ioctlsocket(sock, FIONBIO, (unsigned long*)&value)) {
+    if (ioctlsocket(sock, FIONBIO, (u_long*)&value)) {
 #else
     if (ioctl(sock, FIONBIO, &value)) {
@@ -222,6 +261,12 @@
     /* Create key. */
     key = (pj_ioqueue_key_t*)pj_pool_zalloc(pool, sizeof(pj_ioqueue_key_t));
+    key->ioqueue = ioqueue;
     key->fd = sock;
     key->user_data = user_data;
+    pj_list_init(&key->read_list);
+    pj_list_init(&key->write_list);
+#if PJ_HAS_TCP
+    pj_list_init(&key->accept_list);
+#endif
 
     /* Save callback. */
@@ -229,10 +274,11 @@
 
     /* Register */
-    pj_list_insert_before(&ioque->hlist, key);
-    ++ioque->count;
+    pj_list_insert_before(&ioqueue->key_list, key);
+    ++ioqueue->count;
 
 on_return:
+    /* On error, socket may be left in non-blocking mode. */
     *p_key = key;
-    pj_lock_release(ioque->lock);
+    pj_lock_release(ioqueue->lock);
     
     return rc;
@@ -244,21 +290,24 @@
  * Unregister handle from ioqueue.
  */
-PJ_DEF(pj_status_t) pj_ioqueue_unregister( pj_ioqueue_t *ioque,
-					   pj_ioqueue_key_t *key)
-{
-    PJ_ASSERT_RETURN(ioque && key, PJ_EINVAL);
-
-    pj_lock_acquire(ioque->lock);
-
-    pj_assert(ioque->count > 0);
-    --ioque->count;
+PJ_DEF(pj_status_t) pj_ioqueue_unregister( pj_ioqueue_key_t *key)
+{
+    pj_ioqueue_t *ioqueue;
+
+    PJ_ASSERT_RETURN(key, PJ_EINVAL);
+
+    ioqueue = key->ioqueue;
+
+    pj_lock_acquire(ioqueue->lock);
+
+    pj_assert(ioqueue->count > 0);
+    --ioqueue->count;
     pj_list_erase(key);
-    PJ_FD_CLR(key->fd, &ioque->rfdset);
-    PJ_FD_CLR(key->fd, &ioque->wfdset);
+    PJ_FD_CLR(key->fd, &ioqueue->rfdset);
+    PJ_FD_CLR(key->fd, &ioqueue->wfdset);
 #if PJ_HAS_TCP
-    PJ_FD_CLR(key->fd, &ioque->xfdset);
-#endif
-
-    pj_lock_release(ioque->lock);
+    PJ_FD_CLR(key->fd, &ioqueue->xfdset);
+#endif
+
+    pj_lock_release(ioqueue->lock);
     return PJ_SUCCESS;
 }
@@ -276,9 +325,26 @@
 
 
+/*
+ * pj_ioqueue_set_user_data()
+ */
+PJ_DEF(pj_status_t) pj_ioqueue_set_user_data( pj_ioqueue_key_t *key,
+                                              void *user_data,
+                                              void **old_data)
+{
+    PJ_ASSERT_RETURN(key, PJ_EINVAL);
+
+    if (old_data)
+        *old_data = key->user_data;
+    key->user_data = user_data;
+
+    return PJ_SUCCESS;
+}
+
+
 /* This supposed to check whether the fd_set values are consistent
  * with the operation currently set in each key.
  */
 #if VALIDATE_FD_SET
-static void validate_sets(const pj_ioqueue_t *ioque,
+static void validate_sets(const pj_ioqueue_t *ioqueue,
 			  const pj_fd_set_t *rfdset,
 			  const pj_fd_set_t *wfdset,
@@ -287,11 +353,9 @@
     pj_ioqueue_key_t *key;
 
-    key = ioque->hlist.next;
-    while (key != &ioque->hlist) {
-	if ((key->op & PJ_IOQUEUE_OP_READ) 
-	    || (key->op & PJ_IOQUEUE_OP_RECV)
-	    || (key->op & PJ_IOQUEUE_OP_RECV_FROM)
+    key = ioqueue->key_list.next;
+    while (key != &ioqueue->key_list) {
+	if (!pj_list_empty(&key->read_list)
 #if defined(PJ_HAS_TCP) && PJ_HAS_TCP != 0
-	    || (key->op & PJ_IOQUEUE_OP_ACCEPT)
+	    || !pj_list_empty(&key->accept_list)
 #endif
 	    ) 
@@ -302,9 +366,7 @@
 	    pj_assert(PJ_FD_ISSET(key->fd, rfdset) == 0);
 	}
-	if ((key->op & PJ_IOQUEUE_OP_WRITE)
-	    || (key->op & PJ_IOQUEUE_OP_SEND)
-	    || (key->op & PJ_IOQUEUE_OP_SEND_TO)
+	if (!pj_list_empty(&key->write_list)
 #if defined(PJ_HAS_TCP) && PJ_HAS_TCP != 0
-	    || (key->op & PJ_IOQUEUE_OP_CONNECT)
+	    || key->connecting
 #endif
 	   )
@@ -316,5 +378,5 @@
 	}
 #if defined(PJ_HAS_TCP) && PJ_HAS_TCP != 0
-	if (key->op & PJ_IOQUEUE_OP_CONNECT)
+	if (key->connecting)
 	{
 	    pj_assert(PJ_FD_ISSET(key->fd, xfdset));
@@ -348,5 +410,5 @@
  *    work on fd_set copy of the ioqueue (not the original one).
  */
-PJ_DEF(int) pj_ioqueue_poll( pj_ioqueue_t *ioque, const pj_time_val *timeout)
+PJ_DEF(int) pj_ioqueue_poll( pj_ioqueue_t *ioqueue, const pj_time_val *timeout)
 {
     pj_fd_set_t rfdset, wfdset, xfdset;
@@ -354,12 +416,17 @@
     pj_ioqueue_key_t *h;
 
+    PJ_ASSERT_RETURN(ioqueue, PJ_EINVAL);
+
     /* Lock ioqueue before making fd_set copies */
-    pj_lock_acquire(ioque->lock);
-
-    if (PJ_FD_COUNT(&ioque->rfdset)==0 &&
-        PJ_FD_COUNT(&ioque->wfdset)==0 &&
-        PJ_FD_COUNT(&ioque->xfdset)==0)
+    pj_lock_acquire(ioqueue->lock);
+
+    /* We will only do select() when there are sockets to be polled.
+     * Otherwise select() will return error.
+     */
+    if (PJ_FD_COUNT(&ioqueue->rfdset)==0 &&
+        PJ_FD_COUNT(&ioqueue->wfdset)==0 &&
+        PJ_FD_COUNT(&ioqueue->xfdset)==0)
     {
-        pj_lock_release(ioque->lock);
+        pj_lock_release(ioqueue->lock);
         if (timeout)
             pj_thread_sleep(PJ_TIME_VAL_MSEC(*timeout));
@@ -368,8 +435,8 @@
 
     /* Copy ioqueue's pj_fd_set_t to local variables. */
-    pj_memcpy(&rfdset, &ioque->rfdset, sizeof(pj_fd_set_t));
-    pj_memcpy(&wfdset, &ioque->wfdset, sizeof(pj_fd_set_t));
+    pj_memcpy(&rfdset, &ioqueue->rfdset, sizeof(pj_fd_set_t));
+    pj_memcpy(&wfdset, &ioqueue->wfdset, sizeof(pj_fd_set_t));
 #if PJ_HAS_TCP
-    pj_memcpy(&xfdset, &ioque->xfdset, sizeof(pj_fd_set_t));
+    pj_memcpy(&xfdset, &ioqueue->xfdset, sizeof(pj_fd_set_t));
 #else
     PJ_FD_ZERO(&xfdset);
@@ -377,9 +444,9 @@
 
 #if VALIDATE_FD_SET
-    validate_sets(ioque, &rfdset, &wfdset, &xfdset);
+    validate_sets(ioqueue, &rfdset, &wfdset, &xfdset);
 #endif
 
     /* Unlock ioqueue before select(). */
-    pj_lock_release(ioque->lock);
+    pj_lock_release(ioqueue->lock);
 
     count = pj_sock_select(FD_SETSIZE, &rfdset, &wfdset, &xfdset, timeout);
@@ -388,170 +455,36 @@
 	return count;
 
-    /* Lock ioqueue again before scanning for signalled sockets. */
-    pj_lock_acquire(ioque->lock);
-
-#if PJ_HAS_TCP
-    /* Scan for exception socket */
-    h = ioque->hlist.next;
-do_except_scan:
-    for ( ; h!=&ioque->hlist; h = h->next) {
-	if ((h->op & PJ_IOQUEUE_OP_CONNECT) && PJ_FD_ISSET(h->fd, &xfdset))
-	    break;
-    }
-    if (h != &ioque->hlist) {
-	/* 'connect()' should be the only operation. */
-	pj_assert((h->op == PJ_IOQUEUE_OP_CONNECT));
-
-	/* Clear operation. */
-	h->op &= ~(PJ_IOQUEUE_OP_CONNECT);
-	PJ_FD_CLR(h->fd, &ioque->wfdset);
-	PJ_FD_CLR(h->fd, &ioque->xfdset);
-        PJ_FD_CLR(h->fd, &wfdset);
-        PJ_FD_CLR(h->fd, &xfdset);
-
-	/* Call callback. */
-        if (h->cb.on_connect_complete)
-	    (*h->cb.on_connect_complete)(h, -1);
-
-        /* Re-scan exception list. */
-        goto do_except_scan;
-    }
-#endif	/* PJ_HAS_TCP */
-
-    /* Scan for readable socket. */
-    h = ioque->hlist.next;
-do_readable_scan:
-    for ( ; h!=&ioque->hlist; h = h->next) {
-	if ((PJ_IOQUEUE_IS_READ_OP(h->op) || PJ_IOQUEUE_IS_ACCEPT_OP(h->op)) && 
-	    PJ_FD_ISSET(h->fd, &rfdset))
+    /* Lock ioqueue again before scanning for signalled sockets. 
+     * We must strictly use recursive mutex since application may invoke
+     * the ioqueue again inside the callback.
+     */
+    pj_lock_acquire(ioqueue->lock);
+
+    /* Scan for writable sockets first to handle piggy-back data
+     * coming with accept().
+     */
+    h = ioqueue->key_list.next;
+do_writable_scan:
+    for ( ; h!=&ioqueue->key_list; h = h->next) {
+	if ( (!pj_list_empty(&h->write_list) || h->connecting)
+	     && PJ_FD_ISSET(h->fd, &wfdset))
         {
 	    break;
         }
     }
-    if (h != &ioque->hlist) {
-        pj_status_t rc;
-
-	pj_assert(PJ_IOQUEUE_IS_READ_OP(h->op) ||
-		  PJ_IOQUEUE_IS_ACCEPT_OP(h->op));
-	
-#	if PJ_HAS_TCP
-	if ((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==0 && h->local_addr) {
-		rc = pj_sock_getsockname(*h->accept_fd, h->local_addr, 
-					 h->local_addrlen);
-	    }
-
-	    h->op &= ~(PJ_IOQUEUE_OP_ACCEPT);
-	    PJ_FD_CLR(h->fd, &ioque->rfdset);
-
-	    /* Call callback. */
-            if (h->cb.on_accept_complete)
-	        (*h->cb.on_accept_complete)(h, *h->accept_fd, rc);
-
-            /* Re-scan readable sockets. */
-            goto do_readable_scan;
-        } 
-        else {
-#	endif
-            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 {
-                /*
-                 * User has specified pj_ioqueue_read().
-                 * On Win32, we should do ReadFile(). But because we got
-                 * here because of select() anyway, user must have put a
-                 * socket descriptor on h->fd, which in this case we can
-                 * just call pj_sock_recv() instead of ReadFile().
-                 * On Unix, user may put a file in h->fd, so we'll have
-                 * to call read() here.
-                 * This may not compile on systems which doesn't have 
-                 * read(). That's why we only specify PJ_LINUX here so
-                 * that error is easier to catch.
-                 */
-#	        if defined(PJ_WIN32) && PJ_WIN32 != 0
-                rc = pj_sock_recv(h->fd, h->rd_buf, &bytes_read, 0);
-#               elif (defined(PJ_LINUX) && PJ_LINUX != 0) || \
-		     (defined(PJ_SUNOS) && PJ_SUNOS != 0)
-                bytes_read = read(h->fd, h->rd_buf, bytes_read);
-                rc = (bytes_read >= 0) ? PJ_SUCCESS : pj_get_os_error();
-#		elif defined(PJ_LINUX_KERNEL) && PJ_LINUX_KERNEL != 0
-                bytes_read = sys_read(h->fd, h->rd_buf, bytes_read);
-                rc = (bytes_read >= 0) ? PJ_SUCCESS : -bytes_read;
-#               else
-#               error "Implement read() for this platform!"
-#               endif
-            }
-	    
-	    if (rc != PJ_SUCCESS) {
-#	        if defined(PJ_WIN32) && PJ_WIN32 != 0
-	        /* On Win32, for UDP, WSAECONNRESET on the receive side 
-	         * indicates that previous sending has triggered ICMP Port 
-	         * Unreachable message.
-	         * But we wouldn't know at this point which one of previous 
-	         * key that has triggered the error, since UDP socket can
-	         * be shared!
-	         * So we'll just ignore it!
-	         */
-
-	        if (rc == PJ_STATUS_FROM_OS(WSAECONNRESET)) {
-		    PJ_LOG(4,(THIS_FILE, 
-                              "Ignored ICMP port unreach. on key=%p", h));
-	        }
-#	        endif
-
-                /* In any case we would report this to caller. */
-                bytes_read = -rc;
-	    }
-
-	    h->op &= ~(PJ_IOQUEUE_OP_READ | PJ_IOQUEUE_OP_RECV | 
-                       PJ_IOQUEUE_OP_RECV_FROM);
-	    PJ_FD_CLR(h->fd, &ioque->rfdset);
-            PJ_FD_CLR(h->fd, &rfdset);
-
-	    /* Call callback. */
-            if (h->cb.on_read_complete)
-	        (*h->cb.on_read_complete)(h, bytes_read);
-
-            /* Re-scan readable sockets. */
-            goto do_readable_scan;
-
-        }
-    }
-
-    /* Scan for writable socket  */
-    h = ioque->hlist.next;
-do_writable_scan:
-    for ( ; h!=&ioque->hlist; h = h->next) {
-	if ((PJ_IOQUEUE_IS_WRITE_OP(h->op) || PJ_IOQUEUE_IS_CONNECT_OP(h->op)) 
-	    && PJ_FD_ISSET(h->fd, &wfdset))
-        {
-	    break;
-        }
-    }
-    if (h != &ioque->hlist) {
-	pj_assert(PJ_IOQUEUE_IS_WRITE_OP(h->op) || 
-		  PJ_IOQUEUE_IS_CONNECT_OP(h->op));
+    if (h != &ioqueue->key_list) {
+	pj_assert(!pj_list_empty(&h->write_list) || h->connecting);
 
 #if defined(PJ_HAS_TCP) && PJ_HAS_TCP!=0
-	if ((h->op & PJ_IOQUEUE_OP_CONNECT)) {
+	if (h->connecting) {
 	    /* Completion of connect() operation */
 	    pj_ssize_t bytes_transfered;
 
-#if (defined(PJ_LINUX) && PJ_LINUX!=0) || \
-    (defined(PJ_LINUX_KERNEL) && PJ_LINUX_KERNEL!=0)
+#if (defined(PJ_HAS_SO_ERROR) && PJ_HAS_SO_ERROR!=0)
 	    /* 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).
-		*/
+	     * 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);
@@ -590,7 +523,7 @@
 
 	    /* Clear operation. */
-	    h->op &= (~PJ_IOQUEUE_OP_CONNECT);
-	    PJ_FD_CLR(h->fd, &ioque->wfdset);
-	    PJ_FD_CLR(h->fd, &ioque->xfdset);
+	    h->connecting = 0;
+	    PJ_FD_CLR(h->fd, &ioqueue->wfdset);
+	    PJ_FD_CLR(h->fd, &ioqueue->xfdset);
 
 	    /* Call callback. */
@@ -604,21 +537,227 @@
 #endif /* PJ_HAS_TCP */
 	{
-	    /* Completion of write(), send(), or sendto() operation. */
-
-	    /* Clear operation. */
-	    h->op &= ~(PJ_IOQUEUE_OP_WRITE | PJ_IOQUEUE_OP_SEND | 
-                       PJ_IOQUEUE_OP_SEND_TO);
-	    PJ_FD_CLR(h->fd, &ioque->wfdset);
+	    /* Socket is writable. */
+            struct write_operation *write_op;
+            pj_ssize_t sent;
+            pj_status_t send_rc;
+
+            /* Get the first in the queue. */
+            write_op = h->write_list.next;
+
+            /* Send the data. */
+            sent = write_op->size - write_op->written;
+            if (write_op->op == PJ_IOQUEUE_OP_SEND) {
+                send_rc = pj_sock_send(h->fd, write_op->buf+write_op->written,
+                                       &sent, write_op->flags);
+            } else if (write_op->op == PJ_IOQUEUE_OP_SEND_TO) {
+                send_rc = pj_sock_sendto(h->fd, 
+                                         write_op->buf+write_op->written,
+                                         &sent, write_op->flags,
+                                         &write_op->rmt_addr, 
+                                         write_op->rmt_addrlen);
+            } else {
+                pj_assert(!"Invalid operation type!");
+                send_rc = PJ_EBUG;
+            }
+
+            if (send_rc == PJ_SUCCESS) {
+                write_op->written += sent;
+            } else {
+                pj_assert(send_rc > 0);
+                write_op->written = -send_rc;
+            }
+
+            /* In any case we don't need to process this descriptor again. */
             PJ_FD_CLR(h->fd, &wfdset);
 
-	    /* Call callback. */
-	    /* All data must have been sent? */
-            if (h->cb.on_write_complete)
-	        (*h->cb.on_write_complete)(h, h->wr_buflen);
-
+            /* Are we finished with this buffer? */
+            if (send_rc!=PJ_SUCCESS || 
+                write_op->written == (pj_ssize_t)write_op->size) 
+            {
+                pj_list_erase(write_op);
+
+                /* Clear operation if there's no more data to send. */
+                if (pj_list_empty(&h->write_list))
+                    PJ_FD_CLR(h->fd, &ioqueue->wfdset);
+
+	        /* Call callback. */
+                if (h->cb.on_write_complete) {
+	            (*h->cb.on_write_complete)(h, 
+                                               (pj_ioqueue_op_key_t*)write_op,
+                                               write_op->written);
+                }
+            }
+	    
             /* Re-scan writable sockets. */
             goto do_writable_scan;
 	}
     }
+
+    /* Scan for readable socket. */
+    h = ioqueue->key_list.next;
+do_readable_scan:
+    for ( ; h!=&ioqueue->key_list; h = h->next) {
+	if ((!pj_list_empty(&h->read_list) 
+#if PJ_HAS_TCP
+             || !pj_list_empty(&h->accept_list)
+#endif
+            ) && PJ_FD_ISSET(h->fd, &rfdset))
+        {
+	    break;
+        }
+    }
+    if (h != &ioqueue->key_list) {
+        pj_status_t rc;
+
+#if PJ_HAS_TCP
+	pj_assert(!pj_list_empty(&h->read_list) || 
+                  !pj_list_empty(&h->accept_list));
+#else
+        pj_assert(!pj_list_empty(&h->read_list));
+#endif
+	
+#	if PJ_HAS_TCP
+	if (!pj_list_empty(&h->accept_list)) {
+
+            struct accept_operation *accept_op;
+	    
+            /* Get one accept operation from the list. */
+	    accept_op = h->accept_list.next;
+            pj_list_erase(accept_op);
+
+	    rc=pj_sock_accept(h->fd, accept_op->accept_fd, 
+                              accept_op->rmt_addr, accept_op->addrlen);
+	    if (rc==PJ_SUCCESS && accept_op->local_addr) {
+		rc = pj_sock_getsockname(*accept_op->accept_fd, 
+                                         accept_op->local_addr,
+					 accept_op->addrlen);
+	    }
+
+	    /* Clear bit in fdset if there is no more pending accept */
+            if (pj_list_empty(&h->accept_list))
+	        PJ_FD_CLR(h->fd, &ioqueue->rfdset);
+
+	    /* Call callback. */
+            if (h->cb.on_accept_complete)
+	        (*h->cb.on_accept_complete)(h, (pj_ioqueue_op_key_t*)accept_op,
+                                            *accept_op->accept_fd, rc);
+
+            /* Re-scan readable sockets. */
+            goto do_readable_scan;
+        }
+        else {
+#	endif
+            struct read_operation *read_op;
+            pj_ssize_t bytes_read;
+
+            pj_assert(!pj_list_empty(&h->read_list));
+
+            /* Get one pending read operation from the list. */
+            read_op = h->read_list.next;
+            pj_list_erase(read_op);
+
+            bytes_read = read_op->size;
+
+	    if ((read_op->op == PJ_IOQUEUE_OP_RECV_FROM)) {
+	        rc = pj_sock_recvfrom(h->fd, read_op->buf, &bytes_read, 0,
+				      read_op->rmt_addr, 
+                                      read_op->rmt_addrlen);
+	    } else if ((read_op->op == PJ_IOQUEUE_OP_RECV)) {
+	        rc = pj_sock_recv(h->fd, read_op->buf, &bytes_read, 0);
+            } else {
+                pj_assert(read_op->op == PJ_IOQUEUE_OP_READ);
+                /*
+                 * User has specified pj_ioqueue_read().
+                 * On Win32, we should do ReadFile(). But because we got
+                 * here because of select() anyway, user must have put a
+                 * socket descriptor on h->fd, which in this case we can
+                 * just call pj_sock_recv() instead of ReadFile().
+                 * On Unix, user may put a file in h->fd, so we'll have
+                 * to call read() here.
+                 * This may not compile on systems which doesn't have 
+                 * read(). That's why we only specify PJ_LINUX here so
+                 * that error is easier to catch.
+                 */
+#	        if defined(PJ_WIN32) && PJ_WIN32 != 0
+                rc = pj_sock_recv(h->fd, read_op->buf, &bytes_read, 0);
+                //rc = ReadFile((HANDLE)h->fd, read_op->buf, read_op->size,
+                //              &bytes_read, NULL);
+#               elif (defined(PJ_HAS_UNISTD_H) && PJ_HAS_UNISTD_H != 0)
+                bytes_read = read(h->fd, h->rd_buf, bytes_read);
+                rc = (bytes_read >= 0) ? PJ_SUCCESS : pj_get_os_error();
+#		elif defined(PJ_LINUX_KERNEL) && PJ_LINUX_KERNEL != 0
+                bytes_read = sys_read(h->fd, h->rd_buf, bytes_read);
+                rc = (bytes_read >= 0) ? PJ_SUCCESS : -bytes_read;
+#               else
+#               error "Implement read() for this platform!"
+#               endif
+            }
+	    
+	    if (rc != PJ_SUCCESS) {
+#	        if defined(PJ_WIN32) && PJ_WIN32 != 0
+	        /* On Win32, for UDP, WSAECONNRESET on the receive side 
+	         * indicates that previous sending has triggered ICMP Port 
+	         * Unreachable message.
+	         * But we wouldn't know at this point which one of previous 
+	         * key that has triggered the error, since UDP socket can
+	         * be shared!
+	         * So we'll just ignore it!
+	         */
+
+	        if (rc == PJ_STATUS_FROM_OS(WSAECONNRESET)) {
+		    //PJ_LOG(4,(THIS_FILE, 
+                    //          "Ignored ICMP port unreach. on key=%p", h));
+	        }
+#	        endif
+
+                /* In any case we would report this to caller. */
+                bytes_read = -rc;
+	    }
+
+            /* Clear fdset if there is no pending read. */
+            if (pj_list_empty(&h->read_list))
+	        PJ_FD_CLR(h->fd, &ioqueue->rfdset);
+
+            /* In any case clear from temporary set. */
+            PJ_FD_CLR(h->fd, &rfdset);
+
+	    /* Call callback. */
+            if (h->cb.on_read_complete)
+	        (*h->cb.on_read_complete)(h, (pj_ioqueue_op_key_t*)read_op,
+                                          bytes_read);
+
+            /* Re-scan readable sockets. */
+            goto do_readable_scan;
+
+        }
+    }
+
+#if PJ_HAS_TCP
+    /* Scan for exception socket for TCP connection error. */
+    h = ioqueue->key_list.next;
+do_except_scan:
+    for ( ; h!=&ioqueue->key_list; h = h->next) {
+	if (h->connecting && PJ_FD_ISSET(h->fd, &xfdset))
+	    break;
+    }
+    if (h != &ioqueue->key_list) {
+
+	pj_assert(h->connecting);
+
+	/* Clear operation. */
+	h->connecting = 0;
+	PJ_FD_CLR(h->fd, &ioqueue->wfdset);
+	PJ_FD_CLR(h->fd, &ioqueue->xfdset);
+        PJ_FD_CLR(h->fd, &wfdset);
+        PJ_FD_CLR(h->fd, &xfdset);
+
+	/* Call callback. */
+        if (h->cb.on_connect_complete)
+	    (*h->cb.on_connect_complete)(h, -1);
+
+        /* Re-scan exception list. */
+        goto do_except_scan;
+    }
+#endif	/* PJ_HAS_TCP */
 
     /* Shouldn't happen. */
@@ -629,76 +768,64 @@
     //count = 0;
 
-    pj_lock_release(ioque->lock);
+    pj_lock_release(ioqueue->lock);
     return count;
 }
 
 /*
- * 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_ioqueue_recv()
+ *
+ * Start asynchronous recv() from the socket.
+ */
+PJ_DEF(pj_status_t) pj_ioqueue_recv(  pj_ioqueue_key_t *key,
+                                      pj_ioqueue_op_key_t *op_key,
+				      void *buffer,
+				      pj_ssize_t *length,
+				      unsigned flags )
+{
+    pj_status_t status;
+    pj_ssize_t size;
+    struct read_operation *read_op;
+    pj_ioqueue_t *ioqueue;
+
+    PJ_ASSERT_RETURN(key && op_key && buffer && length, 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_FD_SET(key->fd, &ioque->rfdset);
-
-    pj_lock_release(ioque->lock);
+    /* Try to see if there's data immediately available. 
+     */
+    size = *length;
+    status = pj_sock_recv(key->fd, buffer, &size, flags);
+    if (status == PJ_SUCCESS) {
+        /* Yes! Data is available! */
+        *length = size;
+        return PJ_SUCCESS;
+    } else {
+        /* If error is not EWOULDBLOCK (or EAGAIN on Linux), report
+         * the error to caller.
+         */
+        if (status != PJ_STATUS_FROM_OS(PJ_BLOCKING_ERROR_VAL))
+            return status;
+    }
+
+    /*
+     * No data is immediately available.
+     * Must schedule asynchronous operation to the ioqueue.
+     */
+    ioqueue = key->ioqueue;
+    pj_lock_acquire(ioqueue->lock);
+
+    read_op = (struct read_operation*)op_key;
+
+    read_op->op = PJ_IOQUEUE_OP_RECV;
+    read_op->buf = buffer;
+    read_op->size = *length;
+    read_op->flags = flags;
+
+    pj_list_insert_before(&key->read_list, read_op);
+    PJ_FD_SET(key->fd, &ioqueue->rfdset);
+
+    pj_lock_release(ioqueue->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_FD_SET(key->fd, &ioque->rfdset);
-
-    pj_lock_release(ioque->lock);
-    return PJ_EPENDING;
-}
-
 /*
  * pj_ioqueue_recvfrom()
@@ -706,167 +833,215 @@
  * Start asynchronous recvfrom() from the socket.
  */
-PJ_DEF(pj_status_t) pj_ioqueue_recvfrom( pj_ioqueue_t *ioque,
-				         pj_ioqueue_key_t *key,
+PJ_DEF(pj_status_t) pj_ioqueue_recvfrom( pj_ioqueue_key_t *key,
+                                         pj_ioqueue_op_key_t *op_key,
 				         void *buffer,
-				         pj_size_t buflen,
+				         pj_ssize_t *length,
                                          unsigned flags,
 				         pj_sockaddr_t *addr,
 				         int *addrlen)
 {
-    PJ_ASSERT_RETURN(ioque && key && buffer, PJ_EINVAL);
+    pj_status_t status;
+    pj_ssize_t size;
+    struct read_operation *read_op;
+    pj_ioqueue_t *ioqueue;
+
+    PJ_ASSERT_RETURN(key && op_key && buffer && length, 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_FD_SET(key->fd, &ioque->rfdset);
-
-    pj_lock_release(ioque->lock);
+    /* Try to see if there's data immediately available. 
+     */
+    size = *length;
+    status = pj_sock_recvfrom(key->fd, buffer, &size, flags,
+                              addr, addrlen);
+    if (status == PJ_SUCCESS) {
+        /* Yes! Data is available! */
+        *length = size;
+        return PJ_SUCCESS;
+    } else {
+        /* If error is not EWOULDBLOCK (or EAGAIN on Linux), report
+         * the error to caller.
+         */
+        if (status != PJ_STATUS_FROM_OS(PJ_BLOCKING_ERROR_VAL))
+            return status;
+    }
+
+    /*
+     * No data is immediately available.
+     * Must schedule asynchronous operation to the ioqueue.
+     */
+    ioqueue = key->ioqueue;
+    pj_lock_acquire(ioqueue->lock);
+
+    read_op = (struct read_operation*)op_key;
+
+    read_op->op = PJ_IOQUEUE_OP_RECV_FROM;
+    read_op->buf = buffer;
+    read_op->size = *length;
+    read_op->flags = flags;
+    read_op->rmt_addr = addr;
+    read_op->rmt_addrlen = addrlen;
+
+    pj_list_insert_before(&key->read_list, read_op);
+    PJ_FD_SET(key->fd, &ioqueue->rfdset);
+
+    pj_lock_release(ioqueue->lock);
     return PJ_EPENDING;
 }
 
 /*
- * pj_ioqueue_write()
+ * pj_ioqueue_send()
+ *
+ * Start asynchronous send() to the descriptor.
+ */
+PJ_DEF(pj_status_t) pj_ioqueue_send( pj_ioqueue_key_t *key,
+                                     pj_ioqueue_op_key_t *op_key,
+			             const void *data,
+			             pj_ssize_t *length,
+                                     unsigned flags)
+{
+    pj_ioqueue_t *ioqueue;
+    struct write_operation *write_op;
+    pj_status_t status;
+    pj_ssize_t sent;
+
+    PJ_ASSERT_RETURN(key && op_key && data && length, PJ_EINVAL);
+    PJ_CHECK_STACK();
+
+    /* Fast track:
+     *   Try to send data immediately, only if there's no pending write!
+     * Note:
+     *  We are speculating that the list is empty here without properly
+     *  acquiring ioqueue's mutex first. This is intentional, to maximize
+     *  performance via parallelism.
+     *
+     *  This should be safe, because:
+     *      - by convention, we require caller to make sure that the
+     *        key is not unregistered while other threads are invoking
+     *        an operation on the same key.
+     *      - pj_list_empty() is safe to be invoked by multiple threads,
+     *        even when other threads are modifying the list.
+     */
+    if (pj_list_empty(&key->write_list)) {
+        /*
+         * See if data can be sent immediately.
+         */
+        sent = *length;
+        status = pj_sock_send(key->fd, data, &sent, flags);
+        if (status == PJ_SUCCESS) {
+            /* Success! */
+            *length = sent;
+            return PJ_SUCCESS;
+        } else {
+            /* If error is not EWOULDBLOCK (or EAGAIN on Linux), report
+             * the error to caller.
+             */
+            if (status != PJ_STATUS_FROM_OS(PJ_BLOCKING_ERROR_VAL)) {
+                return status;
+            }
+        }
+    }
+
+    /*
+     * Schedule asynchronous send.
+     */
+    ioqueue = key->ioqueue;
+    pj_lock_acquire(ioqueue->lock);
+
+    write_op = (struct write_operation*)op_key;
+    write_op->op = PJ_IOQUEUE_OP_SEND;
+    write_op->buf = NULL;
+    write_op->size = *length;
+    write_op->written = 0;
+    write_op->flags = flags;
+    
+    pj_list_insert_before(&key->write_list, write_op);
+    PJ_FD_SET(key->fd, &ioqueue->wfdset);
+
+    pj_lock_release(ioqueue->lock);
+
+    return PJ_EPENDING;
+}
+
+
+/*
+ * pj_ioqueue_sendto()
  *
  * 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_FD_SET(key->fd, &ioque->wfdset);
-
-    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_FD_SET(key->fd, &ioque->wfdset);
-
-    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,
+PJ_DEF(pj_status_t) pj_ioqueue_sendto( pj_ioqueue_key_t *key,
+                                       pj_ioqueue_op_key_t *op_key,
 			               const void *data,
-			               pj_size_t datalen,
+			               pj_ssize_t *length,
                                        unsigned flags,
 			               const pj_sockaddr_t *addr,
 			               int addrlen)
 {
-    pj_status_t rc;
+    pj_ioqueue_t *ioqueue;
+    struct write_operation *write_op;
+    pj_status_t status;
     pj_ssize_t sent;
 
-    PJ_ASSERT_RETURN(ioque && key && data, PJ_EINVAL);
+    PJ_ASSERT_RETURN(key && op_key && data && length, 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_FD_SET(key->fd, &ioque->wfdset);
-
-    pj_lock_release(ioque->lock);
+    /* Fast track:
+     *   Try to send data immediately, only if there's no pending write!
+     * Note:
+     *  We are speculating that the list is empty here without properly
+     *  acquiring ioqueue's mutex first. This is intentional, to maximize
+     *  performance via parallelism.
+     *
+     *  This should be safe, because:
+     *      - by convention, we require caller to make sure that the
+     *        key is not unregistered while other threads are invoking
+     *        an operation on the same key.
+     *      - pj_list_empty() is safe to be invoked by multiple threads,
+     *        even when other threads are modifying the list.
+     */
+    if (pj_list_empty(&key->write_list)) {
+        /*
+         * See if data can be sent immediately.
+         */
+        sent = *length;
+        status = pj_sock_sendto(key->fd, data, &sent, flags, addr, addrlen);
+        if (status == PJ_SUCCESS) {
+            /* Success! */
+            *length = sent;
+            return PJ_SUCCESS;
+        } else {
+            /* If error is not EWOULDBLOCK (or EAGAIN on Linux), report
+             * the error to caller.
+             */
+            if (status != PJ_STATUS_FROM_OS(PJ_BLOCKING_ERROR_VAL)) {
+                return status;
+            }
+        }
+    }
+
+    /*
+     * Check that address storage can hold the address parameter.
+     */
+    PJ_ASSERT_RETURN(addrlen <= sizeof(pj_sockaddr_in), PJ_EBUG);
+
+    /*
+     * Schedule asynchronous send.
+     */
+    ioqueue = key->ioqueue;
+    pj_lock_acquire(ioqueue->lock);
+
+    write_op = (struct write_operation*)op_key;
+    write_op->op = PJ_IOQUEUE_OP_SEND_TO;
+    write_op->buf = NULL;
+    write_op->size = *length;
+    write_op->written = 0;
+    write_op->flags = flags;
+    pj_memcpy(&write_op->rmt_addr, addr, addrlen);
+    write_op->rmt_addrlen = addrlen;
+    
+    pj_list_insert_before(&key->write_list, write_op);
+    PJ_FD_SET(key->fd, &ioqueue->wfdset);
+
+    pj_lock_release(ioqueue->lock);
+
     return PJ_EPENDING;
 }
@@ -876,29 +1051,65 @@
  * 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)
-{
+PJ_DEF(pj_status_t) pj_ioqueue_accept( pj_ioqueue_key_t *key,
+                                       pj_ioqueue_op_key_t *op_key,
+			               pj_sock_t *new_sock,
+			               pj_sockaddr_t *local,
+			               pj_sockaddr_t *remote,
+			               int *addrlen)
+{
+    pj_ioqueue_t *ioqueue;
+    struct accept_operation *accept_op;
+    pj_status_t status;
+
     /* 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_ASSERT_RETURN(key && op_key && new_sock, PJ_EINVAL);
+
+    /* Fast track:
+     *  See if there's new connection available immediately.
+     */
+    if (pj_list_empty(&key->accept_list)) {
+        status = pj_sock_accept(key->fd, new_sock, remote, addrlen);
+        if (status == PJ_SUCCESS) {
+            /* Yes! New connection is available! */
+            if (local && addrlen) {
+                status = pj_sock_getsockname(*new_sock, local, addrlen);
+                if (status != PJ_SUCCESS) {
+                    pj_sock_close(*new_sock);
+                    *new_sock = PJ_INVALID_SOCKET;
+                    return status;
+                }
+            }
+            return PJ_SUCCESS;
+        } else {
+            /* If error is not EWOULDBLOCK (or EAGAIN on Linux), report
+             * the error to caller.
+             */
+            if (status != PJ_STATUS_FROM_OS(PJ_BLOCKING_ERROR_VAL)) {
+                return status;
+            }
+        }
+    }
+
+    /*
+     * No connection is available immediately.
+     * Schedule accept() operation to be completed when there is incoming
+     * connection available.
+     */
+    ioqueue = key->ioqueue;
+    accept_op = (struct accept_operation*)op_key;
+
     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 */
-
+    accept_op->op = PJ_IOQUEUE_OP_ACCEPT;
+    accept_op->accept_fd = new_sock;
+    accept_op->rmt_addr = remote;
+    accept_op->addrlen= addrlen;
+    accept_op->local_addr = local;
+
+    pj_list_insert_before(&key->accept_list, accept_op);
     PJ_FD_SET(key->fd, &ioqueue->rfdset);
 
     pj_lock_release(ioqueue->lock);
+
     return PJ_EPENDING;
 }
@@ -908,28 +1119,28 @@
  * 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,
+PJ_DEF(pj_status_t) pj_ioqueue_connect( pj_ioqueue_key_t *key,
 					const pj_sockaddr_t *addr,
 					int addrlen )
 {
-    pj_status_t rc;
+    pj_ioqueue_t *ioqueue;
+    pj_status_t status;
     
     /* 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);
+    PJ_ASSERT_RETURN(key && addr && addrlen, PJ_EINVAL);
+
+    /* Check if socket has not been marked for connecting */
+    if (key->connecting != 0)
+        return PJ_EPENDING;
     
-    rc = pj_sock_connect(key->fd, addr, addrlen);
-    if (rc == PJ_SUCCESS) {
+    status = pj_sock_connect(key->fd, addr, addrlen);
+    if (status == PJ_SUCCESS) {
 	/* Connected! */
 	return PJ_SUCCESS;
     } else {
-	if (rc == PJ_STATUS_FROM_OS(OSERR_EINPROGRESS) || 
-            rc == PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK)) 
-        {
+	if (status == PJ_STATUS_FROM_OS(PJ_BLOCKING_CONNECT_ERROR_VAL)) {
 	    /* Pending! */
+            ioqueue = key->ioqueue;
 	    pj_lock_acquire(ioqueue->lock);
-	    key->op = PJ_IOQUEUE_OP_CONNECT;
+	    key->connecting = PJ_TRUE;
 	    PJ_FD_SET(key->fd, &ioqueue->wfdset);
 	    PJ_FD_SET(key->fd, &ioqueue->xfdset);
@@ -938,5 +1149,5 @@
 	} else {
 	    /* Error! */
-	    return rc;
+	    return status;
 	}
     }
