/* $Id$ */
/* 
 * 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_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;
    pj_ssize_t		 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) {
	pj_mutex_unlock(sock_data.mutex);
	return;
    }

    if (bytes_read < 0) {
	if (-bytes_read != PJ_STATUS_FROM_OS(PJ_BLOCKING_ERROR_VAL))
	    app_perror("ioqueue reported recv error", (pj_status_t)-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_unlock(sock_data.mutex);
	    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 = (pj_ioqueue_t*) 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_bzero(&callback, 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 = (char*) pj_pool_alloc(sock_data.pool, 128);
    sock_data.bufsize = 128;
    sock_data.op_key = (pj_ioqueue_op_key_t*) 
    		       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_bzero(&callback, 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;
	int n;

	pj_gettimeofday(&now);

	if (test_method == UNREGISTER_IN_APP && 
	    PJ_TIME_VAL_GTE(now, time_to_unregister) &&
	    !sock_data.unregistered) 
	{
	    sock_data.unregistered = 1;
	    /* Wait (as much as possible) for callback to complete */
	    pj_mutex_lock(sock_data.mutex);
	    pj_mutex_unlock(sock_data.mutex);
	    pj_ioqueue_unregister(sock_data.key);
	}

	if (PJ_TIME_VAL_GT(now, end_time) && sock_data.unregistered)
	    break;

	timeout.sec = 0; timeout.msec = 10;
	n = pj_ioqueue_poll(ioqueue, &timeout);
	if (n < 0) {
	    app_perror("pj_ioqueue_poll error", -n);
	    pj_thread_sleep(1);
	}
    }

    thread_quitting = 1;

    for (i=0; i<WORKER_CNT; ++i) {
	pj_thread_join(thread[i]);
	pj_thread_destroy(thread[i]);
    }

    /* Destroy data */
    pj_mutex_destroy(sock_data.mutex);
    pj_pool_release(sock_data.pool);
    sock_data.pool = NULL;

    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;
}

static int udp_ioqueue_unreg_test_imp(pj_bool_t allow_concur)
{
    enum { LOOP = 10 };
    int i, rc;
    char title[30];
    pj_ioqueue_t *ioqueue;
    pj_pool_t *test_pool;
	
    PJ_LOG(3,(THIS_FILE, "..testing with concurency=%d", allow_concur));

    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;
    }

    rc = pj_ioqueue_set_default_concurrency(ioqueue, allow_concur);
    if (rc != PJ_SUCCESS) {
	app_perror("Error in pj_ioqueue_set_default_concurrency()", rc);
	return -12;
    }

    PJ_LOG(3, (THIS_FILE, "...ioqueue unregister stress test 0/3, unregister in app (%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, unregister in app (%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, unregister in cb (%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, unregister in cb (%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;
}

int udp_ioqueue_unreg_test(void)
{
    int rc;

    rc = udp_ioqueue_unreg_test_imp(PJ_TRUE);
    if (rc != 0)
    	return rc;

    rc = udp_ioqueue_unreg_test_imp(PJ_FALSE);
    if (rc != 0)
	return rc;

    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 */