Samples: SIP Performance Benchmark

pjsip-perf is a complete program to measure the performance of PJSIP or other SIP endpoints. It consists of two parts:

• the server, to respond incoming requests, and
• the client, who actively submits requests and measure the performance of the server.

Both server and client part can run simultaneously, to measure the performance when both endpoints are co-located in a single program.

The server accepts both INVITE and non-INVITE requests. The server exports several different types of URL, which would control how the request would be handled by the server:

• URL with "0" as the user part will be handled statelessly. It should not be used with INVITE method.
• URL with "1" as the user part will be handled statefully. If the request is an INVITE request, INVITE transaction will be created and 200/OK response will be sent, along with a valid SDP body. However, the SDP is just a static text body, and is not a proper SDP generated by PJMEDIA.
• URL with "2" as the user part is only meaningful for INVITE requests, as it would be handled call-statefully by the server. For this URL, the server also would generate SDP dynamically and perform a proper SDP negotiation for the incoming call. Also for every call, server will limit the call duration to 10 seconds, on which the call will be terminated if the client doesn't hangup the call.

This file is pjsip-apps/src/samples/pjsip-perf.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 all headers. */
#include <pjsip.h>
#include <pjmedia.h>
#include <pjmedia-codec.h>
#include <pjsip_ua.h>
#include <pjsip_simple.h>
#include <pjlib-util.h>
#include <pjlib.h>
#include <stdio.h>
 
#if (defined(PJ_WIN32) && PJ_WIN32!=0) || (defined(PJ_WIN64) && PJ_WIN64!=0)
#  include <windows.h>
#endif
 
#define THIS_FILE           "pjsip-perf.c"
#define DEFAULT_COUNT       (pjsip_cfg()->tsx.max_count/2>10000?10000:pjsip_cfg()->tsx.max_count/2)
#define JOB_WINDOW          1000
#define TERMINATE_TSX(x,c)
 
 
#ifndef CACHING_POOL_SIZE
#   define CACHING_POOL_SIZE   (256*1024*1024)
#endif
 
 
/* Static message body for INVITE, when stateful processing is
 * invoked (instead of call-stateful, where SDP is generated
 * dynamically.
 */
static pj_str_t dummy_sdp_str = 
{
    "v=0\r\n"
    "o=- 3360842071 3360842071 IN IP4 192.168.0.68\r\n"
    "s=pjmedia\r\n"
    "c=IN IP4 192.168.0.68\r\n"
    "t=0 0\r\n"
    "m=audio 4000 RTP/AVP 0 8 3 103 102 101\r\n"
    "a=rtcp:4001 IN IP4 192.168.0.68\r\n"
    "a=rtpmap:103 speex/16000\r\n"
    "a=rtpmap:102 speex/8000\r\n"
    "a=rtpmap:3 GSM/8000\r\n"
    "a=rtpmap:0 PCMU/8000\r\n"
    "a=rtpmap:8 PCMA/8000\r\n"
    "a=sendrecv\r\n"
    "a=rtpmap:101 telephone-event/8000\r\n"
    "a=fmtp:101 0-15\r\n",
    0
};
 
static pj_str_t mime_application = { "application", 11};
static pj_str_t mime_sdp = {"sdp", 3};
 
 
struct srv_state
{
    unsigned        stateless_cnt;
    unsigned        stateful_cnt;
    unsigned        call_cnt;
};
 
 
struct app
{
    pj_caching_pool      cp;
    pj_pool_t           *pool;
    pj_bool_t            use_tcp;
    pj_str_t             local_addr;
    int                  local_port;
    pjsip_endpoint      *sip_endpt;
    pjmedia_endpt       *med_endpt;
    pj_str_t             local_uri;
    pj_str_t             local_contact;
    unsigned             skinfo_cnt;
    pjmedia_sock_info    skinfo[8];
 
    pj_bool_t            thread_quit;
    unsigned             thread_count;
    pj_thread_t         *thread[16];
 
    pj_bool_t            real_sdp;
    pjmedia_sdp_session *dummy_sdp;
 
    int                  log_level;
 
    struct {
        pjsip_method         method;
        pj_str_t             dst_uri;
        pj_bool_t            stateless;
        unsigned             timeout;
        unsigned             job_count,
                             job_submitted, 
                             job_finished,
                             job_window;
        unsigned             stat_max_window;
        pj_time_val          first_request;
        pj_time_val          requests_sent;
        pj_time_val          last_completion;
        unsigned             total_responses;
        unsigned             response_codes[800];
    } client;
 
    struct {
        pj_bool_t send_trying;
        pj_bool_t send_ringing;
        unsigned delay;
        struct srv_state prev_state;
        struct srv_state cur_state;
    } server;
 
 
} app;
 
struct call
{
    pjsip_inv_session   *inv;
    pj_timer_entry       ans_timer;
};
 
 
static void app_perror(const char *sender, const char *title, 
                       pj_status_t status)
{
    char errmsg[PJ_ERR_MSG_SIZE];
 
    pj_strerror(status, errmsg, sizeof(errmsg));
    PJ_LOG(1,(sender, "%s: %s [code=%d]", title, errmsg, status));
}
 
 
/**************************************************************************
 * STATELESS SERVER
 */
static pj_bool_t mod_stateless_on_rx_request(pjsip_rx_data *rdata);
 
/* Module to handle incoming requests statelessly.
 */
static pjsip_module mod_stateless_server =
{
    NULL, NULL,                     /* prev, next.              */
    { "mod-stateless-server", 20 }, /* Name.                    */
    -1,                             /* Id                       */
    PJSIP_MOD_PRIORITY_APPLICATION, /* Priority                 */
    NULL,                           /* load()                   */
    NULL,                           /* start()                  */
    NULL,                           /* stop()                   */
    NULL,                           /* unload()                 */
    &mod_stateless_on_rx_request,   /* on_rx_request()          */
    NULL,                           /* on_rx_response()         */
    NULL,                           /* on_tx_request.           */
    NULL,                           /* on_tx_response()         */
    NULL,                           /* on_tsx_state()           */
};
 
 
static pj_bool_t mod_stateless_on_rx_request(pjsip_rx_data *rdata)
{
    const pj_str_t stateless_user = { "0", 1 };
    pjsip_uri *uri;
    pjsip_sip_uri *sip_uri;
 
    uri = pjsip_uri_get_uri(rdata->msg_info.msg->line.req.uri);
 
    /* Only want to receive SIP/SIPS scheme */
    if (!PJSIP_URI_SCHEME_IS_SIP(uri) && !PJSIP_URI_SCHEME_IS_SIPS(uri))
        return PJ_FALSE;
 
    sip_uri = (pjsip_sip_uri*) uri;
 
    /* Check for matching user part */
    if (pj_strcmp(&sip_uri->user, &stateless_user)!=0)
        return PJ_FALSE;
 
    /*
     * Yes, this is for us.
     */
 
    /* Ignore ACK request */
    if (rdata->msg_info.msg->line.req.method.id == PJSIP_ACK_METHOD)
        return PJ_TRUE;
 
    /*
     * Respond statelessly with 200/OK.
     */
    pjsip_endpt_respond_stateless(app.sip_endpt, rdata, 200, NULL,
                                  NULL, NULL);
    app.server.cur_state.stateless_cnt++;
    return PJ_TRUE;
}
 
 
/**************************************************************************
 * STATEFUL SERVER
 */
static pj_bool_t mod_stateful_on_rx_request(pjsip_rx_data *rdata);
 
/* Module to handle incoming requests statefully.
 */
static pjsip_module mod_stateful_server =
{
    NULL, NULL,                     /* prev, next.              */
    { "mod-stateful-server", 19 },  /* Name.                    */
    -1,                             /* Id                       */
    PJSIP_MOD_PRIORITY_APPLICATION, /* Priority                 */
    NULL,                           /* load()                   */
    NULL,                           /* start()                  */
    NULL,                           /* stop()                   */
    NULL,                           /* unload()                 */
    &mod_stateful_on_rx_request,   /* on_rx_request()           */
    NULL,                           /* on_rx_response()         */
    NULL,                           /* on_tx_request.           */
    NULL,                           /* on_tx_response()         */
    NULL,                           /* on_tsx_state()           */
};
 
 
static pj_bool_t mod_stateful_on_rx_request(pjsip_rx_data *rdata)
{
    const pj_str_t stateful_user = { "1", 1 };
    pjsip_uri *uri;
    pjsip_sip_uri *sip_uri;
 
    uri = pjsip_uri_get_uri(rdata->msg_info.msg->line.req.uri);
 
    /* Only want to receive SIP/SIPS scheme */
    if (!PJSIP_URI_SCHEME_IS_SIP(uri) && !PJSIP_URI_SCHEME_IS_SIPS(uri))
        return PJ_FALSE;
 
    sip_uri = (pjsip_sip_uri*) uri;
 
    /* Check for matching user part */
    if (pj_strcmp(&sip_uri->user, &stateful_user)!=0)
        return PJ_FALSE;
 
    /*
     * Yes, this is for us.
     * Respond statefully with 200/OK.
     */
    switch (rdata->msg_info.msg->line.req.method.id) {
    case PJSIP_INVITE_METHOD:
        {
            pjsip_msg_body *body;
 
            if (dummy_sdp_str.slen == 0)
                dummy_sdp_str.slen = pj_ansi_strlen(dummy_sdp_str.ptr);
 
            body = pjsip_msg_body_create(rdata->tp_info.pool, 
                                         &mime_application, &mime_sdp, 
                                         &dummy_sdp_str);
            pjsip_endpt_respond(app.sip_endpt, &mod_stateful_server, rdata,
                                200, NULL, NULL, body, NULL);
        }
        break;
    case PJSIP_ACK_METHOD:
        return PJ_TRUE;
    default:
        pjsip_endpt_respond(app.sip_endpt, &mod_stateful_server, rdata,
                            200, NULL, NULL, NULL, NULL);
        break;
    }
 
    app.server.cur_state.stateful_cnt++;
    return PJ_TRUE;
}
 
 
/**************************************************************************
 * CALL SERVER
 */
static pj_bool_t mod_call_on_rx_request(pjsip_rx_data *rdata);
 
/* Module to handle incoming requests callly.
 */
static pjsip_module mod_call_server =
{
    NULL, NULL,                     /* prev, next.              */
    { "mod-call-server", 15 },      /* Name.                    */
    -1,                             /* Id                       */
    PJSIP_MOD_PRIORITY_APPLICATION, /* Priority                 */
    NULL,                           /* load()                   */
    NULL,                           /* start()                  */
    NULL,                           /* stop()                   */
    NULL,                           /* unload()                 */
    &mod_call_on_rx_request,        /* on_rx_request()          */
    NULL,                           /* on_rx_response()         */
    NULL,                           /* on_tx_request.           */
    NULL,                           /* on_tx_response()         */
    NULL,                           /* on_tsx_state()           */
};
 
 
static pj_status_t send_response(pjsip_inv_session *inv, 
                                 pjsip_rx_data *rdata,
                                 int code,
                                 pj_bool_t *has_initial)
{
    pjsip_tx_data *tdata;
    pj_status_t status;
 
    if (*has_initial) {
        status = pjsip_inv_answer(inv, code, NULL, NULL, &tdata);
    } else {
        status = pjsip_inv_initial_answer(inv, rdata, code, 
                                          NULL, NULL, &tdata);
    }
 
    if (status != PJ_SUCCESS) {
        if (*has_initial) {
            status = pjsip_inv_answer(inv, PJSIP_SC_NOT_ACCEPTABLE, 
                                      NULL, NULL, &tdata);
        } else {
            status = pjsip_inv_initial_answer(inv, rdata, 
                                              PJSIP_SC_NOT_ACCEPTABLE,
                                              NULL, NULL, &tdata);
        }
 
        if (status == PJ_SUCCESS) {
            *has_initial = PJ_TRUE;
            pjsip_inv_send_msg(inv, tdata); 
        } else {
            pjsip_inv_terminate(inv, 500, PJ_FALSE);
            return -1;
        }
    } else {
        *has_initial = PJ_TRUE;
 
        status = pjsip_inv_send_msg(inv, tdata); 
        if (status != PJ_SUCCESS) {
            pjsip_tx_data_dec_ref(tdata);
            return status;
        }
    }
 
    return status;
}
 
static void answer_timer_cb(pj_timer_heap_t *h, pj_timer_entry *entry)
{
    struct call *call = entry->user_data;
    pj_bool_t has_initial = PJ_TRUE;
 
    PJ_UNUSED_ARG(h);
 
    entry->id = 0;
    send_response(call->inv, NULL, 200, &has_initial);
}
 
static pj_bool_t mod_call_on_rx_request(pjsip_rx_data *rdata)
{
    const pj_str_t call_user = { "2", 1 };
    pjsip_uri *uri;
    pjsip_sip_uri *sip_uri;
    struct call *call;
    pjsip_dialog *dlg;
    pjmedia_sdp_session *sdp;
    pjsip_tx_data *tdata;
    pj_bool_t has_initial = PJ_FALSE;
    pj_status_t status;
 
    uri = pjsip_uri_get_uri(rdata->msg_info.msg->line.req.uri);
 
    /* Only want to receive SIP/SIPS scheme */
    if (!PJSIP_URI_SCHEME_IS_SIP(uri) && !PJSIP_URI_SCHEME_IS_SIPS(uri))
        return PJ_FALSE;
 
    sip_uri = (pjsip_sip_uri*) uri;
 
    /* Only want to handle INVITE requests. */
    if (rdata->msg_info.msg->line.req.method.id != PJSIP_INVITE_METHOD) {
        return PJ_FALSE;
    }
 
 
    /* Check for matching user part. Incoming requests will be handled 
     * call-statefully if:
     *  - user part is "2", or
     *  - user part is not "0" nor "1" and method is INVITE.
     */
    if (pj_strcmp(&sip_uri->user, &call_user) == 0 ||
        sip_uri->user.slen != 1 ||
        (*sip_uri->user.ptr != '0' && *sip_uri->user.ptr != '1'))
    {
        /* Match */
 
    } else {
        return PJ_FALSE;
    }
 
 
    /* Verify that we can handle the request. */
    if (app.real_sdp) {
        unsigned options = 0;
        status = pjsip_inv_verify_request(rdata, &options, NULL, NULL,
                                          app.sip_endpt, &tdata);
        if (status != PJ_SUCCESS) {
 
            /*
             * No we can't handle the incoming INVITE request.
             */
 
            if (tdata) {
                pjsip_response_addr res_addr;
 
                pjsip_get_response_addr(tdata->pool, rdata, &res_addr);
                status = pjsip_endpt_send_response(app.sip_endpt, &res_addr, tdata, 
                                          NULL, NULL);
                if (status != PJ_SUCCESS) pjsip_tx_data_dec_ref(tdata);
            } else {
                /* Respond with 500 (Internal Server Error) */
                pjsip_endpt_respond_stateless(app.sip_endpt, rdata, 500, NULL,
                                              NULL, NULL);
            }
 
            return PJ_TRUE;
        } 
    }
 
    /* Create UAS dialog */
    status = pjsip_dlg_create_uas_and_inc_lock( pjsip_ua_instance(), rdata,
                                                &app.local_contact, &dlg);
    if (status != PJ_SUCCESS) {
        const pj_str_t reason = pj_str("Unable to create dialog");
        pjsip_endpt_respond_stateless( app.sip_endpt, rdata, 
                                       500, &reason,
                                       NULL, NULL);
        return PJ_TRUE;
    }
 
    /* Alloc call structure. */
    call = pj_pool_zalloc(dlg->pool, sizeof(struct call));
 
    /* Create SDP from PJMEDIA */
    if (app.real_sdp) {
        status = pjmedia_endpt_create_sdp(app.med_endpt, rdata->tp_info.pool, 
                                          app.skinfo_cnt, app.skinfo, 
                                          &sdp);
    } else {
        sdp = app.dummy_sdp;
    }
 
    /* Create UAS invite session */
    status = pjsip_inv_create_uas( dlg, rdata, sdp, 0, &call->inv);
    if (status != PJ_SUCCESS) {
        pjsip_dlg_create_response(dlg, rdata, 500, NULL, &tdata);
        pjsip_dlg_send_response(dlg, pjsip_rdata_get_tsx(rdata), tdata);
        pjsip_dlg_dec_lock(dlg);
        return PJ_TRUE;
    }
    
    /* Invite session has been created, decrement & release dialog lock. */
    pjsip_dlg_dec_lock(dlg);
 
    /* Send 100/Trying if needed */
    if (app.server.send_trying) {
        status = send_response(call->inv, rdata, 100, &has_initial);
        if (status != PJ_SUCCESS)
            return PJ_TRUE;
    }
 
    /* Send 180/Ringing if needed */
    if (app.server.send_ringing) {
        status = send_response(call->inv, rdata, 180, &has_initial);
        if (status != PJ_SUCCESS)
            return PJ_TRUE;
    }
 
    /* Simulate call processing delay */
    if (app.server.delay) {
        pj_time_val delay;
 
        call->ans_timer.id = 1;
        call->ans_timer.user_data = call;
        call->ans_timer.cb = &answer_timer_cb;
        
        delay.sec = 0;
        delay.msec = app.server.delay;
        pj_time_val_normalize(&delay);
 
        pjsip_endpt_schedule_timer(app.sip_endpt, &call->ans_timer, &delay);
 
    } else {
        /* Send the 200 response immediately . */  
        status = send_response(call->inv, rdata, 200, &has_initial);
        PJ_ASSERT_ON_FAIL(status == PJ_SUCCESS, return PJ_TRUE);
    }
 
    /* Done */
    app.server.cur_state.call_cnt++;
 
    return PJ_TRUE;
}
 
 
 
/**************************************************************************
 * Default handler when incoming request is not handled by any other
 * modules.
 */
static pj_bool_t mod_responder_on_rx_request(pjsip_rx_data *rdata);
 
/* Module to handle incoming requests statelessly.
 */
static pjsip_module mod_responder =
{
    NULL, NULL,                     /* prev, next.              */
    { "mod-responder", 13 },        /* Name.                    */
    -1,                             /* Id                       */
    PJSIP_MOD_PRIORITY_APPLICATION+1, /* Priority               */
    NULL,                           /* load()                   */
    NULL,                           /* start()                  */
    NULL,                           /* stop()                   */
    NULL,                           /* unload()                 */
    &mod_responder_on_rx_request,   /* on_rx_request()          */
    NULL,                           /* on_rx_response()         */
    NULL,                           /* on_tx_request.           */
    NULL,                           /* on_tx_response()         */
    NULL,                           /* on_tsx_state()           */
};
 
 
static pj_bool_t mod_responder_on_rx_request(pjsip_rx_data *rdata)
{
    const pj_str_t reason = pj_str("Not expecting request at this URI");
 
    /*
     * Respond any requests (except ACK!) with 500.
     */
    if (rdata->msg_info.msg->line.req.method.id != PJSIP_ACK_METHOD) {
        pjsip_endpt_respond_stateless(app.sip_endpt, rdata, 500, &reason,
                                      NULL, NULL);
    }
 
    return PJ_TRUE;
}
 
 
 
/*****************************************************************************
 * Below is a simple module to log all incoming and outgoing SIP messages
 */
 
 
/* Notification on incoming messages */
static pj_bool_t logger_on_rx_msg(pjsip_rx_data *rdata)
{
    PJ_LOG(3,(THIS_FILE, "RX %d bytes %s from %s %s:%d:\n"
                         "%.*s\n"
                         "--end msg--",
                         rdata->msg_info.len,
                         pjsip_rx_data_get_info(rdata),
                         rdata->tp_info.transport->type_name,
                         rdata->pkt_info.src_name,
                         rdata->pkt_info.src_port,
                         (int)rdata->msg_info.len,
                         rdata->msg_info.msg_buf));
    
    /* Always return false, otherwise messages will not get processed! */
    return PJ_FALSE;
}
 
/* Notification on outgoing messages */
static pj_status_t logger_on_tx_msg(pjsip_tx_data *tdata)
{
    
    /* Important note:
     *  tp_info field is only valid after outgoing messages has passed
     *  transport layer. So don't try to access tp_info when the module
     *  has lower priority than transport layer.
     */
 
    PJ_LOG(3,(THIS_FILE, "TX %d bytes %s to %s %s:%d:\n"
                         "%.*s\n"
                         "--end msg--",
                         (tdata->buf.cur - tdata->buf.start),
                         pjsip_tx_data_get_info(tdata),
                         tdata->tp_info.transport->type_name,
                         tdata->tp_info.dst_name,
                         tdata->tp_info.dst_port,
                         (int)(tdata->buf.cur - tdata->buf.start),
                         tdata->buf.start));
 
    /* Always return success, otherwise message will not get sent! */
    return PJ_SUCCESS;
}
 
/* The module instance. */
static pjsip_module msg_logger = 
{
    NULL, NULL,                         /* prev, next.          */
    { "mod-siprtp-log", 14 },           /* Name.                */
    -1,                                 /* Id                   */
    PJSIP_MOD_PRIORITY_TRANSPORT_LAYER-1,/* Priority            */
    NULL,                               /* load()               */
    NULL,                               /* start()              */
    NULL,                               /* stop()               */
    NULL,                               /* unload()             */
    &logger_on_rx_msg,                  /* on_rx_request()      */
    &logger_on_rx_msg,                  /* on_rx_response()     */
    &logger_on_tx_msg,                  /* on_tx_request.       */
    &logger_on_tx_msg,                  /* on_tx_response()     */
    NULL,                               /* on_tsx_state()       */
 
};
 
 
 
/**************************************************************************
 * Test Client.
 */
 
static pj_bool_t mod_test_on_rx_response(pjsip_rx_data *rdata);
 
static void call_on_media_update( pjsip_inv_session *inv,
                                  pj_status_t status);
static void call_on_state_changed( pjsip_inv_session *inv, 
                                   pjsip_event *e);
static void call_on_forked(pjsip_inv_session *inv, pjsip_event *e);
 
 
/* Module to handle incoming requests callly.
 */
static pjsip_module mod_test =
{
    NULL, NULL,                     /* prev, next.              */
    { "mod-test", 8 },              /* Name.                    */
    -1,                             /* Id                       */
    PJSIP_MOD_PRIORITY_APPLICATION, /* Priority                 */
    NULL,                           /* load()                   */
    NULL,                           /* start()                  */
    NULL,                           /* stop()                   */
    NULL,                           /* unload()                 */
    NULL,                           /* on_rx_request()          */
    &mod_test_on_rx_response,       /* on_rx_response()         */
    NULL,                           /* on_tx_request.           */
    NULL,                           /* on_tx_response()         */
    NULL,                           /* on_tsx_state()           */
};
 
 
static void report_completion(int status_code)
{
    app.client.job_finished++;
    if (status_code >= 200 && status_code < 800)
        app.client.response_codes[status_code]++;
    app.client.total_responses++;
    pj_gettimeofday(&app.client.last_completion);
}
 
 
/* Handler when response is received. */
static pj_bool_t mod_test_on_rx_response(pjsip_rx_data *rdata)
{
    if (pjsip_rdata_get_tsx(rdata) == NULL) {
        report_completion(rdata->msg_info.msg->line.status.code);
    }
 
    return PJ_TRUE;
}
 
 
/*
 * Create app
 */
static pj_status_t create_app(void)
{
    pj_status_t status;
 
    status = pj_init();
    if (status != PJ_SUCCESS) {
        app_perror(THIS_FILE, "Error initializing pjlib", status);
        return status;
    }
 
    /* init PJLIB-UTIL: */
    status = pjlib_util_init();
    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
 
    /* Must create a pool factory before we can allocate any memory. */
    pj_caching_pool_init(&app.cp, &pj_pool_factory_default_policy, 
                         CACHING_POOL_SIZE);
 
    /* Create application pool for misc. */
    app.pool = pj_pool_create(&app.cp.factory, "app", 1000, 1000, NULL);
 
    /* Create the endpoint: */
    status = pjsip_endpt_create(&app.cp.factory, pj_gethostname()->ptr, 
                                &app.sip_endpt);
    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
 
 
    return status;
}
 
 
/*
 * Init SIP stack
 */
static pj_status_t init_sip()
{
    pj_status_t status = -1;
 
    /* Add UDP/TCP transport. */
    {
        pj_sockaddr_in addr;
        pjsip_host_port addrname;
        const char *transport_type = NULL;
 
        pj_bzero(&addr, sizeof(addr));
        addr.sin_family = pj_AF_INET();
        addr.sin_addr.s_addr = 0;
        addr.sin_port = pj_htons((pj_uint16_t)app.local_port);
 
        if (app.local_addr.slen) {
            addrname.host = app.local_addr;
            addrname.port = 5060;
        } 
        if (app.local_port != 0)
            addrname.port = app.local_port;
 
        if (0) {
#if defined(PJ_HAS_TCP) && PJ_HAS_TCP!=0
        } else if (app.use_tcp) {
            pj_sockaddr_in local_addr;
            pjsip_tpfactory *tpfactory;
            
            transport_type = "tcp";
            pj_sockaddr_in_init(&local_addr, 0, (pj_uint16_t)app.local_port);
            status = pjsip_tcp_transport_start(app.sip_endpt, &local_addr,
                                               app.thread_count, &tpfactory);
            if (status == PJ_SUCCESS) {
                app.local_addr = tpfactory->addr_name.host;
                app.local_port = tpfactory->addr_name.port;
            }
#endif
        } else {
            pjsip_transport *tp;
 
            transport_type = "udp";
            status = pjsip_udp_transport_start(app.sip_endpt, &addr, 
                                               (app.local_addr.slen ? &addrname:NULL),
                                               app.thread_count, &tp);
            if (status == PJ_SUCCESS) {
                app.local_addr = tp->local_name.host;
                app.local_port = tp->local_name.port;
            }
 
        }
        if (status != PJ_SUCCESS) {
            app_perror(THIS_FILE, "Unable to start transport", status);
            return status;
        }
 
        app.local_uri.ptr = pj_pool_alloc(app.pool, 128);
        app.local_uri.slen = pj_ansi_sprintf(app.local_uri.ptr, 
                                             "<sip:pjsip-perf@%.*s:%d;transport=%s>",
                                             (int)app.local_addr.slen,
                                             app.local_addr.ptr,
                                             app.local_port,
                                             transport_type);
 
        app.local_contact = app.local_uri;
    }
 
    /* 
     * Init transaction layer.
     * This will create/initialize transaction hash tables etc.
     */
    status = pjsip_tsx_layer_init_module(app.sip_endpt);
    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
 
    /*  Initialize UA layer. */
    status = pjsip_ua_init_module( app.sip_endpt, NULL );
    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
 
    /* Initialize 100rel support */
    status = pjsip_100rel_init_module(app.sip_endpt);
    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
 
    /*  Init invite session module. */
    {
        pjsip_inv_callback inv_cb;
 
        /* Init the callback for INVITE session: */
        pj_bzero(&inv_cb, sizeof(inv_cb));
        inv_cb.on_state_changed = &call_on_state_changed;
        inv_cb.on_new_session = &call_on_forked;
        inv_cb.on_media_update = &call_on_media_update;
 
        /* Initialize invite session module:  */
        status = pjsip_inv_usage_init(app.sip_endpt, &inv_cb);
        PJ_ASSERT_RETURN(status == PJ_SUCCESS, 1);
    }
 
    /* Register our module to receive incoming requests. */
    status = pjsip_endpt_register_module( app.sip_endpt, &mod_test);
    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
 
 
    /* Register stateless server module */
    status = pjsip_endpt_register_module( app.sip_endpt, &mod_stateless_server);
    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
 
    /* Register default responder module */
    status = pjsip_endpt_register_module( app.sip_endpt, &mod_responder);
    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
 
    /* Register stateless server module */
    status = pjsip_endpt_register_module( app.sip_endpt, &mod_stateful_server);
    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
 
 
    /* Register call server module */
    status = pjsip_endpt_register_module( app.sip_endpt, &mod_call_server);
    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
 
 
    /* Done */
    return PJ_SUCCESS;
}
 
 
/*
 * Destroy SIP
 */
static void destroy_app()
{
    unsigned i;
 
    app.thread_quit = 1;
    for (i=0; i<app.thread_count; ++i) {
        if (app.thread[i]) {
            pj_thread_join(app.thread[i]);
            pj_thread_destroy(app.thread[i]);
            app.thread[i] = NULL;
        }
    }
 
    if (app.sip_endpt) {
        pjsip_endpt_destroy(app.sip_endpt);
        app.sip_endpt = NULL;
    }
 
    if (app.pool) {
        pj_pool_release(app.pool);
        app.pool = NULL;
        PJ_LOG(3,(THIS_FILE, "Peak memory size: %uMB",
                             app.cp.peak_used_size / 1000000));
        pj_caching_pool_destroy(&app.cp);
    }
 
    /* Shutdown PJLIB */
    pj_shutdown();
}
 
 
/*
 * Init media stack.
 */
static pj_status_t init_media()
{
    unsigned    i;
    pj_uint16_t rtp_port;
    pj_status_t status;
 
 
    /* Initialize media endpoint so that at least error subsystem is properly
     * initialized.
     */
    status = pjmedia_endpt_create(&app.cp.factory, 
                                  pjsip_endpt_get_ioqueue(app.sip_endpt), 0, 
                                  &app.med_endpt);
    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
 
 
    /* Must register all codecs to be supported */
    pjmedia_codec_register_audio_codecs(app.med_endpt, NULL);
 
    /* Init dummy socket addresses */
    app.skinfo_cnt = 0;
    for (i=0, rtp_port=4000; i<PJ_ARRAY_SIZE(app.skinfo); ++i, rtp_port+=2) {
        pjmedia_sock_info *skinfo;
 
        skinfo = &app.skinfo[i];
        
        pj_sockaddr_in_init(&skinfo->rtp_addr_name.ipv4, &app.local_addr,
                            (pj_uint16_t)rtp_port);
        pj_sockaddr_in_init(&skinfo->rtp_addr_name.ipv4, &app.local_addr,
                            (pj_uint16_t)(rtp_port+1));
        app.skinfo_cnt++;
    }
 
    /* Generate dummy SDP */
    dummy_sdp_str.slen = pj_ansi_strlen(dummy_sdp_str.ptr);
    status = pjmedia_sdp_parse(app.pool, dummy_sdp_str.ptr, dummy_sdp_str.slen, 
                               &app.dummy_sdp);
    if (status != PJ_SUCCESS) {
        app_perror(THIS_FILE, "Error parsing dummy SDP", status);
        return status;
    }
 
 
    /* Done */
    return PJ_SUCCESS;
}
 
 
/* This is notification from the call about media negotiation
 * status. This is called for client calls only.
 */
static void call_on_media_update( pjsip_inv_session *inv,
                                  pj_status_t status)
{
    if (status != PJ_SUCCESS) {
        pjsip_tx_data *tdata;
        pj_status_t status2;
 
        status2 = pjsip_inv_end_session(inv, PJSIP_SC_UNSUPPORTED_MEDIA_TYPE,
                                       NULL, &tdata);
        if (status2 == PJ_SUCCESS && tdata)
            status2 = pjsip_inv_send_msg(inv, tdata);
    }
}
 
 
/* This is notification from the call when the call state has changed.
 * This is called for client calls only.
 */
static void call_on_state_changed( pjsip_inv_session *inv, 
                                   pjsip_event *e)
{
    PJ_UNUSED_ARG(e);
 
    /* Bail out if the session has been counted before */
    if (inv->mod_data[mod_test.id] != NULL)
        return;
 
    /* Bail out if this is not an outgoing call */
    if (inv->role != PJSIP_UAC_ROLE)
        return;
 
    if (inv->state == PJSIP_INV_STATE_CONFIRMED) {
        pjsip_tx_data *tdata;
        pj_status_t status;
 
        //report_completion(200);
        //inv->mod_data[mod_test.id] = (void*)1;
 
        status = pjsip_inv_end_session(inv, PJSIP_SC_OK, NULL, &tdata);
        if (status == PJ_SUCCESS && tdata)
            status = pjsip_inv_send_msg(inv, tdata);
 
    } else if (inv->state == PJSIP_INV_STATE_DISCONNECTED) {
        report_completion(inv->cause);
        inv->mod_data[mod_test.id] = (void*)(pj_ssize_t)1;
    }
}
 
 
/* Not implemented for now */
static void call_on_forked(pjsip_inv_session *inv, pjsip_event *e)
{
    /* Do nothing */
    PJ_UNUSED_ARG(inv);
    PJ_UNUSED_ARG(e);
}
 
 
/*
 * Make outgoing call.
 */
static pj_status_t make_call(const pj_str_t *dst_uri)
{
    struct call *call;
    pjsip_dialog *dlg;
    pjmedia_sdp_session *sdp;
    pjsip_tx_data *tdata;
    pj_status_t status;
 
 
    /* Create UAC dialog */
    status = pjsip_dlg_create_uac( pjsip_ua_instance(), 
                                   &app.local_uri,      /* local URI        */
                                   &app.local_contact,  /* local Contact    */
                                   dst_uri,             /* remote URI       */
                                   dst_uri,             /* remote target    */
                                   &dlg);               /* dialog           */
    if (status != PJ_SUCCESS) {
        return status;
    }
 
    /* Create call */
    call = pj_pool_zalloc(dlg->pool, sizeof(struct call));
 
    /* Create SDP */
    if (app.real_sdp) {
        status = pjmedia_endpt_create_sdp(app.med_endpt, dlg->pool, 1, 
                                          app.skinfo, &sdp);
        if (status != PJ_SUCCESS) {
            pjsip_dlg_terminate(dlg);
            return status;
        }
    } else
        sdp = app.dummy_sdp;
 
    /* Create the INVITE session. */
    status = pjsip_inv_create_uac( dlg, sdp, 0, &call->inv);
    if (status != PJ_SUCCESS) {
        pjsip_dlg_terminate(dlg);
        return status;
    }
 
 
    /* Create initial INVITE request.
     * This INVITE request will contain a perfectly good request and 
     * an SDP body as well.
     */
    status = pjsip_inv_invite(call->inv, &tdata);
    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
 
 
    /* Send initial INVITE request. 
     * From now on, the invite session's state will be reported to us
     * via the invite session callbacks.
     */
    status = pjsip_inv_send_msg(call->inv, tdata);
    PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
 
 
    return PJ_SUCCESS;
}
 
 
/*
 * Verify that valid SIP url is given.
 */
static pj_status_t verify_sip_url(const char *c_url)
{
    pjsip_uri *p;
    pj_pool_t *pool;
    char *url;
    pj_size_t len = (c_url ? pj_ansi_strlen(c_url) : 0);
 
    if (!len) return -1;
 
    pool = pj_pool_create(&app.cp.factory, "check%p", 1024, 0, NULL);
    if (!pool) return PJ_ENOMEM;
 
    url = pj_pool_alloc(pool, len+1);
    pj_ansi_strcpy(url, c_url);
    url[len] = '\0';
 
    p = pjsip_parse_uri(pool, url, len, 0);
    if (!p || pj_stricmp2(pjsip_uri_get_scheme(p), "sip") != 0)
        p = NULL;
 
    pj_pool_release(pool);
    return p ? 0 : -1;
}
 
 
static void usage(void)
{
    printf(
        "Usage:\n"
        "   pjsip-perf [OPTIONS]        -- to start as server\n"
        "   pjsip-perf [OPTIONS] URL    -- to call server (possibly itself)\n"
        "\n"
        "where:\n"
        "   URL                     The SIP URL to be contacted.\n"
        "\n"
        "Client options:\n"
        "   --method=METHOD, -m     Set test method (set to INVITE for call benchmark)\n"
        "                           [default: OPTIONS]\n"
        "   --count=N, -n           Set total number of requests to initiate\n"
        "                           [default=%d]\n"
        "   --stateless, -s         Set to operate in stateless mode\n"
        "                           [default: stateful]\n"
        "   --timeout=SEC, -t       Set client timeout [default=60 sec]\n"
        "   --window=COUNT, -w      Set maximum outstanding job [default: %d]\n"
        "\n"
        "SDP options (client and server):\n"
        "   --real-sdp              Generate real SDP from pjmedia, and also perform\n"
        "                           proper SDP negotiation [default: dummy]\n"
        "\n"
        "Client and Server options:\n"
        "   --local-port=PORT, -p   Set local port [default: 5060]\n"
        "   --use-tcp, -T           Use TCP instead of UDP. Note that when started as\n"
        "                           client, you must add ;transport=tcp parameter to URL\n"
        "                           [default: no]\n"
        "   --thread-count=N        Set number of worker threads [default=1]\n"
        "   --trying                Send 100/Trying response (server, default no)\n"
        "   --ringing               Send 180/Ringing response (server, default no)\n"
        "   --delay=MS, -d          Delay answering call by MS (server, default no)\n"
        "\n"
        "Misc options:\n"
        "   --help, -h              Display this screen\n"
        "   --verbose, -v           Verbose logging (put more than once for even more)\n"
        "\n"
        "When started as server, pjsip-perf can be contacted on the following URIs:\n"
        "   - sip:0@server-addr     To handle requests statelessly.\n"
        "   - sip:1@server-addr     To handle requests statefully.\n"
        "   - sip:2@server-addr     To handle INVITE call.\n",
        DEFAULT_COUNT, JOB_WINDOW);
}
 
 
static int my_atoi(const char *s)
{
    pj_str_t ss = pj_str((char*)s);
    return pj_strtoul(&ss);
}
 
 
static pj_status_t init_options(int argc, char *argv[])
{
    enum { OPT_THREAD_COUNT = 1, OPT_REAL_SDP, OPT_TRYING, OPT_RINGING };
    struct pj_getopt_option long_options[] = {
        { "local-port",     1, 0, 'p' },
        { "count",          1, 0, 'c' },
        { "thread-count",   1, 0, OPT_THREAD_COUNT },
        { "method",         1, 0, 'm' },
        { "help",           0, 0, 'h' },
        { "stateless",      0, 0, 's' },
        { "timeout",        1, 0, 't' },
        { "real-sdp",       0, 0, OPT_REAL_SDP },
        { "verbose",        0, 0, 'v' },
        { "use-tcp",        0, 0, 'T' },
        { "window",         1, 0, 'w' },
        { "delay",          1, 0, 'd' },
        { "trying",         0, 0, OPT_TRYING},
        { "ringing",        0, 0, OPT_RINGING},
        { NULL, 0, 0, 0 },
    };
    int c;
    int option_index;
 
    /* Init default application configs */
    app.local_port = 5060;
    app.thread_count = 1;
    app.client.job_count = DEFAULT_COUNT;
    app.client.method = *pjsip_get_options_method();
    app.client.job_window = c = JOB_WINDOW;
    app.client.timeout = 60;
    app.log_level = 3;
 
 
    /* Parse options */
    pj_optind = 0;
    while((c=pj_getopt_long(argc,argv, "p:c:m:t:w:d:hsv", 
                            long_options, &option_index))!=-1) 
    {
        switch (c) {
        case 'p':
            app.local_port = my_atoi(pj_optarg);
            if (app.local_port < 0 || app.local_port > 65535) {
                PJ_LOG(3,(THIS_FILE, "Invalid --local-port %s", pj_optarg));
                return -1;
            }
            break;
 
        case 'c':
            app.client.job_count = my_atoi(pj_optarg);
            if (app.client.job_count > pjsip_cfg()->tsx.max_count)
                PJ_LOG(3,(THIS_FILE, 
                          "Warning: --count value (%d) exceeds maximum "
                          "transaction count (%d)", app.client.job_count,
                          pjsip_cfg()->tsx.max_count));
            break;
 
        case OPT_THREAD_COUNT:
            app.thread_count = my_atoi(pj_optarg);
            if (app.thread_count < 1 || app.thread_count > 16) {
                PJ_LOG(3,(THIS_FILE, "Invalid --thread-count %s", pj_optarg));
                return -1;
            }
            break;
 
        case 'm':
            {
                pj_str_t temp = pj_str((char*)pj_optarg);
                pjsip_method_init_np(&app.client.method, &temp);
            }
            break;
 
        case 'h':
            usage();
            return -1;
 
        case 's':
            app.client.stateless = PJ_TRUE;
            break;
 
        case OPT_REAL_SDP:
            app.real_sdp = 1;
            break;
 
        case 'v':
            app.log_level++;
            break;
 
        case 't':
            app.client.timeout = my_atoi(pj_optarg);
            if (app.client.timeout > 600) {
                PJ_LOG(3,(THIS_FILE, "Invalid --timeout %s", pj_optarg));
                return -1;
            }
            break;
 
        case 'w':
            app.client.job_window = my_atoi(pj_optarg);
            if (app.client.job_window <= 0) {
                PJ_LOG(3,(THIS_FILE, "Invalid --window %s", pj_optarg));
                return -1;
            }
            break;
 
        case 'T':
            app.use_tcp = PJ_TRUE;
            break;
 
        case 'd':
            app.server.delay = my_atoi(pj_optarg);
            if (app.server.delay > 3600) {
                PJ_LOG(3,(THIS_FILE, "I think --delay %s is too long", 
                          pj_optarg));
                return -1;
            }
            break;
 
        case OPT_TRYING:
            app.server.send_trying = 1;
            break;
 
        case OPT_RINGING:
            app.server.send_ringing = 1;
            break;
 
        default:
            PJ_LOG(1,(THIS_FILE, 
                      "Invalid argument. Use --help to see help"));
            return -1;
        }
    }
 
    if (pj_optind != argc) {
 
        if (verify_sip_url(argv[pj_optind]) != PJ_SUCCESS) {
            PJ_LOG(1,(THIS_FILE, "Invalid SIP URI %s", argv[pj_optind]));
            return -1;
        }
        app.client.dst_uri = pj_str(argv[pj_optind]);
        
        pj_optind++;
 
    }
 
    if (pj_optind != argc) {
        PJ_LOG(1,(THIS_FILE, "Error: unknown options %s", argv[pj_optind]));
        return -1;
    }
 
    return 0;
}
 
 
/* Send one stateless request */
static pj_status_t submit_stateless_job(void)
{
    pjsip_tx_data *tdata;
    pj_status_t status;
 
    status = pjsip_endpt_create_request(app.sip_endpt, &app.client.method, 
                                        &app.client.dst_uri, &app.local_uri,
                                        &app.client.dst_uri, &app.local_contact,
                                        NULL, -1, NULL, &tdata);
    if (status != PJ_SUCCESS) {
        app_perror(THIS_FILE, "Error creating request", status);
        report_completion(701);
        return status;
    }
 
    status = pjsip_endpt_send_request_stateless(app.sip_endpt, tdata, NULL,
                                                NULL);
    if (status != PJ_SUCCESS) {
        pjsip_tx_data_dec_ref(tdata);
        app_perror(THIS_FILE, "Error sending stateless request", status);
        report_completion(701);
        return status;
    }
 
    return PJ_SUCCESS;
}
 
 
/* This callback is called when client transaction state has changed */
static void tsx_completion_cb(void *token, pjsip_event *event)
{
    pjsip_transaction *tsx;
 
    PJ_UNUSED_ARG(token);
 
    if (event->type != PJSIP_EVENT_TSX_STATE)
        return;
 
    tsx = event->body.tsx_state.tsx;
 
    if (tsx->mod_data[mod_test.id] != NULL) {
        /* This transaction has been calculated before */
        return;
    }
 
    if (tsx->state==PJSIP_TSX_STATE_TERMINATED) {
        report_completion(tsx->status_code);
        tsx->mod_data[mod_test.id] = (void*)(pj_ssize_t)1;
    }
    else if (tsx->method.id == PJSIP_INVITE_METHOD &&
             tsx->state == PJSIP_TSX_STATE_CONFIRMED) {
 
        report_completion(tsx->status_code);
        tsx->mod_data[mod_test.id] = (void*)(pj_ssize_t)1;
        
    } else if (tsx->state == PJSIP_TSX_STATE_COMPLETED) {
 
        report_completion(tsx->status_code);
        tsx->mod_data[mod_test.id] = (void*)(pj_ssize_t)1;
 
        TERMINATE_TSX(tsx, tsx->status_code);
    }
}
 
 
/* Send one stateful request */
static pj_status_t submit_job(void)
{
    pjsip_tx_data *tdata;
    pj_status_t status;
 
    status = pjsip_endpt_create_request(app.sip_endpt, &app.client.method, 
                                        &app.client.dst_uri, &app.local_uri,
                                        &app.client.dst_uri, &app.local_contact,
                                        NULL, -1, NULL, &tdata);
    if (status != PJ_SUCCESS) {
        app_perror(THIS_FILE, "Error creating request", status);
        report_completion(701);
        return status;
    }
 
    status = pjsip_endpt_send_request(app.sip_endpt, tdata, -1, NULL, 
                                      &tsx_completion_cb);
    if (status != PJ_SUCCESS) {
        app_perror(THIS_FILE, "Error sending stateful request", status);
        //should have been reported by tsx_completion_cb().
        //report_completion(701);
        //No longer necessary (r777)
        //pjsip_tx_data_dec_ref(tdata);
    }
    return status;
}
 
 
/* Client worker thread */
static int client_thread(void *arg)
{
    pj_time_val end_time, last_report, now;
    unsigned thread_index = (unsigned)(long)(pj_ssize_t)arg;
    unsigned cycle = 0, last_cycle = 0;
 
    pj_thread_sleep(100);
 
    pj_gettimeofday(&end_time);
    end_time.sec += app.client.timeout;
 
    pj_gettimeofday(&last_report);
 
    if (app.client.first_request.sec == 0) {
        pj_gettimeofday(&app.client.first_request);
    }
 
    /* Submit all jobs */
    while (app.client.job_submitted < app.client.job_count && !app.thread_quit){
        pj_time_val timeout = { 0, 1 };
        unsigned i;
        int outstanding;
        pj_status_t status;
 
        /* Calculate current outstanding job */
        outstanding = app.client.job_submitted - app.client.job_finished;
 
        /* Update stats on max outstanding jobs */
        if (outstanding > (int)app.client.stat_max_window)
            app.client.stat_max_window = outstanding;
 
        /* Wait if there are more pending jobs than allowed in the
         * window. But spawn a new job anyway if no events are happening
         * after we wait for some time.
         */
        for (i=0; outstanding > (int)app.client.job_window && i<1000; ++i) {
            pj_time_val wait = { 0, 500 };
            unsigned count = 0;
 
            pjsip_endpt_handle_events2(app.sip_endpt, &wait, &count);
            outstanding = app.client.job_submitted - app.client.job_finished;
 
            if (count == 0)
                break;
 
            ++cycle;
        }
 
 
        /* Submit one job */
        if (app.client.method.id == PJSIP_INVITE_METHOD) {
            status = make_call(&app.client.dst_uri);
        } else if (app.client.stateless) {
            status = submit_stateless_job();
        } else {
            status = submit_job();
        }
        PJ_UNUSED_ARG(status);
 
        ++app.client.job_submitted;
        ++cycle;
 
        /* Handle event */
        pjsip_endpt_handle_events2(app.sip_endpt, &timeout, NULL);
 
        /* Check for time out, also print report */
        if (cycle - last_cycle >= 500) {
            pj_gettimeofday(&now);
            if (PJ_TIME_VAL_GTE(now, end_time)) {
                break;
            }
            last_cycle = cycle;
 
            
            if (thread_index == 0 && now.sec-last_report.sec >= 2) {
                printf("\r%d jobs started, %d completed...   ",
                       app.client.job_submitted, app.client.job_finished);
                fflush(stdout);
                last_report = now;
            }
        }
    }
 
    if (app.client.requests_sent.sec == 0) {
        pj_gettimeofday(&app.client.requests_sent);
    }
 
 
    if (thread_index == 0) {
        printf("\r%d jobs started, %d completed%s\n",
               app.client.job_submitted, app.client.job_finished,
               (app.client.job_submitted!=app.client.job_finished ? 
                ", waiting..." : ".") );
        fflush(stdout);
    }
 
    /* Wait until all jobs completes, or timed out */
    pj_gettimeofday(&now);
    while (PJ_TIME_VAL_LT(now, end_time) && 
           app.client.job_finished < app.client.job_count && 
           !app.thread_quit) 
    {
        pj_time_val timeout = { 0, 1 };
        unsigned i;
 
        for (i=0; i<1000; ++i) {
            unsigned count;
            count = 0;
            pjsip_endpt_handle_events2(app.sip_endpt, &timeout, &count);
            if (count == 0)
                break;
        }
 
        pj_gettimeofday(&now);
    }
 
    /* Wait couple of seconds to let jobs completes (e.g. ACKs to be sent)  */
    pj_gettimeofday(&now);
    end_time = now;
    end_time.sec += 2;
    while (PJ_TIME_VAL_LT(now, end_time)) 
    {
        pj_time_val timeout = { 0, 1 };
        unsigned i;
 
        for (i=0; i<1000; ++i) {
            unsigned count;
            count = 0;
            pjsip_endpt_handle_events2(app.sip_endpt, &timeout, &count);
            if (count == 0)
                break;
        }
 
        pj_gettimeofday(&now);
    }
 
    return 0;
}
 
 
static const char *good_number(char *buf, pj_int32_t val)
{
    if (val < 1000) {
        pj_ansi_sprintf(buf, "%d", val);
    } else if (val < 1000000) {
        pj_ansi_sprintf(buf, "%d.%dK", 
                        val / 1000,
                        (val % 1000) / 100);
    } else {
        pj_ansi_sprintf(buf, "%d.%02dM", 
                        val / 1000000,
                        (val % 1000000) / 10000);
    }
 
    return buf;
}
 
 
static int server_thread(void *arg)
{
    pj_time_val timeout = { 0, 1 };
    unsigned thread_index = (unsigned)(long)(pj_ssize_t)arg;
    pj_time_val last_report, next_report;
 
    pj_gettimeofday(&last_report);
    next_report = last_report;
    next_report.sec++;
 
    while (!app.thread_quit) {
        pj_time_val now;
        unsigned i;
 
        for (i=0; i<100; ++i) {
            unsigned count = 0;
            pjsip_endpt_handle_events2(app.sip_endpt, &timeout, &count);
            if (count == 0)
                break;
        }
 
        if (thread_index == 0) {
            pj_gettimeofday(&now);
 
            if (PJ_TIME_VAL_GTE(now, next_report)) {
                pj_time_val tmp;
                unsigned msec;
                unsigned stateless, stateful, call;
                char str_stateless[32], str_stateful[32], str_call[32];
 
                tmp = now;
                PJ_TIME_VAL_SUB(tmp, last_report);
                msec = PJ_TIME_VAL_MSEC(tmp);
 
                last_report = now;
                next_report = last_report;
                next_report.sec++;
 
                stateless = app.server.cur_state.stateless_cnt - app.server.prev_state.stateless_cnt;
                stateful = app.server.cur_state.stateful_cnt - app.server.prev_state.stateful_cnt;
                call = app.server.cur_state.call_cnt - app.server.prev_state.call_cnt;
 
                good_number(str_stateless, app.server.cur_state.stateless_cnt);
                good_number(str_stateful, app.server.cur_state.stateful_cnt);
                good_number(str_call, app.server.cur_state.call_cnt);
 
                printf("Total(rate): stateless:%s (%d/s), statefull:%s (%d/s), call:%s (%d/s)       \r",
                       str_stateless, stateless*1000/msec,
                       str_stateful, stateful*1000/msec,
                       str_call, call*1000/msec);
                fflush(stdout);
 
                app.server.prev_state = app.server.cur_state;
            }
        }
    }
 
    return 0;
}
 
static void write_report(const char *msg)
{
    puts(msg);
 
#if (defined(PJ_WIN32) && PJ_WIN32!=0) || (defined(PJ_WIN64) && PJ_WIN64!=0)
    OutputDebugString(msg);
    OutputDebugString("\n");
#endif
}
 
 
int main(int argc, char *argv[])
{
    static char report[1024];
 
    printf("PJSIP Performance Measurement Tool v%s\n"
           "(c)2006 pjsip.org\n\n",
           PJ_VERSION);
 
    if (create_app() != 0)
        return 1;
 
    if (init_options(argc, argv) != 0)
        return 1;
 
    if (init_sip() != 0)
        return 1;
 
    if (init_media() != 0)
        return 1;
 
    pj_log_set_level(app.log_level);
 
    if (app.log_level > 4) {
        pjsip_endpt_register_module(app.sip_endpt, &msg_logger);
    }
 
 
    /* Misc infos */
    if (app.client.dst_uri.slen != 0) {
        if (app.client.method.id == PJSIP_INVITE_METHOD) {
            if (app.client.stateless) {
                PJ_LOG(3,(THIS_FILE, 
                          "Info: --stateless option makes no sense for INVITE,"
                          " ignored."));
            }
        }
 
    }
 
 
 
    if (app.client.dst_uri.slen) {
        /* Client mode */
        pj_status_t status;
        char test_type[64];
        unsigned msec_req, msec_res;
        unsigned i;
 
        /* Get the job name */
        if (app.client.method.id == PJSIP_INVITE_METHOD) {
            pj_ansi_strcpy(test_type, "INVITE calls");
        } else if (app.client.stateless) {
            pj_ansi_sprintf(test_type, "stateless %.*s requests",
                            (int)app.client.method.name.slen,
                            app.client.method.name.ptr);
        } else {
            pj_ansi_sprintf(test_type, "stateful %.*s requests",
                            (int)app.client.method.name.slen,
                            app.client.method.name.ptr);
        }
        
 
        printf("Sending %d %s to '%.*s' with %d maximum outstanding jobs, please wait..\n", 
                  app.client.job_count, test_type,
                  (int)app.client.dst_uri.slen, app.client.dst_uri.ptr,
                  app.client.job_window);
 
        for (i=0; i<app.thread_count; ++i) {
            status = pj_thread_create(app.pool, NULL, &client_thread, 
                                      (void*)(pj_ssize_t)i, 0, 0, 
                                      &app.thread[i]);
            if (status != PJ_SUCCESS) {
                app_perror(THIS_FILE, "Unable to create thread", status);
                return 1;
            }
        }
 
        for (i=0; i<app.thread_count; ++i) {
            pj_thread_join(app.thread[i]);
            app.thread[i] = NULL;
        }
 
        if (app.client.last_completion.sec) {
            pj_time_val duration;
            duration = app.client.last_completion;
            PJ_TIME_VAL_SUB(duration, app.client.first_request);
            msec_res = PJ_TIME_VAL_MSEC(duration);
        } else {
            msec_res = app.client.timeout * 1000;
        }
 
        if (msec_res == 0) msec_res = 1;
 
        if (app.client.requests_sent.sec) {
            pj_time_val duration;
            duration = app.client.requests_sent;
            PJ_TIME_VAL_SUB(duration, app.client.first_request);
            msec_req = PJ_TIME_VAL_MSEC(duration);
        } else {
            msec_req = app.client.timeout * 1000;
        }
 
        if (msec_req == 0) msec_req = 1;
 
        if (app.client.job_submitted < app.client.job_count)
            puts("\ntimed-out!\n");
        else
            puts("\ndone.\n");
 
        pj_ansi_snprintf(
            report, sizeof(report),
            "Total %d %s sent in %d ms at rate of %d/sec\n"
            "Total %d responses received in %d ms at rate of %d/sec:",
            app.client.job_submitted, test_type, msec_req, 
            app.client.job_submitted * 1000 / msec_req,
            app.client.total_responses, msec_res,
            app.client.total_responses*1000/msec_res);
        write_report(report);
 
        /* Print detailed response code received */
        pj_ansi_sprintf(report, "\nDetailed responses received:");
        write_report(report);
 
        for (i=0; i<PJ_ARRAY_SIZE(app.client.response_codes); ++i) {
            const pj_str_t *reason;
 
            if (app.client.response_codes[i] == 0)
                continue;
 
            reason = pjsip_get_status_text(i);
            pj_ansi_snprintf( report, sizeof(report),
                              " - %d responses:  %7d     (%.*s)",
                              i, app.client.response_codes[i],
                              (int)reason->slen, reason->ptr);
            write_report(report);
        }
 
        /* Total responses and rate */
        pj_ansi_snprintf( report, sizeof(report),
            "                    ------\n"
            " TOTAL responses:  %7d (rate=%d/sec)\n",
            app.client.total_responses, 
            app.client.total_responses*1000/msec_res);
 
        write_report(report);
 
        pj_ansi_sprintf(report, "Maximum outstanding job: %d", 
                        app.client.stat_max_window);
        write_report(report);
 
 
    } else {
        /* Server mode */
        char s[10], *unused;
        pj_status_t status;
        unsigned i;
 
        puts("pjsip-perf started in server-mode");
 
        printf("Receiving requests on the following URIs:\n"
               "  sip:0@%.*s:%d%s    for stateless handling\n"
               "  sip:1@%.*s:%d%s    for stateful handling\n"
               "  sip:2@%.*s:%d%s    for call handling\n",
               (int)app.local_addr.slen,
               app.local_addr.ptr,
               app.local_port,
               (app.use_tcp ? ";transport=tcp" : ""),
               (int)app.local_addr.slen,
               app.local_addr.ptr,
               app.local_port,
               (app.use_tcp ? ";transport=tcp" : ""),
               (int)app.local_addr.slen,
               app.local_addr.ptr,
               app.local_port,
               (app.use_tcp ? ";transport=tcp" : ""));
        printf("INVITE with non-matching user part will be handled call-statefully\n");
 
        for (i=0; i<app.thread_count; ++i) {
            status = pj_thread_create(app.pool, NULL, &server_thread, 
                                      (void*)(pj_ssize_t)i, 0, 0, 
                                      &app.thread[i]);
            if (status != PJ_SUCCESS) {
                app_perror(THIS_FILE, "Unable to create thread", status);
                return 1;
            }
        }
 
        puts("\nPress <ENTER> to quit\n");
        fflush(stdout);
        unused = fgets(s, sizeof(s), stdin);
        PJ_UNUSED_ARG(unused);
 
        app.thread_quit = PJ_TRUE;
        for (i=0; i<app.thread_count; ++i) {
            pj_thread_join(app.thread[i]);
            app.thread[i] = NULL;
        }
 
        puts("");
    }
 
 
    destroy_app();
 
    return 0;
}