|
|
|
|
Home --> Documentations --> PJLIB Reference Test: I/O Queue (UDP) This file provides implementation to test the functionality of the I/O queue when UDP socket is used. This file is pjlib-test/ioq_udp.c /*
* Copyright (C) 2008-2011 Teluu Inc. (http://www.teluu.com)
* Copyright (C) 2003-2008 Benny Prijono <benny@prijono.org>
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 2 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*/
#include "test.h"
#if INCLUDE_UDP_IOQUEUE_TEST
#include <pjlib.h>
#include <pj/compat/socket.h>
#define THIS_FILE "test_udp"
#define PORT 51233
#define LOOP 2
#define BUF_MIN_SIZE 32
#define BUF_MAX_SIZE 2048
#define SOCK_INACTIVE_MIN (1)
#define SOCK_INACTIVE_MAX (PJ_IOQUEUE_MAX_HANDLES - 2)
#define POOL_SIZE (2*BUF_MAX_SIZE + SOCK_INACTIVE_MAX*128 + 2048)
#undef TRACE_
#define TRACE_(msg) PJ_LOG(3,(THIS_FILE,"....." msg))
#if 0
# define TRACE__(args) PJ_LOG(3,args)
#else
# define TRACE__(args)
#endif
static pj_ssize_t callback_read_size,
callback_write_size,
callback_accept_status,
callback_connect_status;
static pj_ioqueue_key_t *callback_read_key,
*callback_write_key,
*callback_accept_key,
*callback_connect_key;
static pj_ioqueue_op_key_t *callback_read_op,
*callback_write_op,
*callback_accept_op;
pj_ssize_t bytes_read)
{
callback_read_key = key;
callback_read_op = op_key;
callback_read_size = bytes_read;
TRACE__((THIS_FILE, " callback_read_key = %p, bytes=%d",
key, bytes_read));
}
pj_ssize_t bytes_written)
{
callback_write_key = key;
callback_write_op = op_key;
callback_write_size = bytes_written;
}
pj_sock_t sock, int status)
{
PJ_UNUSED_ARG(sock);
callback_accept_key = key;
callback_accept_op = op_key;
callback_accept_status = status;
}
{
callback_connect_key = key;
callback_connect_status = status;
}
static pj_ioqueue_callback test_cb =
{
&on_ioqueue_read,
&on_ioqueue_write,
&on_ioqueue_accept,
&on_ioqueue_connect,
};
#if defined(PJ_WIN32) || defined(PJ_WIN64)
# define S_ADDR S_un.S_addr
#else
# define S_ADDR s_addr
#endif
/*
* compliance_test()
* To test that the basic IOQueue functionality works. It will just exchange
* data between two sockets.
*/
{
pj_sock_t ssock=-1, csock=-1;
pj_sockaddr_in addr, dst_addr;
int addrlen;
pj_pool_t *pool = NULL;
char *send_buf, *recv_buf;
pj_ioqueue_t *ioque = NULL;
pj_ioqueue_key_t *skey = NULL, *ckey = NULL;
pj_ioqueue_op_key_t read_op, write_op;
int bufsize = BUF_MIN_SIZE;
pj_ssize_t bytes;
int status = -1;
pj_str_t temp;
pj_bool_t send_pending, recv_pending;
pj_status_t rc;
// Create pool.
pool = pj_pool_create(mem, NULL, POOL_SIZE, 4000, NULL);
// Allocate buffers for send and receive.
send_buf = (char*)pj_pool_alloc(pool, bufsize);
recv_buf = (char*)pj_pool_alloc(pool, bufsize);
// Allocate sockets for sending and receiving.
TRACE_("creating sockets...");
if (rc==PJ_SUCCESS)
else
csock = PJ_INVALID_SOCKET;
if (rc != PJ_SUCCESS) {
app_perror("...ERROR in pj_sock_socket()", rc);
status=-1; goto on_error;
}
// Bind server socket.
TRACE_("bind socket...");
pj_bzero(&addr, sizeof(addr));
addr.sin_family = pj_AF_INET();
status=-10; goto on_error;
}
// Create I/O Queue.
TRACE_("create ioqueue...");
rc = pj_ioqueue_create2(pool, PJ_IOQUEUE_MAX_HANDLES, cfg, &ioque);
if (rc != PJ_SUCCESS) {
status=-20; goto on_error;
}
// Register server and client socket.
// We put this after inactivity socket, hopefully this can represent the
// worst waiting time.
TRACE_("registering first sockets...");
rc = pj_ioqueue_register_sock(pool, ioque, ssock, NULL,
&test_cb, &skey);
if (rc != PJ_SUCCESS) {
app_perror("...error(10): ioqueue_register error", rc);
status=-25; goto on_error;
}
TRACE_("registering second sockets...");
rc = pj_ioqueue_register_sock( pool, ioque, csock, NULL,
&test_cb, &ckey);
if (rc != PJ_SUCCESS) {
app_perror("...error(11): ioqueue_register error", rc);
status=-26; goto on_error;
}
// Randomize send_buf.
pj_create_random_string(send_buf, bufsize);
// Init operation keys.
pj_ioqueue_op_key_init(&read_op, sizeof(read_op));
pj_ioqueue_op_key_init(&write_op, sizeof(write_op));
// Register reading from ioqueue.
TRACE_("start recvfrom...");
pj_bzero(&addr, sizeof(addr));
addrlen = sizeof(addr);
bytes = bufsize;
rc = pj_ioqueue_recvfrom(skey, &read_op, recv_buf, &bytes, 0,
&addr, &addrlen);
app_perror("...error: pj_ioqueue_recvfrom", rc);
status=-28; goto on_error;
recv_pending = 1;
PJ_LOG(3, (THIS_FILE,
"......ok: recvfrom returned pending"));
} else {
PJ_LOG(3, (THIS_FILE,
"......error: recvfrom returned immediate ok!"));
status=-29; goto on_error;
}
// Set destination address to send the packet.
TRACE_("set destination address...");
temp = pj_str("127.0.0.1");
if ((rc=pj_sockaddr_in_init(&dst_addr, &temp, PORT)) != 0) {
app_perror("...error: unable to resolve 127.0.0.1", rc);
status=-290; goto on_error;
}
// Write must return the number of bytes.
TRACE_("start sendto...");
bytes = bufsize;
rc = pj_ioqueue_sendto(ckey, &write_op, send_buf, &bytes, 0, &dst_addr,
sizeof(dst_addr));
app_perror("...error: pj_ioqueue_sendto", rc);
status=-30; goto on_error;
send_pending = 1;
PJ_LOG(3, (THIS_FILE,
"......ok: sendto returned pending"));
} else {
send_pending = 0;
PJ_LOG(3, (THIS_FILE,
"......ok: sendto returned immediate success"));
}
// reset callback variables.
callback_read_size = callback_write_size = 0;
callback_accept_status = callback_connect_status = -2;
callback_read_key = callback_write_key =
callback_accept_key = callback_connect_key = NULL;
callback_read_op = callback_write_op = NULL;
// Poll if pending.
while (send_pending || recv_pending) {
int ret;
pj_time_val timeout = { 5, 0 };
TRACE_("poll...");
#ifdef PJ_SYMBIAN
ret = pj_symbianos_poll(-1, PJ_TIME_VAL_MSEC(timeout));
#else
ret = pj_ioqueue_poll(ioque, &timeout);
#endif
if (ret == 0) {
PJ_LOG(1,(THIS_FILE, "...ERROR: timed out..."));
status=-45; goto on_error;
} else if (ret < 0) {
app_perror("...ERROR in ioqueue_poll()", -ret);
status=-50; goto on_error;
}
if (callback_read_key != NULL) {
if (callback_read_size != bufsize) {
status=-61; goto on_error;
}
if (callback_read_key != skey) {
status=-65; goto on_error;
}
if (callback_read_op != &read_op) {
status=-66; goto on_error;
}
if (pj_memcmp(send_buf, recv_buf, bufsize) != 0) {
status=-67; goto on_error;
}
status=-68; goto on_error;
}
status=-69; goto on_error;
}
recv_pending = 0;
}
if (callback_write_key != NULL) {
if (callback_write_size != bufsize) {
status=-73; goto on_error;
}
if (callback_write_key != ckey) {
status=-75; goto on_error;
}
if (callback_write_op != &write_op) {
status=-76; goto on_error;
}
send_pending = 0;
}
}
// Success
status = 0;
on_error:
if (skey)
pj_ioqueue_unregister(skey);
else if (ssock != -1)
pj_sock_close(ssock);
if (ckey)
pj_ioqueue_unregister(ckey);
else if (csock != -1)
pj_sock_close(csock);
if (ioque != NULL)
pj_ioqueue_destroy(ioque);
pj_pool_release(pool);
return status;
}
pj_ssize_t bytes_read)
{
PJ_UNUSED_ARG(bytes_read);
(*p_packet_cnt)++;
}
/*
* unregister_test()
* Check if callback is still called after socket has been unregistered or
* closed.
*/
{
enum { RPORT = 50000, SPORT = 50001 };
pj_pool_t *pool;
pj_ioqueue_t *ioqueue;
pj_sock_t ssock;
pj_sock_t rsock, rsock2;
int i, addrlen;
pj_sockaddr_in addr;
pj_ioqueue_key_t *key, *key2;
pj_ioqueue_op_key_t opkey;
unsigned packet_cnt;
char sendbuf[10], recvbuf[10];
void *user_data2 = (void*)(long)2;
pj_ssize_t bytes;
pj_time_val timeout;
pj_status_t status;
pool = pj_pool_create(mem, "test", 4000, 4000, NULL);
if (!pool) {
app_perror("Unable to create pool", PJ_ENOMEM);
return -100;
}
status = pj_ioqueue_create2(pool, 1, cfg, &ioqueue);
if (status != PJ_SUCCESS) {
app_perror("Error creating ioqueue", status);
return -110;
}
/* Create sender socket */
status = app_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, SPORT, &ssock);
if (status != PJ_SUCCESS) {
app_perror("Error initializing socket", status);
return -120;
}
/* Create receiver socket. */
status = app_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, RPORT, &rsock);
if (status != PJ_SUCCESS) {
app_perror("Error initializing socket", status);
return -130;
}
/* Register rsock to ioqueue. */
pj_bzero(&cb, sizeof(cb));
cb.on_read_complete = &on_read_complete;
packet_cnt = 0;
status = pj_ioqueue_register_sock(pool, ioqueue, rsock, &packet_cnt,
&cb, &key);
if (status != PJ_SUCCESS) {
app_perror("Error registering to ioqueue", status);
return -140;
}
/* Init operation key. */
pj_ioqueue_op_key_init(&opkey, sizeof(opkey));
/* Start reading. */
bytes = sizeof(recvbuf);
status = pj_ioqueue_recv( key, &opkey, recvbuf, &bytes, 0);
if (status != PJ_EPENDING) {
app_perror("Expecting PJ_EPENDING, but got this", status);
return -150;
}
/* Init destination address. */
addrlen = sizeof(addr);
status = pj_sock_getsockname(rsock, &addr, &addrlen);
if (status != PJ_SUCCESS) {
app_perror("getsockname error", status);
return -160;
}
/* Override address with 127.0.0.1, since getsockname will return
* zero in the address field.
*/
/* Init buffer to send */
pj_ansi_strcpy(sendbuf, "Hello0123");
/* Send one packet. */
bytes = sizeof(sendbuf);
status = pj_sock_sendto(ssock, sendbuf, &bytes, 0,
&addr, sizeof(addr));
if (status != PJ_SUCCESS) {
app_perror("sendto error", status);
return -170;
}
/* Check if packet is received. */
#ifdef PJ_SYMBIAN
pj_symbianos_poll(-1, 1000);
#else
pj_ioqueue_poll(ioqueue, &timeout);
#endif
if (packet_cnt != 1) {
return -180;
}
/* Just to make sure things are settled.. */
pj_thread_sleep(100);
/* Start reading again. */
bytes = sizeof(recvbuf);
status = pj_ioqueue_recv( key, &opkey, recvbuf, &bytes, 0);
if (status != PJ_EPENDING) {
app_perror("Expecting PJ_EPENDING, but got this", status);
return -190;
}
/* Reset packet counter */
packet_cnt = 0;
/* Send one packet. */
bytes = sizeof(sendbuf);
status = pj_sock_sendto(ssock, sendbuf, &bytes, 0,
&addr, sizeof(addr));
if (status != PJ_SUCCESS) {
app_perror("sendto error", status);
return -200;
}
/* Now unregister and close socket. */
status = pj_ioqueue_unregister(key);
if (status != PJ_SUCCESS) {
app_perror("pj_ioqueue_unregister error", status);
return -201;
}
/* Poll ioqueue. */
for (i=0; i<10; ++i) {
#ifdef PJ_SYMBIAN
pj_symbianos_poll(-1, 100);
#else
pj_ioqueue_poll(ioqueue, &timeout);
#endif
}
/* Must NOT receive any packets after socket is closed! */
if (packet_cnt > 0) {
PJ_LOG(3,(THIS_FILE, "....errror: not expecting to receive packet "
"after socket has been closed"));
return -210;
}
/* Now unregister again, immediately and after PJ_IOQUEUE_KEY_FREE_DELAY.
* It should return error, and most importantly, it must not crash.. */
for (i=0; i<2; ++i) {
status = pj_ioqueue_unregister(key);
/*
* as it turns out, double unregistration returns PJ_SUCCESS
if (status == PJ_SUCCESS) {
PJ_LOG(1, (THIS_FILE,
"Expecting pj_ioqueue_unregister() error (i=%d)", i));
return -220;
}
*/
}
/*
* Second stage of the test. Register another socket. Then unregister using
* the previous key.
*/
status = app_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, RPORT, &rsock2);
if (status != PJ_SUCCESS) {
app_perror("Error initializing socket (2)", status);
return -330;
}
/* Register rsock to ioqueue. */
status = pj_ioqueue_register_sock(pool, ioqueue, rsock2, user_data2,
&cb, &key2);
if (status != PJ_SUCCESS) {
app_perror("Error registering to ioqueue (2)", status);
return -340;
}
/* We shouldn't be able to unregister using the first key. Or should we?
*
* So basically we're simulating buggy application that is unregistering
* an old key.
*
* With current ioqueue implementation, it will return success because
* "key" is the same as "key2" (because ioueue's max_handles is 1 and due to
* PJ_IOQUEUE_HAS_SAFE_UNREG in ioqueue). But what is the expected status
* anyway? Should it return an error? Ideally, I think so. But since we
* can't do that, I'm putting this as an INFO to remind us about this
* behavior.
*/
status = pj_ioqueue_unregister(key);
if (status == PJ_SUCCESS) {
PJ_LOG(3,(THIS_FILE, "....info: unregistering dead key was successful"));
}
/* Success */
pj_sock_close(ssock);
pj_ioqueue_destroy(ioqueue);
pj_pool_release(pool);
return 0;
}
/*
* Testing with many handles.
* This will just test registering PJ_IOQUEUE_MAX_HANDLES count
* of sockets to the ioqueue.
*/
{
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**)
/* Create IOQueue */
rc = pj_ioqueue_create2(pool, MAX, cfg, &ioqueue);
if (rc != PJ_SUCCESS || ioqueue == NULL) {
app_perror("...error in pj_ioqueue_create", rc);
return -10;
}
/* Register as many sockets. */
for (count=0; count<MAX; ++count) {
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) {
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;
}
#if PJ_HAS_THREADS
typedef struct parallel_recv_data
{
unsigned buffer;
pj_ssize_t len;
} parallel_recv_data;
pj_ssize_t bytes_read)
{
parallel_recv_data *ud = (parallel_recv_data*)op_key->user_data;
if (bytes_read < 0) {
pj_status_t status = (pj_status_t)-bytes_read;
if (status==PJ_STATUS_FROM_OS(PJ_BLOCKING_ERROR_VAL)) {
TRACE__((THIS_FILE, "......recv() fail with status=%d, retrying",
status));
ud->len = bytes_read = sizeof(ud->buffer);
status = pj_ioqueue_recv(key, op_key, &ud->buffer, &ud->len, 0);
if (status == PJ_EPENDING)
return;
}
if (status != PJ_SUCCESS) {
PJ_PERROR(3,(THIS_FILE, status, "......status=%d", status));
return;
}
}
assert (bytes_read==sizeof(unsigned));
if (ud->buffer != *p_packet_cnt) {
PJ_LOG(1,(THIS_FILE, "......error: invalid packet sequence "
"(expecting %d, got %d)",
*p_packet_cnt, ud->buffer));
} else {
TRACE__((THIS_FILE, "......recv() sequence %d", ud->buffer));
}
(*p_packet_cnt)++;
}
typedef struct parallel_thread_data
{
pj_ioqueue_t *ioqueue;
pj_bool_t quit_flag;
unsigned id, timeout, wakeup_cnt, event_cnt, err_cnt;
} parallel_thread_data;
static int parallel_worker_thread(void *p)
{
parallel_thread_data *arg = (parallel_thread_data*)p;
pj_time_val t_end;
pj_gettickcount(&t_end);
t_end.sec += arg->timeout;
while (!arg->quit_flag) {
pj_time_val timeout = {0, 0};
int rc;
timeout.sec = arg->timeout;
rc = pj_ioqueue_poll(arg->ioqueue, &timeout);
if (rc >= 1) {
assert(rc==1); /* we should receive packet one by one! */
TRACE__((THIS_FILE, "......thread %d got event (total=%d)",
arg->id, arg->wakeup_cnt));
++arg->wakeup_cnt;
arg->event_cnt += rc;
} else if (rc == 0) {
if (!arg->quit_flag) {
TRACE__((THIS_FILE, "......thread %d wakeup, no event (total=%d)",
arg->id, arg->wakeup_cnt));
++arg->wakeup_cnt;
}
} else if (rc < 0) {
TRACE__((THIS_FILE, "......thread %d got error", arg->id));
++arg->wakeup_cnt;
++arg->err_cnt;
}
}
return 0;
}
/*
* Parallel recv test. Test this scenario:
* - create socket
* - spawn N ioqueue_recv() operations on N threads
* - repeat N times:
* - send one packet to the socket
* - on recv callback, do not re-invoke ioqueue_recv()
* Expected result: we should receive N packets
*/
{
pj_pool_t *pool;
pj_ioqueue_t *ioqueue = NULL;
pj_ioqueue_key_t *skey = NULL;
enum {
ASYNC_CNT = 16,
PKT_SIZE = 16,
SEND_DELAY_MSECS = 250,
TIMEOUT_SECS = (SEND_DELAY_MSECS*ASYNC_CNT/1000)+2,
};
typedef int packet_t;
pj_thread_t *threads[ASYNC_CNT];
parallel_thread_data thread_datas[ASYNC_CNT], threads_total;
pj_ioqueue_op_key_t recv_ops[ASYNC_CNT];
parallel_recv_data recv_datas[ASYNC_CNT];
unsigned i, async_send = 0, recv_packet_count = 0;
int retcode;
pool = pj_pool_create(mem, "test", 4000, 4000, NULL);
if (!pool) {
app_perror("Unable to create pool", PJ_ENOMEM);
return -100;
}
CHECK(-110, app_socketpair(pj_AF_INET(), pj_SOCK_STREAM(), 0,
&ssock, &csock));
CHECK(-120, pj_ioqueue_create2(pool, 2, cfg, &ioqueue));
pj_bzero(&cb, sizeof(cb));
cb.on_read_complete = &on_read_complete2;
CHECK(-130, pj_ioqueue_register_sock(pool, ioqueue, ssock, &recv_packet_count,
&cb, &skey));
/* spawn parallel recv()s */
pj_bzero(recv_datas, sizeof(recv_datas));
for (i=0; i<ASYNC_CNT; ++i) {
recv_ops[i].user_data = &recv_datas[i];
recv_datas[i].len = sizeof(packet_t);
CHECK(-140, pj_ioqueue_recv(skey, &recv_ops[i], &recv_datas[i].buffer,
&recv_datas[i].len, 0));
}
/* spawn polling threads */
pj_bzero(thread_datas, sizeof(thread_datas));
for (i=0; i<ASYNC_CNT; ++i) {
parallel_thread_data *arg = &thread_datas[i];
arg->ioqueue = ioqueue;
arg->id = i;
arg->timeout = TIMEOUT_SECS;
CHECK(-150, pj_thread_create(pool, "parallel_thread",
parallel_worker_thread, arg,
0, 0,&threads[i]));
}
/* now slowly send packet one by one. Let's hope the OS doesn't drop
* our packet, since it's UDP
*/
pj_thread_sleep(100); /* allow thread to start */
for (i=0; i<ASYNC_CNT; ++i) {
packet_t send_buf = i;
pj_ssize_t len = sizeof(send_buf);
pj_status_t status;
pj_thread_sleep((i>0)*SEND_DELAY_MSECS);
TRACE__((THIS_FILE, "....sending"));
status = pj_sock_send(csock, &send_buf, &len, 0);
if (status==PJ_EPENDING) {
++async_send;
TRACE__((THIS_FILE, "......(was async sent)"));
PJ_PERROR(1,(THIS_FILE, status, "......send error"));
retcode = -160;
goto on_return;
}
}
/* Signal threads that it's done */
for (i=0; i<ASYNC_CNT; ++i) {
parallel_thread_data *arg = &thread_datas[i];
arg->quit_flag = 1;
}
/* Wait until all threads quits */
for (i=0; i<ASYNC_CNT; ++i) {
CHECK(-170, pj_thread_join(threads[i]));
CHECK(-180, pj_thread_destroy(threads[i]));
}
/* Display thread statistics */
PJ_LOG(3,(THIS_FILE, "....Threads statistics:"));
PJ_LOG(3,(THIS_FILE, " ============================="));
PJ_LOG(3,(THIS_FILE, " Thread Wakeups Events Errors"));
PJ_LOG(3,(THIS_FILE, " ============================="));
pj_bzero(&threads_total, sizeof(threads_total));
for (i=0; i<ASYNC_CNT; ++i) {
parallel_thread_data *arg = &thread_datas[i];
threads_total.wakeup_cnt += arg->wakeup_cnt;
threads_total.event_cnt += arg->event_cnt;
threads_total.err_cnt += arg->err_cnt;
PJ_LOG(3,(THIS_FILE, " %6d %6d %6d %6d",
arg->id, arg->wakeup_cnt, arg->event_cnt, arg->err_cnt));
}
retcode = 0;
/* Analyze results */
//assert(threads_total.event_cnt == recv_packet_count);
if (recv_packet_count != ASYNC_CNT) {
PJ_LOG(1,(THIS_FILE, "....error: rx packet count is %d (expecting %d)",
recv_packet_count, ASYNC_CNT));
retcode = -500;
}
if (threads_total.wakeup_cnt > ASYNC_CNT+async_send) {
PJ_LOG(3,(THIS_FILE, "....info: total wakeup count is %d "
"(the perfect count is %d). This shows that "
"threads are woken up without getting any events",
threads_total.wakeup_cnt, ASYNC_CNT+async_send));
}
if (threads_total.err_cnt > 0) {
PJ_LOG(3,(THIS_FILE, "....info: total error count is %d "
"(it should be 0)",
threads_total.err_cnt));
}
if (retcode==0)
PJ_LOG(3,(THIS_FILE, "....success"));
on_return:
if (skey)
pj_ioqueue_unregister(skey);
if (csock != PJ_INVALID_SOCKET)
pj_sock_close(csock);
if (ioqueue)
pj_ioqueue_destroy(ioqueue);
pj_pool_release(pool);
return retcode;
}
#endif /* PJ_HAS_THREADS */
/*
* Multi-operation test.
*/
/*
* Benchmarking IOQueue
*/
int inactive_sock_count)
{
pj_sock_t ssock=-1, csock=-1;
pj_sockaddr_in addr;
pj_pool_t *pool = NULL;
pj_sock_t *inactive_sock=NULL;
pj_ioqueue_op_key_t *inactive_read_op;
char *send_buf, *recv_buf;
pj_ioqueue_t *ioque = NULL;
pj_ioqueue_key_t *skey, *ckey, *keys[SOCK_INACTIVE_MAX+2];
pj_timestamp t1, t2, t_elapsed;
int rc=0, i; /* i must be signed */
pj_str_t temp;
char errbuf[PJ_ERR_MSG_SIZE];
TRACE__((THIS_FILE, " bench test %d", inactive_sock_count));
// Create pool.
pool = pj_pool_create(mem, NULL, POOL_SIZE, 4000, NULL);
// Allocate buffers for send and receive.
send_buf = (char*)pj_pool_alloc(pool, bufsize);
recv_buf = (char*)pj_pool_alloc(pool, bufsize);
// Allocate sockets for sending and receiving.
if (rc == PJ_SUCCESS) {
} else
csock = PJ_INVALID_SOCKET;
if (rc != PJ_SUCCESS) {
app_perror("...error: pj_sock_socket()", rc);
goto on_error;
}
// Bind server socket.
pj_bzero(&addr, sizeof(addr));
addr.sin_family = pj_AF_INET();
goto on_error;
pj_assert(inactive_sock_count+2 <= PJ_IOQUEUE_MAX_HANDLES);
// Create I/O Queue.
rc = pj_ioqueue_create2(pool, PJ_IOQUEUE_MAX_HANDLES, cfg, &ioque);
if (rc != PJ_SUCCESS) {
app_perror("...error: pj_ioqueue_create()", rc);
goto on_error;
}
// Allocate inactive sockets, and bind them to some arbitrary address.
// Then register them to the I/O queue, and start a read operation.
inactive_sock = (pj_sock_t*)pj_pool_alloc(pool,
inactive_sock_count*sizeof(pj_sock_t));
inactive_read_op = (pj_ioqueue_op_key_t*)pj_pool_alloc(pool,
inactive_sock_count*sizeof(pj_ioqueue_op_key_t));
pj_bzero(&addr, sizeof(addr));
addr.sin_family = pj_AF_INET();
for (i=0; i<inactive_sock_count; ++i) {
pj_ssize_t bytes;
if (rc != PJ_SUCCESS || inactive_sock[i] < 0) {
app_perror("...error: pj_sock_socket()", rc);
goto on_error;
}
pj_sock_close(inactive_sock[i]);
inactive_sock[i] = PJ_INVALID_SOCKET;
app_perror("...error: pj_sock_bind()", rc);
goto on_error;
}
rc = pj_ioqueue_register_sock(pool, ioque, inactive_sock[i],
NULL, &test_cb, &keys[i]);
if (rc != PJ_SUCCESS) {
pj_sock_close(inactive_sock[i]);
inactive_sock[i] = PJ_INVALID_SOCKET;
app_perror("...error(1): pj_ioqueue_register_sock()", rc);
PJ_LOG(3,(THIS_FILE, "....i=%d", i));
goto on_error;
}
bytes = bufsize;
pj_ioqueue_op_key_init(&inactive_read_op[i],
sizeof(inactive_read_op[i]));
rc = pj_ioqueue_recv(keys[i], &inactive_read_op[i], recv_buf, &bytes, 0);
if (rc != PJ_EPENDING) {
pj_sock_close(inactive_sock[i]);
inactive_sock[i] = PJ_INVALID_SOCKET;
app_perror("...error: pj_ioqueue_read()", rc);
goto on_error;
}
}
// Register server and client socket.
// We put this after inactivity socket, hopefully this can represent the
// worst waiting time.
rc = pj_ioqueue_register_sock(pool, ioque, ssock, NULL,
&test_cb, &skey);
if (rc != PJ_SUCCESS) {
app_perror("...error(2): pj_ioqueue_register_sock()", rc);
goto on_error;
}
rc = pj_ioqueue_register_sock(pool, ioque, csock, NULL,
&test_cb, &ckey);
if (rc != PJ_SUCCESS) {
app_perror("...error(3): pj_ioqueue_register_sock()", rc);
goto on_error;
}
// Set destination address to send the packet.
// Test loop.
t_elapsed.u64 = 0;
for (i=0; i<LOOP; ++i) {
pj_ssize_t bytes;
pj_ioqueue_op_key_t read_op, write_op;
// Randomize send buffer.
pj_create_random_string(send_buf, bufsize);
// Init operation keys.
pj_ioqueue_op_key_init(&read_op, sizeof(read_op));
pj_ioqueue_op_key_init(&write_op, sizeof(write_op));
// Start reading on the server side.
bytes = bufsize;
rc = pj_ioqueue_recv(skey, &read_op, recv_buf, &bytes, 0);
if (rc != PJ_EPENDING) {
app_perror("...error: pj_ioqueue_read()", rc);
break;
}
// Starts send on the client side.
bytes = bufsize;
rc = pj_ioqueue_sendto(ckey, &write_op, send_buf, &bytes, 0,
&addr, sizeof(addr));
app_perror("...error: pj_ioqueue_write()", rc);
break;
}
if (rc == PJ_SUCCESS) {
if (bytes < 0) {
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;
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",
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;
}
{
int status;
int bufsize, sock_count;
char title[64];
pj_ansi_snprintf(title, sizeof(title), "%s (concur:%d, epoll_flags:0x%x)",
pj_ioqueue_name(), cfg->default_concurrency,
cfg->epoll_flags);
//goto pass1;
PJ_LOG(3, (THIS_FILE, "...compliance test (%s)", title));
if ((status=compliance_test(cfg)) != 0) {
return status;
}
PJ_LOG(3, (THIS_FILE, "....compliance test ok"));
PJ_LOG(3, (THIS_FILE, "...unregister test (%s)", title));
if ((status=unregister_test(cfg)) != 0) {
return status;
}
PJ_LOG(3, (THIS_FILE, "....unregister test ok"));
if ((status=many_handles_test(cfg)) != 0) {
return status;
}
//return 0;
PJ_LOG(4, (THIS_FILE, "...benchmarking different buffer size:"));
PJ_LOG(4, (THIS_FILE, "... note: buf=bytes sent, fds=# of fds, "
"elapsed=in timer ticks"));
//pass1:
PJ_LOG(3, (THIS_FILE, "...Benchmarking poll times for %s:", title));
PJ_LOG(3, (THIS_FILE, "...====================================="));
PJ_LOG(3, (THIS_FILE, "...Buf.size #inactive-socks Time/poll"));
PJ_LOG(3, (THIS_FILE, "... (bytes) (nanosec)"));
PJ_LOG(3, (THIS_FILE, "...====================================="));
//goto pass2;
for (bufsize=BUF_MIN_SIZE; bufsize <= BUF_MAX_SIZE; bufsize *= 2) {
if ((status=bench_test(cfg, bufsize, SOCK_INACTIVE_MIN)) != 0)
return status;
}
//pass2:
bufsize = 512;
for (sock_count=SOCK_INACTIVE_MIN+2;
sock_count<=SOCK_INACTIVE_MAX+2;
sock_count *= 2)
{
//PJ_LOG(3,(THIS_FILE, "...testing with %d fds", sock_count));
if ((status=bench_test(cfg, bufsize, sock_count-2)) != 0)
return status;
}
return 0;
}
int udp_ioqueue_test()
{
pj_ioqueue_epoll_flag epoll_flags[] = {
#if PJ_HAS_LINUX_EPOLL
0,
#else
#endif
};
int i, rc, err = 0;
for (i=0; i<PJ_ARRAY_SIZE(epoll_flags); ++i) {
pj_ioqueue_cfg cfg;
pj_ioqueue_cfg_default(&cfg);
cfg.epoll_flags = epoll_flags[i];
PJ_LOG(3, (THIS_FILE, "..%s UDP compliance test, epoll_flags=0x%x",
pj_ioqueue_name(), cfg.epoll_flags));
rc = udp_ioqueue_test_imp(&cfg);
if (rc != 0 && err==0)
err = rc;
}
for (i=0; i<PJ_ARRAY_SIZE(concurs); ++i) {
pj_ioqueue_cfg cfg;
pj_ioqueue_cfg_default(&cfg);
cfg.default_concurrency = concurs[i];
PJ_LOG(3, (THIS_FILE, "..%s UDP compliance test, concurrency=%d",
pj_ioqueue_name(), cfg.default_concurrency));
rc = udp_ioqueue_test_imp(&cfg);
if (rc != 0 && err==0)
err = rc;
}
#if PJ_HAS_THREADS
for (i=0; i<PJ_ARRAY_SIZE(epoll_flags); ++i) {
pj_ioqueue_cfg cfg;
pj_ioqueue_cfg_default(&cfg);
cfg.epoll_flags = epoll_flags[i];
PJ_LOG(3, (THIS_FILE, "..%s UDP parallel compliance test, epoll_flags=0x%x",
pj_ioqueue_name(), cfg.epoll_flags));
rc = parallel_recv_test(&cfg);
if (rc != 0 && err==0)
err = rc;
}
#endif
return err;
}
#else
/* To prevent warning about "translation unit is empty"
* when this test is disabled.
*/
int dummy_uiq_udp;
#endif /* INCLUDE_UDP_IOQUEUE_TEST */
void pj_ioqueue_op_key_init(pj_ioqueue_op_key_t *op_key, pj_size_t size) pj_status_t pj_ioqueue_recvfrom(pj_ioqueue_key_t *key, pj_ioqueue_op_key_t *op_key, void *buffer, pj_ssize_t *length, pj_uint32_t flags, pj_sockaddr_t *addr, int *addrlen) pj_status_t pj_ioqueue_create2(pj_pool_t *pool, pj_size_t max_fd, const pj_ioqueue_cfg *cfg, pj_ioqueue_t **ioqueue) pj_status_t pj_ioqueue_destroy(pj_ioqueue_t *ioque) pj_status_t pj_ioqueue_recv(pj_ioqueue_key_t *key, pj_ioqueue_op_key_t *op_key, void *buffer, pj_ssize_t *length, pj_uint32_t flags) void * pj_ioqueue_get_user_data(pj_ioqueue_key_t *key) pj_status_t pj_ioqueue_unregister(pj_ioqueue_key_t *key) void pj_ioqueue_cfg_default(pj_ioqueue_cfg *cfg) int pj_ioqueue_poll(pj_ioqueue_t *ioque, const pj_time_val *timeout) pj_status_t pj_ioqueue_sendto(pj_ioqueue_key_t *key, pj_ioqueue_op_key_t *op_key, const void *data, pj_ssize_t *length, pj_uint32_t flags, const pj_sockaddr_t *addr, int addrlen) 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 **key) const char * pj_ioqueue_name(void) void * pj_pool_alloc(pj_pool_t *pool, pj_size_t size) pj_pool_t * pj_pool_create(pj_pool_factory *factory, const char *name, pj_size_t initial_size, pj_size_t increment_size, pj_pool_callback *callback) void pj_pool_release(pj_pool_t *pool) pj_str_t pj_str(char *str) char * pj_create_random_string(char *str, pj_size_t length) pj_status_t pj_sock_bind(pj_sock_t sockfd, const pj_sockaddr_t *my_addr, int addrlen) pj_status_t pj_sock_send(pj_sock_t sockfd, const void *buf, pj_ssize_t *len, unsigned flags) pj_uint16_t pj_htons(pj_uint16_t hostshort) pj_status_t pj_sock_close(pj_sock_t sockfd) pj_in_addr pj_inet_addr2(const char *cp) pj_status_t pj_sock_sendto(pj_sock_t sockfd, const void *buf, pj_ssize_t *len, unsigned flags, const pj_sockaddr_t *to, int tolen) pj_status_t pj_sock_getsockname(pj_sock_t sockfd, pj_sockaddr_t *addr, int *namelen) pj_status_t pj_sockaddr_in_init(pj_sockaddr_in *addr, const pj_str_t *cp, pj_uint16_t port) pj_status_t pj_sock_socket(int family, int type, int protocol, pj_sock_t *sock) pj_bool_t pj_symbianos_poll(int priority, int ms_timeout) pj_status_t pj_thread_destroy(pj_thread_t *thread) pj_status_t pj_thread_join(pj_thread_t *thread) pj_status_t pj_thread_create(pj_pool_t *pool, const char *thread_name, pj_thread_proc *proc, void *arg, pj_size_t stack_size, unsigned flags, pj_thread_t **thread) pj_status_t pj_thread_sleep(unsigned msec) pj_status_t pj_gettickcount(pj_time_val *tv) pj_status_t pj_get_timestamp(pj_timestamp *ts) pj_uint32_t pj_elapsed_usec(const pj_timestamp *start, const pj_timestamp *stop) pj_str_t pj_strerror(pj_status_t statcode, char *buf, pj_size_t bufsize) void pj_set_os_error(pj_status_t code) pj_status_t pj_get_netos_error(void) Definition: udp_echo_srv_ioqueue.c:27 Definition: ioqueue.h:219 void(* on_read_complete)(pj_ioqueue_key_t *key, pj_ioqueue_op_key_t *op_key, pj_ssize_t bytes_read) Definition: ioqueue.h:231 Definition: ioqueue.h:362 Definition: ioqueue.h:208 Definition: pool.h:310 Definition: sock.h:534 Definition: types.h:120 Definition: types.h:397 Definition: types.h:134 struct pj_timestamp::@9 u32
Copyright (C) 2006-2009 Teluu Inc. |