/**
 * Close UDP transport.
 */
PJ_DEF(pj_status_t) pjmedia_transport_udp_close(pjmedia_transport *tp)
{
    struct transport_udp *udp = (struct transport_udp*) tp;

    /* Sanity check */
    PJ_ASSERT_RETURN(tp, PJ_EINVAL);

    /* Must not close while application is using this */
    PJ_ASSERT_RETURN(!udp->attached, PJ_EINVALIDOP);
    

    if (udp->rtp_key) {
	pj_ioqueue_unregister(udp->rtp_key);
	udp->rtp_key = NULL;
    } else if (udp->rtp_sock != PJ_INVALID_SOCKET) {
	pj_sock_close(udp->rtp_sock);
	udp->rtp_sock = PJ_INVALID_SOCKET;
    }

    if (udp->rtcp_key) {
	pj_ioqueue_unregister(udp->rtcp_key);
	udp->rtcp_key = NULL;
    } else if (udp->rtcp_sock != PJ_INVALID_SOCKET) {
	pj_sock_close(udp->rtcp_sock);
	udp->rtcp_sock = PJ_INVALID_SOCKET;
    }

    pj_pool_release(udp->pool);

    return PJ_SUCCESS;
}

 //
 // Unregister this handler from the ioqueue.
 //
 void unregister()
 {
     if (key_) {
         pj_ioqueue_unregister(key_);
         key_ = NULL;
     }
 }

/*
 * Destroy listener.
 */
static pj_status_t udp_destroy(pj_turn_listener *listener)
{
    struct udp_listener *udp = (struct udp_listener *)listener;
    unsigned i;

    if (udp->key) {
	pj_ioqueue_unregister(udp->key);
	udp->key = NULL;
	udp->base.sock = PJ_INVALID_SOCKET;
    } else if (udp->base.sock != PJ_INVALID_SOCKET) {
	pj_sock_close(udp->base.sock);
	udp->base.sock = PJ_INVALID_SOCKET;
    }

    for (i=0; i<udp->read_cnt; ++i) {
	if (udp->read_op[i]->pkt.pool) {
	    pj_pool_t *rpool = udp->read_op[i]->pkt.pool;
	    udp->read_op[i]->pkt.pool = NULL;
	    pj_pool_release(rpool);
	}
    }

    if (udp->base.pool) {
	pj_pool_t *pool = udp->base.pool;

	PJ_LOG(4,(udp->base.obj_name, "Listener %s destroyed", 
		  udp->base.info));

	udp->base.pool = NULL;
	pj_pool_release(pool);
    }
    return PJ_SUCCESS;
}

/*
 * Destroy listener.
 */
static pj_status_t lis_destroy(pj_turn_listener *listener)
{
    struct tcp_listener *tcp_lis = (struct tcp_listener *)listener;
    unsigned i;

    if (tcp_lis->key) {
	pj_ioqueue_unregister(tcp_lis->key);
	tcp_lis->key = NULL;
	tcp_lis->base.sock = PJ_INVALID_SOCKET;
    } else if (tcp_lis->base.sock != PJ_INVALID_SOCKET) {
	pj_sock_close(tcp_lis->base.sock);
	tcp_lis->base.sock = PJ_INVALID_SOCKET;
    }

    for (i=0; i<tcp_lis->accept_cnt; ++i) {
	/* Nothing to do */
    }

    if (tcp_lis->base.pool) {
	pj_pool_t *pool = tcp_lis->base.pool;

	PJ_LOG(4,(tcp_lis->base.obj_name, "Listener %s destroyed", 
		  tcp_lis->base.info));

	tcp_lis->base.pool = NULL;
	pj_pool_release(pool);
    }
    return PJ_SUCCESS;
}

 //
 // Unregister handler.
 //
 static void unregister_handler(Pj_Event_Handler *handler)
 {
     if (handler->key_) {
         pj_ioqueue_unregister( handler->key_ );
         handler->key_ = NULL;
     }
 }

/*
 * udp_destroy()
 *
 * This function is called by transport manager (by transport->destroy()).
 */
static pj_status_t udp_destroy( pjsip_transport *transport )
{
    struct udp_transport *tp = (struct udp_transport*)transport;
    int i;

    /* Mark this transport as closing. */
    tp->is_closing = 1;

    /* Cancel all pending operations. */
    /* blp: NO NO NO...
     *      No need to post queued completion as we poll the ioqueue until
     *      we've got events anyway. Posting completion will only cause
     *      callback to be called twice with IOCP: one for the post completion
     *      and another one for closing the socket.
     *
    for (i=0; i<tp->rdata_cnt; ++i) {
	pj_ioqueue_post_completion(tp->key, 
				   &tp->rdata[i]->tp_info.op_key.op_key, -1);
    }
    */

    /* Unregister from ioqueue. */
    if (tp->key) {
	pj_ioqueue_unregister(tp->key);
	tp->key = NULL;
    } else {
	/* Close socket. */
	if (tp->sock && tp->sock != PJ_INVALID_SOCKET) {
	    pj_sock_close(tp->sock);
	    tp->sock = PJ_INVALID_SOCKET;
	}
    }

    /* Must poll ioqueue because IOCP calls the callback when socket
     * is closed. We poll the ioqueue until all pending callbacks 
     * have been called.
     */
    for (i=0; i<50 && tp->is_closing < 1+tp->rdata_cnt; ++i) {
	int cnt;
	pj_time_val timeout = {0, 1};

	cnt = pj_ioqueue_poll(pjsip_endpt_get_ioqueue(transport->endpt), 
			      &timeout);
	if (cnt == 0)
	    break;
    }

    if (tp->grp_lock) {
	pj_grp_lock_t *grp_lock = tp->grp_lock;
	tp->grp_lock = NULL;
	pj_grp_lock_dec_ref(grp_lock);
	/* Transport may have been deleted at this point */
    } else {
	udp_on_destroy(tp);
    }

    return PJ_SUCCESS;
}

PJ_DEF(pj_status_t) pj_activesock_close(pj_activesock_t *asock)
{
    PJ_ASSERT_RETURN(asock, PJ_EINVAL);
    if (asock->key) {
	pj_ioqueue_unregister(asock->key);
	asock->key = NULL;
    }
    return PJ_SUCCESS;
}

/* Close UDP socket */
static void close_sock(pj_dns_resolver *resv)
{
    /* Close existing socket */
    if (resv->udp_key != NULL) {
	pj_ioqueue_unregister(resv->udp_key);
	resv->udp_key = NULL;
	resv->udp_sock = PJ_INVALID_SOCKET;
    } else if (resv->udp_sock != PJ_INVALID_SOCKET) {
	pj_sock_close(resv->udp_sock);
	resv->udp_sock = PJ_INVALID_SOCKET;
    }
}

/* Timer callback */
static void on_sess_timer(pj_timer_heap_t *th,
			     pj_timer_entry *te)
{
    nat_detect_session *sess;

    sess = (nat_detect_session*) te->user_data;

    if (te->id == TIMER_DESTROY) {
	pj_grp_lock_acquire(sess->grp_lock);
	pj_ioqueue_unregister(sess->key);
	sess->key = NULL;
	sess->sock = PJ_INVALID_SOCKET;
	te->id = 0;
	pj_grp_lock_release(sess->grp_lock);

	sess_destroy(sess);

    } else if (te->id == TIMER_TEST) {

	pj_bool_t next_timer;

	pj_grp_lock_acquire(sess->grp_lock);

	next_timer = PJ_FALSE;

	if (sess->timer_executed == 0) {
	    send_test(sess, ST_TEST_1, NULL, 0);
	    next_timer = PJ_TRUE;
	} else if (sess->timer_executed == 1) {
	    send_test(sess, ST_TEST_2, NULL, CHANGE_IP_PORT_FLAG);
	    next_timer = PJ_TRUE;
	} else if (sess->timer_executed == 2) {
	    send_test(sess, ST_TEST_3, NULL, CHANGE_PORT_FLAG);
	} else {
	    pj_assert(!"Shouldn't have timer at this state");
	}

	++sess->timer_executed;

	if (next_timer) {
	    pj_time_val delay = {0, TEST_INTERVAL};
	    pj_timer_heap_schedule(th, te, &delay);
	} else {
	    te->id = 0;
	}

	pj_grp_lock_release(sess->grp_lock);

    } else {
	pj_assert(!"Invalid timer ID");
    }
}

PJ_DEF(pj_status_t) pj_activesock_close(pj_activesock_t *asock)
{
    PJ_ASSERT_RETURN(asock, PJ_EINVAL);
    if (asock->key) {
#if defined(PJ_IPHONE_OS_HAS_MULTITASKING_SUPPORT) && \
    PJ_IPHONE_OS_HAS_MULTITASKING_SUPPORT!=0
	activesock_destroy_iphone_os_stream(asock);
#endif	
	
	pj_ioqueue_unregister(asock->key);
	asock->key = NULL;
    }
    return PJ_SUCCESS;
}

/*
 * Temporarily pause or shutdown the transport. 
 */
PJ_DEF(pj_status_t) pjsip_udp_transport_pause(pjsip_transport *transport,
					      unsigned option)
{
    struct udp_transport *tp;
    unsigned i;

    PJ_ASSERT_RETURN(transport != NULL, PJ_EINVAL);

    /* Flag must be specified */
    PJ_ASSERT_RETURN((option & 0x03) != 0, PJ_EINVAL);

    tp = (struct udp_transport*) transport;

    /* Transport must not have been paused */
    PJ_ASSERT_RETURN(tp->is_paused==0, PJ_EINVALIDOP);

    /* Set transport to paused first, so that when the read callback is 
     * called by pj_ioqueue_post_completion() it will not try to
     * re-register the rdata.
     */
    tp->is_paused = PJ_TRUE;

    /* Cancel the ioqueue operation. */
    for (i=0; i<(unsigned)tp->rdata_cnt; ++i) {
	pj_ioqueue_post_completion(tp->key, 
				   &tp->rdata[i]->tp_info.op_key.op_key, -1);
    }

    /* Destroy the socket? */
    if (option & PJSIP_UDP_TRANSPORT_DESTROY_SOCKET) {
	if (tp->key) {
	    /* This implicitly closes the socket */
	    pj_ioqueue_unregister(tp->key);
	    tp->key = NULL;
	} else {
	    /* Close socket. */
	    if (tp->sock && tp->sock != PJ_INVALID_SOCKET) {
		pj_sock_close(tp->sock);
		tp->sock = PJ_INVALID_SOCKET;
	    }
	}
	tp->sock = PJ_INVALID_SOCKET;
	
    }

    PJ_LOG(4,(tp->base.obj_name, "SIP UDP transport paused"));

    return PJ_SUCCESS;
}

static void sess_destroy(nat_detect_session *sess)
{
    if (sess->stun_sess) { 
	pj_stun_session_destroy(sess->stun_sess);
	sess->stun_sess = NULL;
    }

    if (sess->key) {
	pj_ioqueue_unregister(sess->key);
	sess->key = NULL;
	sess->sock = PJ_INVALID_SOCKET;
    } else if (sess->sock && sess->sock != PJ_INVALID_SOCKET) {
	pj_sock_close(sess->sock);
	sess->sock = PJ_INVALID_SOCKET;
    }

    if (sess->grp_lock) {
	pj_grp_lock_dec_ref(sess->grp_lock);
    }
}

static void sess_destroy(nat_detect_session *sess)
{
    if (sess->stun_sess) { 
	pj_stun_session_destroy(sess->stun_sess);
    }

    if (sess->key) {
	pj_ioqueue_unregister(sess->key);
    } else if (sess->sock && sess->sock != PJ_INVALID_SOCKET) {
	pj_sock_close(sess->sock);
    }

    if (sess->mutex) {
	pj_mutex_destroy(sess->mutex);
    }

    if (sess->pool) {
	pj_pool_release(sess->pool);
    }
}

/* Destroy relay resource */
static void destroy_relay(pj_turn_relay_res *relay)
{
    if (relay->timer.id) {
	pj_timer_heap_cancel(relay->allocation->server->core.timer_heap,
			     &relay->timer);
	relay->timer.id = PJ_FALSE;
    }

    if (relay->tp.key) {
	pj_ioqueue_unregister(relay->tp.key);
	relay->tp.key = NULL;
	relay->tp.sock = PJ_INVALID_SOCKET;
    } else if (relay->tp.sock != PJ_INVALID_SOCKET) {
	pj_sock_close(relay->tp.sock);
	relay->tp.sock = PJ_INVALID_SOCKET;
    }

    /* Mark as shutdown */
    relay->lifetime = 0;
}

/*
 * Destroy DNS resolver instance.
 */
PJ_DEF(pj_status_t) pj_dns_resolver_destroy( pj_dns_resolver *resolver,
					     pj_bool_t notify)
{
    pj_hash_iterator_t it_buf, *it;
    PJ_ASSERT_RETURN(resolver, PJ_EINVAL);

    if (notify) {
	/*
	 * Notify pending queries if requested.
	 */
	it = pj_hash_first(resolver->hquerybyid, &it_buf);
	while (it) {
	    pj_dns_async_query *q = (pj_dns_async_query *)
	    			    pj_hash_this(resolver->hquerybyid, it);
	    pj_dns_async_query *cq;
	    if (q->cb)
		(*q->cb)(q->user_data, PJ_ECANCELLED, NULL);

	    cq = q->child_head.next;
	    while (cq != (pj_dns_async_query*)&q->child_head) {
		if (cq->cb)
		    (*cq->cb)(cq->user_data, PJ_ECANCELLED, NULL);
		cq = cq->next;
	    }
	    it = pj_hash_next(resolver->hquerybyid, it);
	}
    }

    /* Destroy cached entries */
    it = pj_hash_first(resolver->hrescache, &it_buf);
    while (it) {
	struct cached_res *cache;

	cache = (struct cached_res*) pj_hash_this(resolver->hrescache, it);
	pj_hash_set(NULL, resolver->hrescache, &cache->key, 
		    sizeof(cache->key), 0, NULL);
	pj_pool_release(cache->pool);

	it = pj_hash_first(resolver->hrescache, &it_buf);
    }

    if (resolver->own_timer && resolver->timer) {
	pj_timer_heap_destroy(resolver->timer);
	resolver->timer = NULL;
    }

    if (resolver->udp_key != NULL) {
	pj_ioqueue_unregister(resolver->udp_key);
	resolver->udp_key = NULL;
	resolver->udp_sock = PJ_INVALID_SOCKET;
    } else if (resolver->udp_sock != PJ_INVALID_SOCKET) {
	pj_sock_close(resolver->udp_sock);
	resolver->udp_sock = PJ_INVALID_SOCKET;
    }

    if (resolver->own_ioqueue && resolver->ioqueue) {
	pj_ioqueue_destroy(resolver->ioqueue);
	resolver->ioqueue = NULL;
    }

    if (resolver->mutex) {
	pj_mutex_destroy(resolver->mutex);
	resolver->mutex = NULL;
    }

    if (resolver->pool) {
	pj_pool_t *pool = resolver->pool;
	resolver->pool = NULL;
	pj_pool_release(pool);
    }
    return PJ_SUCCESS;
}

//.........这里部分代码省略.........

        rc = pj_thread_create( pool, NULL, 
                               &worker_thread, 
                               arg, 
                               PJ_THREAD_DEFAULT_STACK_SIZE, 
                               PJ_THREAD_SUSPENDED, &thread[i] );
        if (rc != PJ_SUCCESS) {
            app_perror("...error: unable to create thread", rc);
            return -80;
        }
    }

    /* Mark start time. */
    rc = pj_get_timestamp(&start);
    if (rc != PJ_SUCCESS)
        return -90;

    /* Start the thread. */
    TRACE_((THIS_FILE, "     resuming all threads.."));
    for (i=0; i<thread_cnt; ++i) {
        rc = pj_thread_resume(thread[i]);
        if (rc != 0)
            return -100;
    }

    /* Wait for MSEC_DURATION seconds. 
     * This should be as simple as pj_thread_sleep(MSEC_DURATION) actually,
     * but unfortunately it doesn't work when system doesn't employ
     * timeslicing for threads.
     */
    TRACE_((THIS_FILE, "     wait for few seconds.."));
    do {
	pj_thread_sleep(1);

	/* Mark end time. */
	rc = pj_get_timestamp(&stop);

	if (thread_quit_flag) {
	    TRACE_((THIS_FILE, "      transfer limit reached.."));
	    break;
	}

	if (pj_elapsed_usec(&start,&stop)<MSEC_DURATION * 1000) {
	    TRACE_((THIS_FILE, "      time limit reached.."));
	    break;
	}

    } while (1);

    /* Terminate all threads. */
    TRACE_((THIS_FILE, "     terminating all threads.."));
    thread_quit_flag = 1;

    for (i=0; i<thread_cnt; ++i) {
	TRACE_((THIS_FILE, "      join thread %d..", i));
        pj_thread_join(thread[i]);
    }

    /* Close all sockets. */
    TRACE_((THIS_FILE, "     closing all sockets.."));
    for (i=0; i<sockpair_cnt; ++i) {
        pj_ioqueue_unregister(items[i].server_key);
        pj_ioqueue_unregister(items[i].client_key);
    }

    /* Destroy threads */
    for (i=0; i<thread_cnt; ++i) {
        pj_thread_destroy(thread[i]);
    }

    /* Destroy ioqueue. */
    TRACE_((THIS_FILE, "     destroying ioqueue.."));
    pj_ioqueue_destroy(ioqueue);

    /* Calculate actual time in usec. */
    total_elapsed_usec = pj_elapsed_usec(&start, &stop);

    /* Calculate total bytes received. */
    total_received = 0;
    for (i=0; i<sockpair_cnt; ++i) {
        total_received = (pj_uint32_t)items[i].bytes_recv;
    }

    /* bandwidth = total_received*1000/total_elapsed_usec */
    bandwidth = total_received;
    pj_highprec_mul(bandwidth, 1000);
    pj_highprec_div(bandwidth, total_elapsed_usec);
    
    *p_bandwidth = (pj_uint32_t)bandwidth;

    PJ_LOG(3,(THIS_FILE, "   %.4s    %2d        %2d       %8d KB/s",
              type_name, thread_cnt, sockpair_cnt,
              *p_bandwidth));

    /* Done. */
    pj_pool_release(pool);

    TRACE_((THIS_FILE, "    done.."));
    return 0;
}

PJ_DEF(pj_status_t) pjsip_udp_transport_restart2(pjsip_transport *transport,
					         unsigned option,
					         pj_sock_t sock,
					         const pj_sockaddr *local,
					         const pjsip_host_port *a_name)
{
    struct udp_transport *tp;
    pj_status_t status;
    char addr[PJ_INET6_ADDRSTRLEN+10];
    int i;

    PJ_ASSERT_RETURN(transport != NULL, PJ_EINVAL);
    /* Flag must be specified */
    PJ_ASSERT_RETURN((option & 0x03) != 0, PJ_EINVAL);

    tp = (struct udp_transport*) transport;

    /* Pause the transport first, so that any active read loop spin will
     * quit as soon as possible.
     */
    tp->is_paused = PJ_TRUE;
    
    if (option & PJSIP_UDP_TRANSPORT_DESTROY_SOCKET) {
	char addr_buf[PJ_INET6_ADDRSTRLEN];
	pjsip_host_port bound_name;

	/* Request to recreate transport */

	/* Destroy existing socket, if any. */
	if (tp->key) {
	    /* This implicitly closes the socket */
	    pj_ioqueue_unregister(tp->key);
	    tp->key = NULL;
	} else {
	    /* Close socket. */
	    if (tp->sock && tp->sock != PJ_INVALID_SOCKET) {
		pj_sock_close(tp->sock);
		tp->sock = PJ_INVALID_SOCKET;
	    }
	}
	tp->sock = PJ_INVALID_SOCKET;

	/* Create the socket if it's not specified */
	if (sock == PJ_INVALID_SOCKET) {
	    status = create_socket(local?local->addr.sa_family:pj_AF_UNSPEC(), 
				   local, local?pj_sockaddr_get_len(local):0, 
				   &sock);
	    if (status != PJ_SUCCESS)
		return status;
	}

	/* If transport published name is not specified, calculate it
	 * from the bound address.
	 */
	if (a_name == NULL) {
	    status = get_published_name(sock, addr_buf, sizeof(addr_buf),
					&bound_name);
	    if (status != PJ_SUCCESS) {
		pj_sock_close(sock);
		return status;
	    }

	    a_name = &bound_name;
	}

        /* Init local address. */
        status = pj_sock_getsockname(sock, &tp->base.local_addr, 
				     &tp->base.addr_len);
        if (status != PJ_SUCCESS) {
            pj_sock_close(sock);
            return status;
        }

	/* Assign the socket and published address to transport. */
	udp_set_socket(tp, sock, a_name);

    } else {

	/* For KEEP_SOCKET, transport must have been paused before */
	PJ_ASSERT_RETURN(tp->is_paused, PJ_EINVALIDOP);

	/* If address name is specified, update it */
	if (a_name != NULL)
	    udp_set_pub_name(tp, a_name);
    }

    /* Make sure all udp_on_read_complete() loop spin are stopped */
    do {
	pj_thread_sleep(1);
    } while (tp->read_loop_spin);

    /* Re-register new or existing socket to ioqueue. */
    status = register_to_ioqueue(tp);
    if (status != PJ_SUCCESS) {
	return status;
    }

    /* Re-init op_key. */
    for (i = 0; i < tp->rdata_cnt; ++i) {
	pj_ioqueue_op_key_init(&tp->rdata[i]->tp_info.op_key.op_key,
//.........这里部分代码省略.........

//.........这里部分代码省略.........
		app_perror("...error: pj_ioqueue_sendto()",(pj_status_t)-bytes);
		break;
	    }
	}

	// Begin time.
	pj_get_timestamp(&t1);

	// Poll the queue until we've got completion event in the server side.
        callback_read_key = NULL;
        callback_read_size = 0;
	TRACE__((THIS_FILE, "     waiting for key = %p", skey));
	do {
	    pj_time_val timeout = { 1, 0 };
#ifdef PJ_SYMBIAN
	    rc = pj_symbianos_poll(-1, PJ_TIME_VAL_MSEC(timeout));
#else
	    rc = pj_ioqueue_poll(ioque, &timeout);
#endif
	    TRACE__((THIS_FILE, "     poll rc=%d", rc));
	} while (rc >= 0 && callback_read_key != skey);

	// End time.
	pj_get_timestamp(&t2);
	t_elapsed.u64 += (t2.u64 - t1.u64);

	if (rc < 0) {
	    app_perror("   error: pj_ioqueue_poll", -rc);
	    break;
	}

	// Compare recv buffer with send buffer.
	if (callback_read_size != bufsize || 
	    pj_memcmp(send_buf, recv_buf, bufsize)) 
	{
	    rc = -10;
	    PJ_LOG(3,(THIS_FILE, "   error: size/buffer mismatch"));
	    break;
	}

	// Poll until all events are exhausted, before we start the next loop.
	do {
	    pj_time_val timeout = { 0, 10 };
#ifdef PJ_SYMBIAN
	    PJ_UNUSED_ARG(timeout);
	    rc = pj_symbianos_poll(-1, 100);
#else	    
	    rc = pj_ioqueue_poll(ioque, &timeout);
#endif
	} while (rc>0);

	rc = 0;
    }

    // Print results
    if (rc == 0) {
	pj_timestamp tzero;
	pj_uint32_t usec_delay;

	tzero.u32.hi = tzero.u32.lo = 0;
	usec_delay = pj_elapsed_usec( &tzero, &t_elapsed);

	PJ_LOG(3, (THIS_FILE, "...%10d %15d  % 9d", 
	           bufsize, inactive_sock_count, usec_delay));

    } else {
	PJ_LOG(2, (THIS_FILE, "...ERROR rc=%d (buf:%d, fds:%d)", 
			      rc, bufsize, inactive_sock_count+2));
    }

    // Cleaning up.
    for (i=inactive_sock_count-1; i>=0; --i) {
	pj_ioqueue_unregister(keys[i]);
    }

    pj_ioqueue_unregister(skey);
    pj_ioqueue_unregister(ckey);


    pj_ioqueue_destroy(ioque);
    pj_pool_release( pool);
    return rc;

on_error:
    PJ_LOG(1,(THIS_FILE, "...ERROR: %s", 
	      pj_strerror(pj_get_netos_error(), errbuf, sizeof(errbuf))));
    if (ssock)
	pj_sock_close(ssock);
    if (csock)
	pj_sock_close(csock);
    for (i=0; i<inactive_sock_count && inactive_sock && 
	      inactive_sock[i]!=PJ_INVALID_SOCKET; ++i) 
    {
	pj_sock_close(inactive_sock[i]);
    }
    if (ioque != NULL)
	pj_ioqueue_destroy(ioque);
    pj_pool_release( pool);
    return -1;
}

//.........这里部分代码省略.........
                if (callback_accept_status != 0) {
                    status=-41; goto on_error;
                }
                if (callback_accept_key != skey) {
                    status=-42; goto on_error;
                }
                if (callback_accept_op != &accept_op) {
                    status=-43; goto on_error;
                }
                callback_accept_status = -2;
            }

            if (callback_connect_status != -2) {
                if (callback_connect_status != 0) {
                    status=-50; goto on_error;
                }
                if (callback_connect_key != ckey1) {
                    status=-51; goto on_error;
                }
                callback_connect_status = -2;
            }

	    if (status > pending_op) {
		PJ_LOG(3,(THIS_FILE,
			  "...error: pj_ioqueue_poll() returned %d "
			  "(only expecting %d)",
			  status, pending_op));
		return -52;
	    }
	    pending_op -= status;

	    if (pending_op == 0) {
		status = 0;
	    }
	}
    }

    // There's no pending operation.
    // When we poll the ioqueue, there must not be events.
    if (pending_op == 0) {
        pj_time_val timeout = {1, 0};
#ifdef PJ_SYMBIAN
	status = pj_symbianos_poll(-1, 1000);
#else
        status = pj_ioqueue_poll(ioque, &timeout);
#endif
        if (status != 0) {
            status=-60; goto on_error;
        }
    }

    // Check accepted socket.
    if (csock0 == PJ_INVALID_SOCKET) {
	status = -69;
        app_perror("...accept() error", pj_get_os_error());
	goto on_error;
    }

    // Register newly accepted socket.
    rc = pj_ioqueue_register_sock(pool, ioque, csock0, NULL, 
                                  &test_cb, &ckey0);
    if (rc != PJ_SUCCESS) {
        app_perror("...ERROR in pj_ioqueue_register_sock", rc);
	status = -70;
	goto on_error;
    }

    // Test send and receive.
    t_elapsed.u32.lo = 0;
    status = send_recv_test(ioque, ckey0, ckey1, send_buf, 
                            recv_buf, bufsize, &t_elapsed);
    if (status != 0) {
	goto on_error;
    }

    // Success
    status = 0;

on_error:
    if (skey != NULL)
    	pj_ioqueue_unregister(skey);
    else if (ssock != PJ_INVALID_SOCKET)
	pj_sock_close(ssock);
    
    if (ckey1 != NULL)
    	pj_ioqueue_unregister(ckey1);
    else if (csock1 != PJ_INVALID_SOCKET)
	pj_sock_close(csock1);
    
    if (ckey0 != NULL)
    	pj_ioqueue_unregister(ckey0);
    else if (csock0 != PJ_INVALID_SOCKET)
	pj_sock_close(csock0);
    
    if (ioque != NULL)
	pj_ioqueue_destroy(ioque);
    pj_pool_release(pool);
    return status;

}

/*
 * Testing with many handles.
 * This will just test registering PJ_IOQUEUE_MAX_HANDLES count
 * of sockets to the ioqueue.
 */
static int many_handles_test(pj_bool_t allow_concur)
{
    enum { MAX = PJ_IOQUEUE_MAX_HANDLES };
    pj_pool_t *pool;
    pj_ioqueue_t *ioqueue;
    pj_sock_t *sock;
    pj_ioqueue_key_t **key;
    pj_status_t rc;
    int count, i; /* must be signed */

    PJ_LOG(3,(THIS_FILE,"...testing with so many handles"));

    pool = pj_pool_create(mem, NULL, 4000, 4000, NULL);
    if (!pool)
	return PJ_ENOMEM;

    key = (pj_ioqueue_key_t**) 
    	  pj_pool_alloc(pool, MAX*sizeof(pj_ioqueue_key_t*));
    sock = (pj_sock_t*) pj_pool_alloc(pool, MAX*sizeof(pj_sock_t));
    
    /* Create IOQueue */
    rc = pj_ioqueue_create(pool, MAX, &ioqueue);
    if (rc != PJ_SUCCESS || ioqueue == NULL) {
	app_perror("...error in pj_ioqueue_create", rc);
	return -10;
    }

    // Set concurrency
    rc = pj_ioqueue_set_default_concurrency(ioqueue, allow_concur);
    if (rc != PJ_SUCCESS) {
	return -11;
    }

    /* Register as many sockets. */
    for (count=0; count<MAX; ++count) {
	sock[count] = PJ_INVALID_SOCKET;
	rc = pj_sock_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, &sock[count]);
	if (rc != PJ_SUCCESS || sock[count] == PJ_INVALID_SOCKET) {
	    PJ_LOG(3,(THIS_FILE, "....unable to create %d-th socket, rc=%d", 
				 count, rc));
	    break;
	}
	key[count] = NULL;
	rc = pj_ioqueue_register_sock(pool, ioqueue, sock[count],
				      NULL, &test_cb, &key[count]);
	if (rc != PJ_SUCCESS || key[count] == NULL) {
	    PJ_LOG(3,(THIS_FILE, "....unable to register %d-th socket, rc=%d", 
				 count, rc));
	    return -30;
	}
    }

    /* Test complete. */

    /* Now deregister and close all handles. */ 

    /* NOTE for RTEMS:
     *  It seems that the order of close(sock) is pretty important here.
     *  If we close the sockets with the same order as when they were created,
     *  RTEMS doesn't seem to reuse the sockets, thus next socket created
     *  will have descriptor higher than the last socket created.
     *  If we close the sockets in the reverse order, then the descriptor will
     *  get reused.
     *  This used to cause problem with select ioqueue, since the ioqueue
     *  always gives FD_SETSIZE for the first select() argument. This ioqueue
     *  behavior can be changed with setting PJ_SELECT_NEEDS_NFDS macro.
     */
    for (i=count-1; i>=0; --i) {
    ///for (i=0; i<count; ++i) {
	rc = pj_ioqueue_unregister(key[i]);
	if (rc != PJ_SUCCESS) {
	    app_perror("...error in pj_ioqueue_unregister", rc);
	}
    }

    rc = pj_ioqueue_destroy(ioqueue);
    if (rc != PJ_SUCCESS) {
	app_perror("...error in pj_ioqueue_destroy", rc);
    }
    
    pj_pool_release(pool);

    PJ_LOG(3,(THIS_FILE,"....many_handles_test() ok"));

    return 0;
}