diff options
author | Benny Prijono <bennylp@teluu.com> | 2006-03-30 16:32:18 +0000 |
---|---|---|
committer | Benny Prijono <bennylp@teluu.com> | 2006-03-30 16:32:18 +0000 |
commit | 974fbe67d6d62efadd129cc81b9072faf3b2f029 (patch) | |
tree | 82a44cd7c10d447766280047e035928166833348 /pjlib/src/pjlib-test | |
parent | 3cf609b42e573adf8e7183070176a450a7b4959e (diff) |
Fixed race condition bug in ioqueue unregistration for select and Win32 IOCP backend
git-svn-id: http://svn.pjsip.org/repos/pjproject/trunk@365 74dad513-b988-da41-8d7b-12977e46ad98
Diffstat (limited to 'pjlib/src/pjlib-test')
-rw-r--r-- | pjlib/src/pjlib-test/ioq_perf.c | 2 | ||||
-rw-r--r-- | pjlib/src/pjlib-test/ioq_udp.c | 5 | ||||
-rw-r--r-- | pjlib/src/pjlib-test/ioq_unreg.c | 361 | ||||
-rw-r--r-- | pjlib/src/pjlib-test/test.c | 4 | ||||
-rw-r--r-- | pjlib/src/pjlib-test/test.h | 2 |
5 files changed, 367 insertions, 7 deletions
diff --git a/pjlib/src/pjlib-test/ioq_perf.c b/pjlib/src/pjlib-test/ioq_perf.c index bf0e6273..fea9c184 100644 --- a/pjlib/src/pjlib-test/ioq_perf.c +++ b/pjlib/src/pjlib-test/ioq_perf.c @@ -398,8 +398,6 @@ static int perform_test(int sock_type, const char *type_name, for (i=0; i<sockpair_cnt; ++i) { pj_ioqueue_unregister(items[i].server_key); pj_ioqueue_unregister(items[i].client_key); - pj_sock_close(items[i].server_fd); - pj_sock_close(items[i].client_fd); } /* Destroy threads */ diff --git a/pjlib/src/pjlib-test/ioq_udp.c b/pjlib/src/pjlib-test/ioq_udp.c index b81764f2..89e2a1e8 100644 --- a/pjlib/src/pjlib-test/ioq_udp.c +++ b/pjlib/src/pjlib-test/ioq_udp.c @@ -457,7 +457,6 @@ static int unregister_test(void) /* Now unregister and close socket. */ pj_ioqueue_unregister(key); - pj_sock_close(rsock); /* Poll ioqueue. */ timeout.sec = 1; timeout.msec = 0; @@ -539,10 +538,6 @@ static int many_handles_test(void) if (rc != PJ_SUCCESS) { app_perror("...error in pj_ioqueue_unregister", rc); } - rc = pj_sock_close(sock[i]); - if (rc != PJ_SUCCESS) { - app_perror("...error in pj_sock_close", rc); - } } rc = pj_ioqueue_destroy(ioqueue); diff --git a/pjlib/src/pjlib-test/ioq_unreg.c b/pjlib/src/pjlib-test/ioq_unreg.c new file mode 100644 index 00000000..475d62fe --- /dev/null +++ b/pjlib/src/pjlib-test/ioq_unreg.c @@ -0,0 +1,361 @@ +/* $Id$ */ +/* + * Copyright (C)2003-2006 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_IOQUEUE_UNREG_TEST +/* + * This tests the thread safety of ioqueue unregistration operation. + */ + +#include <pj/errno.h> +#include <pj/ioqueue.h> +#include <pj/log.h> +#include <pj/os.h> +#include <pj/pool.h> +#include <pj/sock.h> +#include <pj/compat/socket.h> +#include <pj/string.h> + + +#define THIS_FILE "ioq_unreg.c" + + +enum test_method +{ + UNREGISTER_IN_APP, + UNREGISTER_IN_CALLBACK, +}; + +static int thread_quitting; +static enum test_method test_method; +static pj_time_val time_to_unregister; + +struct sock_data +{ + pj_sock_t sock; + pj_sock_t csock; + pj_pool_t *pool; + pj_ioqueue_key_t *key; + pj_mutex_t *mutex; + pj_ioqueue_op_key_t *op_key; + char *buffer; + pj_size_t bufsize; + pj_bool_t unregistered; + unsigned received; +} sock_data; + +static void on_read_complete(pj_ioqueue_key_t *key, + pj_ioqueue_op_key_t *op_key, + pj_ssize_t bytes_read) +{ + pj_ssize_t size; + char *sendbuf = "Hello world"; + pj_status_t status; + + if (sock_data.unregistered) + return; + + pj_mutex_lock(sock_data.mutex); + + if (sock_data.unregistered) { + /* No need to unlock. Mutex may have been destroyed */ + return; + } + + if (bytes_read < 0) { + if (-bytes_read != PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK)) + app_perror("ioqueue reported recv error", -bytes_read); + } else { + sock_data.received += bytes_read; + } + + if (test_method == UNREGISTER_IN_CALLBACK) { + pj_time_val now; + + pj_gettimeofday(&now); + if (PJ_TIME_VAL_GTE(now, time_to_unregister)) { + sock_data.unregistered = 1; + pj_ioqueue_unregister(key); + pj_mutex_destroy(sock_data.mutex); + pj_pool_release(sock_data.pool); + sock_data.pool = NULL; + return; + } + } + + do { + size = sock_data.bufsize; + status = pj_ioqueue_recv(key, op_key, sock_data.buffer, &size, 0); + if (status != PJ_EPENDING && status != PJ_SUCCESS) + app_perror("recv() error", status); + + } while (status == PJ_SUCCESS); + + pj_mutex_unlock(sock_data.mutex); + + size = pj_ansi_strlen(sendbuf); + status = pj_sock_send(sock_data.csock, sendbuf, &size, 0); + if (status != PJ_SUCCESS) + app_perror("send() error", status); + + size = pj_ansi_strlen(sendbuf); + status = pj_sock_send(sock_data.csock, sendbuf, &size, 0); + if (status != PJ_SUCCESS) + app_perror("send() error", status); + +} + +static int worker_thread(void *arg) +{ + pj_ioqueue_t *ioqueue = arg; + + while (!thread_quitting) { + pj_time_val timeout = { 0, 20 }; + pj_ioqueue_poll(ioqueue, &timeout); + } + + return 0; +} + +/* + * Perform unregistration test. + * + * This will create ioqueue and register a server socket. Depending + * on the test method, either the callback or the main thread will + * unregister and destroy the server socket after some period of time. + */ +static int perform_unreg_test(pj_ioqueue_t *ioqueue, + pj_pool_t *test_pool, + const char *title, + pj_bool_t other_socket) +{ + enum { WORKER_CNT = 1, MSEC = 500, QUIT_MSEC = 500 }; + int i; + pj_thread_t *thread[WORKER_CNT]; + struct sock_data osd; + pj_ioqueue_callback callback; + pj_time_val end_time; + pj_status_t status; + + + /* Sometimes its important to have other sockets registered to + * the ioqueue, because when no sockets are registered, the ioqueue + * will return from the poll early. + */ + if (other_socket) { + status = app_socket(PJ_AF_INET, PJ_SOCK_DGRAM, 0, 56127, &osd.sock); + if (status != PJ_SUCCESS) { + app_perror("Error creating other socket", status); + return -12; + } + + pj_memset(&callback, 0, sizeof(callback)); + status = pj_ioqueue_register_sock(test_pool, ioqueue, osd.sock, + NULL, &callback, &osd.key); + if (status != PJ_SUCCESS) { + app_perror("Error registering other socket", status); + return -13; + } + + } else { + osd.key = NULL; + osd.sock = PJ_INVALID_SOCKET; + } + + /* Init both time duration of testing */ + thread_quitting = 0; + pj_gettimeofday(&time_to_unregister); + time_to_unregister.msec += MSEC; + pj_time_val_normalize(&time_to_unregister); + + end_time = time_to_unregister; + end_time.msec += QUIT_MSEC; + pj_time_val_normalize(&end_time); + + + /* Create polling thread */ + for (i=0; i<WORKER_CNT; ++i) { + status = pj_thread_create(test_pool, "unregtest", &worker_thread, + ioqueue, 0, 0, &thread[i]); + if (status != PJ_SUCCESS) { + app_perror("Error creating thread", status); + return -20; + } + } + + /* Create pair of client/server sockets */ + status = app_socketpair(PJ_AF_INET, PJ_SOCK_DGRAM, 0, + &sock_data.sock, &sock_data.csock); + if (status != PJ_SUCCESS) { + app_perror("app_socketpair error", status); + return -30; + } + + + /* Initialize test data */ + sock_data.pool = pj_pool_create(mem, "sd", 1000, 1000, NULL); + sock_data.buffer = pj_pool_alloc(sock_data.pool, 128); + sock_data.bufsize = 128; + sock_data.op_key = pj_pool_alloc(sock_data.pool, + sizeof(*sock_data.op_key)); + sock_data.received = 0; + sock_data.unregistered = 0; + + pj_ioqueue_op_key_init(sock_data.op_key, sizeof(*sock_data.op_key)); + + status = pj_mutex_create_simple(sock_data.pool, "sd", &sock_data.mutex); + if (status != PJ_SUCCESS) { + app_perror("create_mutex() error", status); + return -35; + } + + /* Register socket to ioqueue */ + pj_memset(&callback, 0, sizeof(callback)); + callback.on_read_complete = &on_read_complete; + status = pj_ioqueue_register_sock(sock_data.pool, ioqueue, sock_data.sock, + NULL, &callback, &sock_data.key); + if (status != PJ_SUCCESS) { + app_perror("pj_ioqueue_register error", status); + return -40; + } + + /* Bootstrap the first send/receive */ + on_read_complete(sock_data.key, sock_data.op_key, 0); + + /* Loop until test time ends */ + for (;;) { + pj_time_val now, timeout; + + pj_gettimeofday(&now); + + if (test_method == UNREGISTER_IN_APP && + PJ_TIME_VAL_GTE(now, time_to_unregister) && + sock_data.pool) + { + pj_mutex_lock(sock_data.mutex); + + sock_data.unregistered = 1; + pj_ioqueue_unregister(sock_data.key); + pj_mutex_destroy(sock_data.mutex); + pj_pool_release(sock_data.pool); + sock_data.pool = NULL; + } + + if (PJ_TIME_VAL_GT(now, end_time) && sock_data.unregistered) + break; + + timeout.sec = 0; timeout.msec = 10; + pj_ioqueue_poll(ioqueue, &timeout); + //pj_thread_sleep(1); + + } + + thread_quitting = 1; + + for (i=0; i<WORKER_CNT; ++i) { + pj_thread_join(thread[i]); + pj_thread_destroy(thread[i]); + } + + if (other_socket) { + pj_ioqueue_unregister(osd.key); + } + + pj_sock_close(sock_data.csock); + + PJ_LOG(3,(THIS_FILE, "....%s: done (%d KB/s)", + title, sock_data.received * 1000 / MSEC / 1000)); + return 0; +} + +int udp_ioqueue_unreg_test(void) +{ + enum { LOOP = 10 }; + int i, rc; + char title[30]; + pj_ioqueue_t *ioqueue; + pj_pool_t *test_pool; + + test_method = UNREGISTER_IN_APP; + + test_pool = pj_pool_create(mem, "unregtest", 4000, 4000, NULL); + + rc = pj_ioqueue_create(test_pool, 16, &ioqueue); + if (rc != PJ_SUCCESS) { + app_perror("Error creating ioqueue", rc); + return -10; + } + + + PJ_LOG(3, (THIS_FILE, "...ioqueue unregister stress test 0/3 (%s)", + pj_ioqueue_name())); + for (i=0; i<LOOP; ++i) { + pj_ansi_sprintf(title, "repeat %d/%d", i, LOOP); + rc = perform_unreg_test(ioqueue, test_pool, title, 0); + if (rc != 0) + return rc; + } + + + PJ_LOG(3, (THIS_FILE, "...ioqueue unregister stress test 1/3 (%s)", + pj_ioqueue_name())); + for (i=0; i<LOOP; ++i) { + pj_ansi_sprintf(title, "repeat %d/%d", i, LOOP); + rc = perform_unreg_test(ioqueue, test_pool, title, 1); + if (rc != 0) + return rc; + } + + test_method = UNREGISTER_IN_CALLBACK; + + PJ_LOG(3, (THIS_FILE, "...ioqueue unregister stress test 2/3 (%s)", + pj_ioqueue_name())); + for (i=0; i<LOOP; ++i) { + pj_ansi_sprintf(title, "repeat %d/%d", i, LOOP); + rc = perform_unreg_test(ioqueue, test_pool, title, 0); + if (rc != 0) + return rc; + } + + + PJ_LOG(3, (THIS_FILE, "...ioqueue unregister stress test 3/3 (%s)", + pj_ioqueue_name())); + for (i=0; i<LOOP; ++i) { + pj_ansi_sprintf(title, "repeat %d/%d", i, LOOP); + rc = perform_unreg_test(ioqueue, test_pool, title, 1); + if (rc != 0) + return rc; + } + + pj_ioqueue_destroy(ioqueue); + pj_pool_release(test_pool); + + return 0; +} + + + +#else +/* To prevent warning about "translation unit is empty" + * when this test is disabled. + */ +int dummy_uiq_unreg; +#endif /* INCLUDE_IOQUEUE_UNREG_TEST */ + + diff --git a/pjlib/src/pjlib-test/test.c b/pjlib/src/pjlib-test/test.c index 1d504ae5..97ba2991 100644 --- a/pjlib/src/pjlib-test/test.c +++ b/pjlib/src/pjlib-test/test.c @@ -145,6 +145,10 @@ int test_inner(void) DO_TEST( ioqueue_perf_test() ); #endif +#if INCLUDE_IOQUEUE_UNREG_TEST + DO_TEST( udp_ioqueue_unreg_test() ); +#endif + #if INCLUDE_FILE_TEST DO_TEST( file_test() ); #endif diff --git a/pjlib/src/pjlib-test/test.h b/pjlib/src/pjlib-test/test.h index 8571d9dd..7a0aeed2 100644 --- a/pjlib/src/pjlib-test/test.h +++ b/pjlib/src/pjlib-test/test.h @@ -48,6 +48,7 @@ #define INCLUDE_UDP_IOQUEUE_TEST GROUP_NETWORK #define INCLUDE_TCP_IOQUEUE_TEST GROUP_NETWORK #define INCLUDE_IOQUEUE_PERF_TEST GROUP_NETWORK +#define INCLUDE_IOQUEUE_UNREG_TEST GROUP_NETWORK #define INCLUDE_FILE_TEST GROUP_FILE #define INCLUDE_ECHO_SERVER 0 @@ -82,6 +83,7 @@ extern int sock_test(void); extern int sock_perf_test(void); extern int select_test(void); extern int udp_ioqueue_test(void); +extern int udp_ioqueue_unreg_test(void); extern int tcp_ioqueue_test(void); extern int ioqueue_perf_test(void); extern int file_test(void); |