1
/* $Id: ioq_unreg.c 3553 2011-05-05 06:14:19Z nanang $ */
3
* Copyright (C) 2008-2011 Teluu Inc. (http://www.teluu.com)
4
* Copyright (C) 2003-2008 Benny Prijono <benny@prijono.org>
6
* This program is free software; you can redistribute it and/or modify
7
* it under the terms of the GNU General Public License as published by
8
* the Free Software Foundation; either version 2 of the License, or
9
* (at your option) any later version.
11
* This program is distributed in the hope that it will be useful,
12
* but WITHOUT ANY WARRANTY; without even the implied warranty of
13
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14
* GNU General Public License for more details.
16
* You should have received a copy of the GNU General Public License
17
* along with this program; if not, write to the Free Software
18
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
22
#if INCLUDE_IOQUEUE_UNREG_TEST
24
* This tests the thread safety of ioqueue unregistration operation.
28
#include <pj/ioqueue.h>
33
#include <pj/compat/socket.h>
34
#include <pj/string.h>
37
#define THIS_FILE "ioq_unreg.c"
43
UNREGISTER_IN_CALLBACK,
46
static int thread_quitting;
47
static enum test_method test_method;
48
static pj_time_val time_to_unregister;
55
pj_ioqueue_key_t *key;
57
pj_ioqueue_op_key_t *op_key;
60
pj_bool_t unregistered;
64
static void on_read_complete(pj_ioqueue_key_t *key,
65
pj_ioqueue_op_key_t *op_key,
66
pj_ssize_t bytes_read)
69
char *sendbuf = "Hello world";
72
if (sock_data.unregistered)
75
pj_mutex_lock(sock_data.mutex);
77
if (sock_data.unregistered) {
78
pj_mutex_unlock(sock_data.mutex);
83
if (-bytes_read != PJ_STATUS_FROM_OS(PJ_BLOCKING_ERROR_VAL))
84
app_perror("ioqueue reported recv error", -bytes_read);
86
sock_data.received += bytes_read;
89
if (test_method == UNREGISTER_IN_CALLBACK) {
92
pj_gettimeofday(&now);
93
if (PJ_TIME_VAL_GTE(now, time_to_unregister)) {
94
sock_data.unregistered = 1;
95
pj_ioqueue_unregister(key);
96
pj_mutex_unlock(sock_data.mutex);
102
size = sock_data.bufsize;
103
status = pj_ioqueue_recv(key, op_key, sock_data.buffer, &size, 0);
104
if (status != PJ_EPENDING && status != PJ_SUCCESS)
105
app_perror("recv() error", status);
107
} while (status == PJ_SUCCESS);
109
pj_mutex_unlock(sock_data.mutex);
111
size = pj_ansi_strlen(sendbuf);
112
status = pj_sock_send(sock_data.csock, sendbuf, &size, 0);
113
if (status != PJ_SUCCESS)
114
app_perror("send() error", status);
116
size = pj_ansi_strlen(sendbuf);
117
status = pj_sock_send(sock_data.csock, sendbuf, &size, 0);
118
if (status != PJ_SUCCESS)
119
app_perror("send() error", status);
123
static int worker_thread(void *arg)
125
pj_ioqueue_t *ioqueue = (pj_ioqueue_t*) arg;
127
while (!thread_quitting) {
128
pj_time_val timeout = { 0, 20 };
129
pj_ioqueue_poll(ioqueue, &timeout);
136
* Perform unregistration test.
138
* This will create ioqueue and register a server socket. Depending
139
* on the test method, either the callback or the main thread will
140
* unregister and destroy the server socket after some period of time.
142
static int perform_unreg_test(pj_ioqueue_t *ioqueue,
143
pj_pool_t *test_pool,
145
pj_bool_t other_socket)
147
enum { WORKER_CNT = 1, MSEC = 500, QUIT_MSEC = 500 };
149
pj_thread_t *thread[WORKER_CNT];
150
struct sock_data osd;
151
pj_ioqueue_callback callback;
152
pj_time_val end_time;
156
/* Sometimes its important to have other sockets registered to
157
* the ioqueue, because when no sockets are registered, the ioqueue
158
* will return from the poll early.
161
status = app_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, 56127, &osd.sock);
162
if (status != PJ_SUCCESS) {
163
app_perror("Error creating other socket", status);
167
pj_bzero(&callback, sizeof(callback));
168
status = pj_ioqueue_register_sock(test_pool, ioqueue, osd.sock,
169
NULL, &callback, &osd.key);
170
if (status != PJ_SUCCESS) {
171
app_perror("Error registering other socket", status);
177
osd.sock = PJ_INVALID_SOCKET;
180
/* Init both time duration of testing */
182
pj_gettimeofday(&time_to_unregister);
183
time_to_unregister.msec += MSEC;
184
pj_time_val_normalize(&time_to_unregister);
186
end_time = time_to_unregister;
187
end_time.msec += QUIT_MSEC;
188
pj_time_val_normalize(&end_time);
191
/* Create polling thread */
192
for (i=0; i<WORKER_CNT; ++i) {
193
status = pj_thread_create(test_pool, "unregtest", &worker_thread,
194
ioqueue, 0, 0, &thread[i]);
195
if (status != PJ_SUCCESS) {
196
app_perror("Error creating thread", status);
201
/* Create pair of client/server sockets */
202
status = app_socketpair(pj_AF_INET(), pj_SOCK_DGRAM(), 0,
203
&sock_data.sock, &sock_data.csock);
204
if (status != PJ_SUCCESS) {
205
app_perror("app_socketpair error", status);
210
/* Initialize test data */
211
sock_data.pool = pj_pool_create(mem, "sd", 1000, 1000, NULL);
212
sock_data.buffer = (char*) pj_pool_alloc(sock_data.pool, 128);
213
sock_data.bufsize = 128;
214
sock_data.op_key = (pj_ioqueue_op_key_t*)
215
pj_pool_alloc(sock_data.pool,
216
sizeof(*sock_data.op_key));
217
sock_data.received = 0;
218
sock_data.unregistered = 0;
220
pj_ioqueue_op_key_init(sock_data.op_key, sizeof(*sock_data.op_key));
222
status = pj_mutex_create_simple(sock_data.pool, "sd", &sock_data.mutex);
223
if (status != PJ_SUCCESS) {
224
app_perror("create_mutex() error", status);
228
/* Register socket to ioqueue */
229
pj_bzero(&callback, sizeof(callback));
230
callback.on_read_complete = &on_read_complete;
231
status = pj_ioqueue_register_sock(sock_data.pool, ioqueue, sock_data.sock,
232
NULL, &callback, &sock_data.key);
233
if (status != PJ_SUCCESS) {
234
app_perror("pj_ioqueue_register error", status);
238
/* Bootstrap the first send/receive */
239
on_read_complete(sock_data.key, sock_data.op_key, 0);
241
/* Loop until test time ends */
243
pj_time_val now, timeout;
246
pj_gettimeofday(&now);
248
if (test_method == UNREGISTER_IN_APP &&
249
PJ_TIME_VAL_GTE(now, time_to_unregister) &&
250
!sock_data.unregistered)
252
sock_data.unregistered = 1;
253
/* Wait (as much as possible) for callback to complete */
254
pj_mutex_lock(sock_data.mutex);
255
pj_mutex_unlock(sock_data.mutex);
256
pj_ioqueue_unregister(sock_data.key);
259
if (PJ_TIME_VAL_GT(now, end_time) && sock_data.unregistered)
262
timeout.sec = 0; timeout.msec = 10;
263
n = pj_ioqueue_poll(ioqueue, &timeout);
265
app_perror("pj_ioqueue_poll error", -n);
272
for (i=0; i<WORKER_CNT; ++i) {
273
pj_thread_join(thread[i]);
274
pj_thread_destroy(thread[i]);
278
pj_mutex_destroy(sock_data.mutex);
279
pj_pool_release(sock_data.pool);
280
sock_data.pool = NULL;
283
pj_ioqueue_unregister(osd.key);
286
pj_sock_close(sock_data.csock);
288
PJ_LOG(3,(THIS_FILE, "....%s: done (%d KB/s)",
289
title, sock_data.received * 1000 / MSEC / 1000));
293
static int udp_ioqueue_unreg_test_imp(pj_bool_t allow_concur)
298
pj_ioqueue_t *ioqueue;
299
pj_pool_t *test_pool;
301
PJ_LOG(3,(THIS_FILE, "..testing with concurency=%d", allow_concur));
303
test_method = UNREGISTER_IN_APP;
305
test_pool = pj_pool_create(mem, "unregtest", 4000, 4000, NULL);
307
rc = pj_ioqueue_create(test_pool, 16, &ioqueue);
308
if (rc != PJ_SUCCESS) {
309
app_perror("Error creating ioqueue", rc);
313
rc = pj_ioqueue_set_default_concurrency(ioqueue, allow_concur);
314
if (rc != PJ_SUCCESS) {
315
app_perror("Error in pj_ioqueue_set_default_concurrency()", rc);
319
PJ_LOG(3, (THIS_FILE, "...ioqueue unregister stress test 0/3, unregister in app (%s)",
321
for (i=0; i<LOOP; ++i) {
322
pj_ansi_sprintf(title, "repeat %d/%d", i, LOOP);
323
rc = perform_unreg_test(ioqueue, test_pool, title, 0);
329
PJ_LOG(3, (THIS_FILE, "...ioqueue unregister stress test 1/3, unregister in app (%s)",
331
for (i=0; i<LOOP; ++i) {
332
pj_ansi_sprintf(title, "repeat %d/%d", i, LOOP);
333
rc = perform_unreg_test(ioqueue, test_pool, title, 1);
338
test_method = UNREGISTER_IN_CALLBACK;
340
PJ_LOG(3, (THIS_FILE, "...ioqueue unregister stress test 2/3, unregister in cb (%s)",
342
for (i=0; i<LOOP; ++i) {
343
pj_ansi_sprintf(title, "repeat %d/%d", i, LOOP);
344
rc = perform_unreg_test(ioqueue, test_pool, title, 0);
350
PJ_LOG(3, (THIS_FILE, "...ioqueue unregister stress test 3/3, unregister in cb (%s)",
352
for (i=0; i<LOOP; ++i) {
353
pj_ansi_sprintf(title, "repeat %d/%d", i, LOOP);
354
rc = perform_unreg_test(ioqueue, test_pool, title, 1);
359
pj_ioqueue_destroy(ioqueue);
360
pj_pool_release(test_pool);
365
int udp_ioqueue_unreg_test(void)
369
rc = udp_ioqueue_unreg_test_imp(PJ_TRUE);
373
rc = udp_ioqueue_unreg_test_imp(PJ_FALSE);
381
/* To prevent warning about "translation unit is empty"
382
* when this test is disabled.
385
#endif /* INCLUDE_IOQUEUE_UNREG_TEST */