Index: pjproject/trunk/pjnath/src/pjnath-test/ice_test.c
===================================================================
--- pjproject/trunk/pjnath/src/pjnath-test/ice_test.c	(revision 3553)
+++ pjproject/trunk/pjnath/src/pjnath-test/ice_test.c	(revision 4360)
@@ -30,5 +30,7 @@
 #define NODELAY		0xFFFFFFFF
 #define SRV_DOMAIN	"pjsip.lab.domain"
-
+#define MAX_THREADS	16
+
+#define THIS_FILE	"ice_test.c"
 #define INDENT		"    "
 
@@ -49,5 +51,5 @@
 };
 
-
+/*  Role    comp#   host?   stun?   turn?   flag?  ans_del snd_del des_del */
 /* Test session configuration */
 struct test_cfg
@@ -61,6 +63,6 @@
 
     unsigned    answer_delay;	/* Delay before sending SDP	*/
-    unsigned    send_delay;	/* Delay before sending data	*/
-    unsigned    destroy_delay;	/* Delay before destroy()	*/
+    unsigned	send_delay;	/* unused */
+    unsigned	destroy_delay;	/* unused */
 
     struct test_result expected;/* Expected result		*/
@@ -80,4 +82,15 @@
 };
 
+/* Session param */
+struct sess_param
+{
+    unsigned		 worker_cnt;
+    unsigned		 worker_timeout;
+    pj_bool_t		 worker_quit;
+
+    pj_bool_t		 destroy_after_create;
+    pj_bool_t		 destroy_after_one_done;
+};
+
 /* The test session */
 struct test_sess
@@ -87,5 +100,9 @@
     pj_dns_resolver	*resolver;
 
+    struct sess_param	*param;
+
     test_server		*server;
+
+    pj_thread_t		*worker_threads[MAX_THREADS];
 
     unsigned		 server_flag;
@@ -191,4 +208,5 @@
 		       struct test_cfg *caller_cfg,
 		       struct test_cfg *callee_cfg,
+		       struct sess_param *test_param,
 		       struct test_sess **p_sess)
 {
@@ -205,4 +223,5 @@
     sess->pool = pool;
     sess->stun_cfg = stun_cfg;
+    sess->param = test_param;
 
     pj_memcpy(&sess->caller.cfg, caller_cfg, sizeof(*caller_cfg));
@@ -262,4 +281,6 @@
 static void destroy_sess(struct test_sess *sess, unsigned wait_msec)
 {
+    unsigned i;
+
     if (sess->caller.ice) {
 	pj_ice_strans_destroy(sess->caller.ice);
@@ -270,4 +291,10 @@
 	pj_ice_strans_destroy(sess->callee.ice);
 	sess->callee.ice = NULL;
+    }
+
+    sess->param->worker_quit = PJ_TRUE;
+    for (i=0; i<sess->param->worker_cnt; ++i) {
+	if (sess->worker_threads[i])
+	    pj_thread_join(sess->worker_threads[i]);
     }
 
@@ -327,4 +354,7 @@
 	ept->result.nego_status = status;
 	break;
+    case PJ_ICE_STRANS_OP_KEEP_ALIVE:
+	/* keep alive failed? */
+	break;
     default:
 	pj_assert(!"Unknown op");
@@ -385,5 +415,5 @@
 	c1 = pj_ice_strans_get_valid_pair(ept1->ice, i+1);
 	if (c1 == NULL) {
-	    PJ_LOG(3,("", INDENT "err: unable to get valid pair for ice1 "
+	    PJ_LOG(3,(THIS_FILE, INDENT "err: unable to get valid pair for ice1 "
 			  "component %d", i+1));
 	    return start_err - 2;
@@ -392,5 +422,5 @@
 	c2 = pj_ice_strans_get_valid_pair(ept2->ice, i+1);
 	if (c2 == NULL) {
-	    PJ_LOG(3,("", INDENT "err: unable to get valid pair for ice2 "
+	    PJ_LOG(3,(THIS_FILE, INDENT "err: unable to get valid pair for ice2 "
 			  "component %d", i+1));
 	    return start_err - 4;
@@ -398,5 +428,5 @@
 
 	if (pj_sockaddr_cmp(&c1->rcand->addr, &c2->lcand->addr) != 0) {
-	    PJ_LOG(3,("", INDENT "err: candidate pair does not match "
+	    PJ_LOG(3,(THIS_FILE, INDENT "err: candidate pair does not match "
 			  "for component %d", i+1));
 	    return start_err - 6;
@@ -409,5 +439,5 @@
 	    pj_ice_strans_get_valid_pair(ept1->ice, i+1) != NULL) 
 	{
-	    PJ_LOG(3,("", INDENT "err: ice1 shouldn't have valid pair "
+	    PJ_LOG(3,(THIS_FILE, INDENT "err: ice1 shouldn't have valid pair "
 		          "for component %d", i+1));
 	    return start_err - 8;
@@ -416,5 +446,5 @@
 	    pj_ice_strans_get_valid_pair(ept2->ice, i+1) != NULL) 
 	{
-	    PJ_LOG(3,("", INDENT "err: ice2 shouldn't have valid pair "
+	    PJ_LOG(3,(THIS_FILE, INDENT "err: ice2 shouldn't have valid pair "
 		          "for component %d", i+1));
 	    return start_err - 9;
@@ -437,24 +467,41 @@
 					break; \
 				    } \
-				    if (t.sec - t0.sec > (timeout)) break; \
+				    PJ_TIME_VAL_SUB(t, t0); \
+				    if (PJ_TIME_VAL_MSEC(t) >= (timeout)) break; \
 				} \
 			    }
 
-
-static int perform_test(const char *title,
-			pj_stun_config *stun_cfg,
-			unsigned server_flag,
-		        struct test_cfg *caller_cfg,
-		        struct test_cfg *callee_cfg)
+int worker_thread_proc(void *data)
+{
+    pj_status_t rc;
+    struct test_sess *sess = (struct test_sess *) data;
+    pj_stun_config *stun_cfg = sess->stun_cfg;
+    
+    /* Wait until negotiation is complete on both endpoints */
+#define ALL_DONE    (sess->param->worker_quit || \
+			(sess->caller.result.nego_status!=PJ_EPENDING && \
+			 sess->callee.result.nego_status!=PJ_EPENDING))
+    WAIT_UNTIL(sess->param->worker_timeout, ALL_DONE, rc);
+    
+    return 0;
+}
+
+static int perform_test2(const char *title,
+			 pj_stun_config *stun_cfg,
+                         unsigned server_flag,
+		         struct test_cfg *caller_cfg,
+		         struct test_cfg *callee_cfg,
+		         struct sess_param *test_param)
 {
     pjlib_state pjlib_state;
     struct test_sess *sess;
+    unsigned i;
     int rc;
 
-    PJ_LOG(3,("", INDENT "%s", title));
+    PJ_LOG(3,(THIS_FILE, INDENT "%s", title));
 
     capture_pjlib_state(stun_cfg, &pjlib_state);
 
-    rc = create_sess(stun_cfg, server_flag, caller_cfg, callee_cfg, &sess);
+    rc = create_sess(stun_cfg, server_flag, caller_cfg, callee_cfg, test_param, &sess);
     if (rc != 0)
 	return rc;
@@ -464,8 +511,8 @@
 
     /* Wait until both ICE transports are initialized */
-    WAIT_UNTIL(30, ALL_READY, rc);
+    WAIT_UNTIL(30000, ALL_READY, rc);
 
     if (!ALL_READY) {
-	PJ_LOG(3,("", INDENT "err: init timed-out"));
+	PJ_LOG(3,(THIS_FILE, INDENT "err: init timed-out"));
 	destroy_sess(sess, 500);
 	return -100;
@@ -490,5 +537,4 @@
 	goto on_return;
     }
-
     /* Init ICE on caller */
     rc = pj_ice_strans_init_ice(sess->caller.ice, sess->caller.cfg.role, 
@@ -508,5 +554,4 @@
 	return -110;
     }
-
     /* Start ICE on callee */
     rc = start_ice(&sess->callee, &sess->caller);
@@ -515,8 +560,6 @@
 	return -120;
     }
-
     /* Wait for callee's answer_delay */
     poll_events(stun_cfg, sess->callee.cfg.answer_delay, PJ_FALSE);
-
     /* Start ICE on caller */
     rc = start_ice(&sess->caller, &sess->callee);
@@ -526,11 +569,35 @@
     }
 
-    /* Wait until negotiation is complete on both endpoints */
-#define ALL_DONE    (sess->caller.result.nego_status!=PJ_EPENDING && \
-		     sess->callee.result.nego_status!=PJ_EPENDING)
-    WAIT_UNTIL(30, ALL_DONE, rc);
-
+    for (i=0; i<sess->param->worker_cnt; ++i) {
+	pj_status_t status;
+
+	status = pj_thread_create(sess->pool, "worker_thread",
+				  worker_thread_proc, sess, 0, 0,
+				  &sess->worker_threads[i]);
+	if (status != PJ_SUCCESS) {
+	    PJ_LOG(3,(THIS_FILE, INDENT "err: create thread"));
+	    destroy_sess(sess, 500);
+	    return -135;
+	}
+    }
+
+    if (sess->param->destroy_after_create)
+	goto on_destroy;
+
+    if (sess->param->destroy_after_one_done) {
+	while (sess->caller.result.init_status==PJ_EPENDING &&
+	       sess->callee.result.init_status==PJ_EPENDING)
+	{
+	    if (sess->param->worker_cnt)
+		pj_thread_sleep(0);
+	    else
+		poll_events(stun_cfg, 0, PJ_FALSE);
+	}
+	goto on_destroy;
+    }
+    
+    WAIT_UNTIL(30000, ALL_DONE, rc);
     if (!ALL_DONE) {
-	PJ_LOG(3,("", INDENT "err: negotiation timed-out"));
+	PJ_LOG(3,(THIS_FILE, INDENT "err: negotiation timed-out"));
 	destroy_sess(sess, 500);
 	return -140;
@@ -562,4 +629,5 @@
 
     /* Looks like everything is okay */
+on_destroy:
 
     /* Destroy ICE stream transports first to let it de-allocate
@@ -579,5 +647,5 @@
 on_return:
     /* Wait.. */
-    poll_events(stun_cfg, 500, PJ_FALSE);
+    poll_events(stun_cfg, 200, PJ_FALSE);
 
     /* Now destroy everything */
@@ -592,5 +660,18 @@
     }
 
-    return 0;
+    return rc;
+}
+
+static int perform_test(const char *title,
+                        pj_stun_config *stun_cfg,
+                        unsigned server_flag,
+                        struct test_cfg *caller_cfg,
+                        struct test_cfg *callee_cfg)
+{
+    struct sess_param test_param;
+
+    pj_bzero(&test_param, sizeof(test_param));
+    return perform_test2(title, stun_cfg, server_flag, caller_cfg,
+                         callee_cfg, &test_param);
 }
 
@@ -681,5 +762,5 @@
 	    goto on_return;
     }
-
+    
     /* Simple test first with srflx candidate */
     if (1) {
@@ -745,5 +826,5 @@
 	};
 
-	rc = perform_test(cfg.title, &stun_cfg, cfg.server_flag, 
+	rc = perform_test(cfg.title, &stun_cfg, cfg.server_flag,
 			  &cfg.ua1, &cfg.ua2);
 	if (rc != 0)
@@ -786,4 +867,5 @@
     }
 
+
     /* STUN failure, testing TURN deallocation */
     if (1) {
@@ -793,9 +875,9 @@
 	    0xFFFF & (~(CREATE_STUN_SERVER)),
 	    /*  Role    comp#   host?   stun?   turn?   flag?  ans_del snd_del des_del */
-	    {ROLE1,	2,	 YES,    YES,	YES,	0,    0,	    0,	    0, {PJNATH_ESTUNTIMEDOUT, -1}},
-	    {ROLE2,	2,	 YES,    YES,	YES,	0,    0,	    0,	    0, {PJNATH_ESTUNTIMEDOUT, -1}}
+	    {ROLE1,	1,	 YES,    YES,	YES,	0,    0,	    0,	    0, {PJNATH_ESTUNTIMEDOUT, -1}},
+	    {ROLE2,	1,	 YES,    YES,	YES,	0,    0,	    0,	    0, {PJNATH_ESTUNTIMEDOUT, -1}}
 	};
 
-	rc = perform_test(cfg.title, &stun_cfg, cfg.server_flag, 
+	rc = perform_test(cfg.title, &stun_cfg, cfg.server_flag,
 			  &cfg.ua1, &cfg.ua2);
 	if (rc != 0)
@@ -819,5 +901,5 @@
 	unsigned d;
 
-	PJ_LOG(3,("", "  %s", cfg->title));
+	PJ_LOG(3,(THIS_FILE, "  %s", cfg->title));
 
 	/* For each test item, test with various answer delay */
@@ -877,2 +959,85 @@
 }
 
+int ice_one_conc_test(pj_stun_config *stun_cfg, int err_quit)
+{
+    struct sess_cfg_t {
+	const char	*title;
+	unsigned	 server_flag;
+	struct test_cfg	 ua1;
+	struct test_cfg	 ua2;
+    } cfg =
+    {
+	"Concurrency test",
+	0xFFFF,
+	/*  Role    comp#   host?   stun?   turn?   flag?  ans_del snd_del des_del */
+	{ROLE1,	1,	YES,     YES,	    YES,    0,	    0,	    0,	    0, {PJ_SUCCESS, PJ_SUCCESS}},
+	{ROLE2,	1,	YES,     YES,	    YES,    0,	    0,	    0,	    0, {PJ_SUCCESS, PJ_SUCCESS}}
+    };
+    struct sess_param test_param;
+    int rc;
+
+
+    /* test a: destroy as soon as nego starts */
+    cfg.title = "    ice test a: immediate destroy";
+    pj_bzero(&test_param, sizeof(test_param));
+    test_param.worker_cnt = 4;
+    test_param.worker_timeout = 1000;
+    test_param.destroy_after_create = PJ_TRUE;
+
+    rc = perform_test2(cfg.title, stun_cfg, cfg.server_flag,
+                       &cfg.ua1, &cfg.ua2, &test_param);
+    if (rc != 0 && err_quit)
+        return rc;
+
+    /* test b: destroy as soon as one is done */
+    cfg.title = "    ice test b: destroy after 1 success";
+    test_param.destroy_after_create = PJ_FALSE;
+    test_param.destroy_after_one_done = PJ_TRUE;
+
+    rc = perform_test2(cfg.title, stun_cfg, cfg.server_flag,
+                       &cfg.ua1, &cfg.ua2, &test_param);
+    if (rc != 0 && err_quit)
+        return rc;
+
+    /* test c: normal */
+    cfg.title = "    ice test c: normal flow";
+    pj_bzero(&test_param, sizeof(test_param));
+    test_param.worker_cnt = 4;
+    test_param.worker_timeout = 1000;
+
+    rc = perform_test2(cfg.title, stun_cfg, cfg.server_flag,
+                       &cfg.ua1, &cfg.ua2, &test_param);
+    if (rc != 0 && err_quit)
+        return rc;
+
+    return 0;
+}
+
+int ice_conc_test(void)
+{
+    const int LOOP = 100;
+    pj_pool_t *pool;
+    pj_stun_config stun_cfg;
+    unsigned i;
+    int rc;
+    
+    pool = pj_pool_create(mem, NULL, 512, 512, NULL);
+    rc = create_stun_config(pool, &stun_cfg);
+    if (rc != PJ_SUCCESS) {
+	pj_pool_release(pool);
+	return -7;
+    }
+    
+    for (i = 0; i < LOOP; i++) {
+	PJ_LOG(3,(THIS_FILE, INDENT "Test %d of %d", i+1, LOOP));
+	rc = ice_one_conc_test(&stun_cfg, PJ_TRUE);
+	if (rc)
+	    break;
+    }
+    
+on_return:
+    destroy_stun_config(&stun_cfg);
+    pj_pool_release(pool);
+
+    return rc;
+}
