test_smsc.c

The Kannel Open Source WAP and SMS gateway works as both an SMS gateway, for implementing keyword b

	    	gwthread_create(smpp_emu_reader, thread[num_threads]);
	    if (thread[num_threads]->id == -1)
	    	panic(0, "Couldn't start SMPP subthread.");
    	    ++num_threads;
	}
    }
    
    for (i = 0; i < num_threads; ++i) {
	thread[i]->quit = 1;
    	gwthread_wakeup(thread[i]->id);
	gwthread_join(thread[i]->id);
	conn_destroy(thread[i]->conn);
	semaphore_destroy(thread[i]->ok_to_send);
	gw_free(thread[i]);
    }

    eq_remove_producer(eq);
}


/***********************************************************************
 * SMS center emulator, generic interface.
 */


static long smpp_emu_id = -1;


/*
 * Start all SMS center emulators.
 */
static void smsc_emu_create(EventQueue *eq)
{
    struct smsc_emu_arg *arg;
    
    gw_assert(smpp_emu_id == -1);

    arg = gw_malloc(sizeof(*arg));
    arg->sema = semaphore_create(0);
    arg->eq = eq;
    smpp_emu_id = gwthread_create(smpp_emu, arg);
    if (smpp_emu_id == -1)
    	panic(0, "Couldn't start SMPP emulator thread.");
    semaphore_down(arg->sema);
    semaphore_destroy(arg->sema);
    gw_free(arg);
}


static void smsc_emu_destroy(void)
{
    eq_remove_producer(undelivered_messages);
    gw_assert(smpp_emu_id != -1);
    gwthread_wakeup(smpp_emu_id);
    gwthread_join(smpp_emu_id);
}


static void smsc_emu_deliver(void)
{
    eq_append(undelivered_messages, eq_create_event(deliver));
}


static void smsc_emu_submit_ack(Event *e)
{
    SMPP_PDU *resp;
    Octstr *os;

    resp = smpp_pdu_create(submit_sm_resp, e->sequence_number);
    os = smpp_pdu_pack(resp);
    conn_write(e->conn, os);
    octstr_destroy(os);
    smpp_pdu_destroy(resp);
}


static void smsc_emu_init(void)
{
    smpp_emu_counter = counter_create();
    undelivered_messages = eq_create();
    eq_add_producer(undelivered_messages);
}


static void smsc_emu_shutdown(void)
{
    counter_destroy(smpp_emu_counter);
    eq_destroy(undelivered_messages);
}


/***********************************************************************
 * HTTP server emulator.
 */


static List *httpd_emu_headers = NULL;


struct httpd_emu_arg {
    int port;
    Semaphore *sema;
    EventQueue *eq;
};


/*
 * This is the HTTP server emulator thread.
 */
static void httpd_emu(void *arg)
{
    HTTPClient *client;
    Octstr *ip;
    Octstr *url;
    List *headers;
    Octstr *body;
    List *cgivars;
    struct httpd_emu_arg *p;
    EventQueue *eq;

    p = arg;
    eq = p->eq;
    eq_add_producer(eq);
    semaphore_up(p->sema);

    for (;;) {
	client = http_accept_request(p->port, &ip, &url, &headers, &body, 
	    	    	    	     &cgivars);
	if (client == NULL)
	    break;
    	
	eq_append(eq, eq_create_http_request(client, 
	    	    	    	    http_cgi_variable(cgivars, "arg")));
    	octstr_destroy(ip);
    	octstr_destroy(url);
	http_destroy_headers(headers);
    	octstr_destroy(body);
    	http_destroy_cgiargs(cgivars);
    }
    eq_remove_producer(eq);
    gw_free(p);
}


/*
 * Thread id for HTTP server emulator thread. It is needed for proper
 * shutdown.
 */
static long httpd_emu_tid = -1;


/*
 * Start the HTTP server emulator thread and return when it is 
 * ready to accept clients.
 */
static void httpd_emu_create(EventQueue *eq)
{
    struct httpd_emu_arg *arg;
    int ssl = 0;   /* indicate if SSL-enabled server should be used */

    if (http_open_port(http_port, ssl) == -1)
    	panic(0, "Can't open HTTP server emulator port %ld.", http_port);

    gw_assert(httpd_emu_tid == -1);
    arg = gw_malloc(sizeof(*arg));
    arg->port = http_port;
    arg->sema = semaphore_create(0);
    arg->eq = eq;
    httpd_emu_tid = gwthread_create(httpd_emu, arg);
    if (httpd_emu_tid == -1)
    	panic(0, "Can't start the HTTP server emulator thread.");
    semaphore_down(arg->sema);
    semaphore_destroy(arg->sema);
}


/*
 * Terminate the HTTP server emulator thread. Return when the thread
 * is quite dead.
 */
static void httpd_emu_destroy(void)
{
    gw_assert(httpd_emu_tid != -1);
    http_close_all_ports();
    gwthread_join(httpd_emu_tid);
    httpd_emu_tid = -1;
}


/*
 * Send a reply to an HTTP response.
 */
static void httpd_emu_reply(Event *e)
{
    http_send_reply(e->client, HTTP_OK, httpd_emu_headers, e->body);
}


static void httpd_emu_init(void)
{
    httpd_emu_headers = http_create_empty_headers();
    http_header_add(httpd_emu_headers, "Content-Type", "text/plain");
}


static void httpd_emu_shutdown(void)
{
    http_destroy_headers(httpd_emu_headers);
}


/***********************************************************************
 * Main program for N SMS messages benchmark.
 */


static void kill_kannel(void)
{
    Octstr *url;
    Octstr *final_url;
    List *req_headers;
    List *reply_headers;
    Octstr *reply_body;
    int ret;
    
    url = octstr_format("http://localhost:%ld/shutdown?password=%s",
    	    	    	admin_port, admin_password);
    req_headers = http_create_empty_headers();
    http_header_add(req_headers, "Content-Type", "text/plain");
    ret = http_get_real(HTTP_METHOD_GET, url, req_headers, &final_url, 
                        &reply_headers, &reply_body);
    if (ret != -1) {
    	octstr_destroy(final_url);
	http_destroy_headers(reply_headers);
    	octstr_destroy(reply_body);
    }
    octstr_destroy(url);
    http_destroy_headers(req_headers);
}


/*
 * This will try to have as large a sustained level of traffic as possible.
 */

enum { MAX_IN_AVERAGE = 100 };
enum { MAX_RTT = 1 };
enum { MAX_WAITING = 100 };

static void sustained_level_benchmark(void)
{
    EventQueue *eq;
    Event *e;
    long i;
    long num_deliver;
    long num_submit;
    long rtt;
    long times[MAX_IN_AVERAGE];
    long next_time;
    double time_sum;
    long num_unanswered;

    eq = eq_create();

    httpd_emu_create(eq);
    smsc_emu_create(eq);
    
    /* Wait for an SMS center client to appear. */
    while ((e = eq_extract(eq)) != NULL && e->type != got_smsc)
    	debug("test_smsc", 0, "Discarding event of type %s", eq_type(e));
    debug("test_smsc", 0, "Got event got_smsc.");
    eq_destroy_event(e);

    /* 
     * Send message when there are at most MAX_WAITING unanswered messages
     * and current average round trip time is less than MAX_RTT.
     */
    num_submit = 0;
    for (i = 0; i < MAX_IN_AVERAGE; ++i)
    	times[i] = 0;
    next_time = 0;
    time_sum = 0.0;
    num_unanswered = 0;
    num_deliver = 0;

    while (num_submit < num_messages) {
	for (;;) {
	    if (num_deliver >= num_messages || num_unanswered >= MAX_WAITING)
	    	break;
    	    if (time_sum / MAX_IN_AVERAGE >= MAX_RTT && num_unanswered > 0)
	    	break;
	    smsc_emu_deliver();
	    ++num_unanswered;
	    ++num_deliver;
	}

    	e = eq_extract(eq);
	if (e == NULL)
	    break;
	eq_log(e);

	switch (e->type) {
	case deliver_ack:
	    break;
	    
	case http_request:
	    httpd_emu_reply(e);
	    break;

	case submit:
	    rtt = eq_round_trip_time(e);
	    time_sum -= times[next_time];
	    times[next_time] = rtt;
	    time_sum += times[next_time];
	    debug("", 0, "RTT = %ld", rtt);
	    next_time = (next_time + 1) % MAX_IN_AVERAGE;
	    ++num_submit;
	    --num_unanswered;
	    smsc_emu_submit_ack(e);
	    break;
	    
    	case got_enquire_link:
	    break;

	default:
	    debug("test_smsc", 0, "Ignoring event of type %s", eq_type(e));
	    break;
	}
	
	eq_destroy_event(e);
    }

    kill_kannel();

    debug("test_smsc", 0, "Terminating benchmark.");
    smsc_emu_destroy();
    httpd_emu_destroy();
    eq_destroy(eq);
}


/*
 * This will send `num_messages' SMS messages as quickly as possible.
 */

enum { MAX_IN_QUEUE = 1000 };

static void n_messages_benchmark(void)
{
    EventQueue *eq;
    Event *e;
    long i;
    long num_submit;
    long num_in_queue;
    long num_deliver;

    eq = eq_create();

    httpd_emu_create(eq);
    smsc_emu_create(eq);
    
    /* Wait for an SMS center client to appear. */
    while ((e = eq_extract(eq)) != NULL && e->type != got_smsc)
    	debug("test_smsc", 0, "Discarding event of type %s", eq_type(e));
    debug("test_smsc", 0, "Got event got_smsc.");
    eq_destroy_event(e);

    /* Send the SMS messages, or at least fill the send queue. */
    for (i = 0; i < num_messages && i < MAX_IN_QUEUE; ++i)
    	smsc_emu_deliver();
    num_in_queue = i;
    num_deliver = i;

    /* 
     * Wait for results to be processed. When send queue is not full,
     * fill it.
     */
    num_submit = 0;
    while (num_submit < num_messages && (e = eq_extract(eq)) != NULL) {
    	while (num_deliver < num_messages && num_in_queue < MAX_IN_QUEUE) {
	    smsc_emu_deliver();
	    ++num_in_queue;
	    ++num_deliver;
	}

	eq_log(e);

	switch (e->type) {
	case deliver_ack:
	    break;
	    
	case http_request:
	    httpd_emu_reply(e);
	    break;

	case submit:
	    debug("", 0, "RTT = %ld", eq_round_trip_time(e));
	    smsc_emu_submit_ack(e);
	    ++num_submit;
	    --num_in_queue;
	    break;
	    
    	case got_enquire_link:
	    break;

	default:
	    debug("test_smsc", 0, "Ignoring event of type %s", eq_type(e));
	    break;
	}
	
	eq_destroy_event(e);
    }

    kill_kannel();

    debug("test_smsc", 0, "Terminating benchmark.");
    smsc_emu_destroy();
    httpd_emu_destroy();
    eq_destroy(eq);
}


/***********************************************************************
 * Main program.
 */


int main(int argc, char **argv)
{
    int opt;
    char *main_name;
    int i;
    static struct {
	char *name;
	void (*func)(void);
    } tab[] = {
	{ "n_messages", n_messages_benchmark },
	{ "sustained_level", sustained_level_benchmark },
    };

    gwlib_init();
    eq_init();
    httpd_emu_init();
    smsc_emu_init();

    main_name = "n_messages_benchmark";

    while ((opt = getopt(argc, argv, "m:r:")) != EOF) {
	switch (opt) {
	case 'm':
	    main_name = optarg;
	    break;
	case 'r':
	    num_messages = atoi(optarg);
	    break;
	}
    }

    for (i = 0; (size_t) i < sizeof(tab) / sizeof(tab[0]); ++i) {
	if (strcmp(main_name, tab[i].name) == 0) {
	    tab[i].func();
	    break;
	}
    }

    smsc_emu_shutdown();
    httpd_emu_shutdown();
    eq_shutdown();
    gwlib_shutdown();
    return 0;
}