pjsip-perf.c

一个开源SIP协议栈

/* 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)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();
	}

	++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)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
    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*)(long)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 receieved 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];
	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*)(long)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);
	fgets(s, sizeof(s), stdin);

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