Changeset 4360

Timestamp:

Feb 21, 2013 11:26:35 AM (11 years ago)

Author:

bennylp

Message:

Fixed #1617: major synchronization fixes in PJNATH with incorporation of group lock to avoid deadlock and crashes due to race conditions

Location:

pjproject/trunk/pjnath

Files:

• build/Makefile (modified) (1 diff)
• include/pjnath/ice_session.h (modified) (3 diffs)
• include/pjnath/stun_session.h (modified) (4 diffs)
• include/pjnath/stun_sock.h (modified) (2 diffs)
• include/pjnath/stun_transaction.h (modified) (5 diffs)
• include/pjnath/turn_session.h (modified) (2 diffs)
• include/pjnath/turn_sock.h (modified) (1 diff)
• src/pjnath-test/concur_test.c (added)
• src/pjnath-test/ice_test.c (modified) (30 diffs)
• src/pjnath-test/sess_auth.c (modified) (2 diffs)
• src/pjnath-test/stun_sock_test.c (modified) (6 diffs)
• src/pjnath-test/test.c (modified) (7 diffs)
• src/pjnath-test/test.h (modified) (2 diffs)
• src/pjnath/ice_session.c (modified) (59 diffs)
• src/pjnath/ice_strans.c (modified) (26 diffs)
• src/pjnath/nat_detect.c (modified) (1 diff)
• src/pjnath/stun_session.c (modified) (41 diffs)
• src/pjnath/stun_sock.c (modified) (31 diffs)
• src/pjnath/stun_transaction.c (modified) (22 diffs)
• src/pjnath/turn_session.c (modified) (33 diffs)
• src/pjnath/turn_sock.c (modified) (26 diffs)
• src/pjturn-srv/allocation.c (modified) (1 diff)
• src/pjturn-srv/server.c (modified) (1 diff)

pjproject/trunk/pjnath/build/Makefile

@@ -41 +41 @@
 #
 export PJNATH_TEST_SRCDIR = ../src/pjnath-test
-export PJNATH_TEST_OBJS += ice_test.o stun.o sess_auth.o server.o \
+export PJNATH_TEST_OBJS += ice_test.o stun.o sess_auth.o server.o concur_test.o \
                             stun_sock_test.o turn_sock_test.o test.o
 export PJNATH_TEST_CFLAGS += $(_CFLAGS)

pjproject/trunk/pjnath/include/pjnath/ice_session.h

@@ -613 +613 @@
     pj_pool_t           *pool;                      /**< Pool instance.     */
     void                *user_data;                 /**< App. data.         */
-    pj_mutex_t          *mutex;                     /**< Mutex.             */
+    pj_grp_lock_t       *grp_lock;                  /**< Group lock         */
     pj_ice_sess_role     role;                      /**< ICE role.          */
     pj_ice_sess_options  opt;                       /**< Options            */
@@ -731 +731 @@
  *                      generated.
  * @param local_passwd  Optional string to be used as local password.
+ * @param grp_lock      Optional group lock to be used by this session.
+ *                      If NULL, the session will create one itself.
  * @param p_ice         Pointer to receive the ICE session instance.
  *
@@ -742 +744 @@
                                         const pj_str_t *local_ufrag,
                                         const pj_str_t *local_passwd,
+                                        pj_grp_lock_t *grp_lock,
                                         pj_ice_sess **p_ice);
 

pjproject/trunk/pjnath/include/pjnath/stun_session.h

@@ -31 +31 @@
 #include <pjnath/stun_transaction.h>
 #include <pj/list.h>
+#include <pj/lock.h>
 #include <pj/timer.h>
 
@@ -385 +386 @@
  * @param cb            Session callback.
  * @param fingerprint   Enable message fingerprint for outgoing messages.
+ * @param grp_lock      Optional group lock to be used by this session.
+ *                      If NULL, the session will create one itself.
  * @param p_sess        Pointer to receive STUN session instance.
  *
@@ -393 +396 @@
                                             const pj_stun_session_cb *cb,
                                             pj_bool_t fingerprint,
+                                            pj_grp_lock_t *grp_lock,
                                             pj_stun_session **p_sess);
 
@@ -430 +434 @@
  */
 PJ_DECL(void*) pj_stun_session_get_user_data(pj_stun_session *sess);
-
-/**
- * Change the lock object used by the STUN session. By default, the STUN
- * session uses a mutex to protect its internal data. If application already
- * protects access to STUN session with higher layer lock, it may disable
- * the mutex protection in the STUN session by changing the STUN session
- * lock to a NULL mutex.
- *
- * @param sess      The STUN session instance.
- * @param lock      New lock instance to be used by the STUN session.
- * @param auto_del  Specify whether STUN session should destroy this
- *                  lock instance when it's destroyed.
- */
-PJ_DECL(pj_status_t) pj_stun_session_set_lock(pj_stun_session *sess,
-                                              pj_lock_t *lock,
-                                              pj_bool_t auto_del);
 
 /**

pjproject/trunk/pjnath/include/pjnath/stun_sock.h

@@ -28 +28 @@
 #include <pjlib-util/resolver.h>
 #include <pj/ioqueue.h>
+#include <pj/lock.h>
 #include <pj/sock.h>
 #include <pj/sock_qos.h>
@@ -218 +219 @@
 typedef struct pj_stun_sock_cfg
 {
+    /**
+     * The group lock to be used by the STUN socket. If NULL, the STUN socket
+     * will create one internally.
+     *
+     * Default: NULL
+     */
+    pj_grp_lock_t *grp_lock;
+
     /**
      * Packet buffer size.

pjproject/trunk/pjnath/include/pjnath/stun_transaction.h

@@ -28 +28 @@
 #include <pjnath/stun_msg.h>
 #include <pjnath/stun_config.h>
+#include <pj/lock.h>
 
 
@@ -125 +126 @@
  *                      various settings for the transaction.
  * @param pool          Pool to be used to allocate memory from.
+ * @param grp_lock      Group lock to synchronize.
  * @param cb            Callback structure, to be used by the transaction
  *                      to send message and to notify the application about
@@ -134 +136 @@
 PJ_DECL(pj_status_t) pj_stun_client_tsx_create( pj_stun_config *cfg,
                                                 pj_pool_t *pool,
+                                                pj_grp_lock_t *grp_lock,
                                                 const pj_stun_tsx_cb *cb,
                                                 pj_stun_client_tsx **p_tsx);
@@ -160 +163 @@
 
 /**
- * Destroy a STUN client transaction immediately. This function can be 
- * called at any time to stop the transaction and destroy it.
+ * Stop the client transaction.
  *
  * @param tsx           The STUN transaction.
@@ -168 +170 @@
  *                      is NULL.
  */
-PJ_DECL(pj_status_t) pj_stun_client_tsx_destroy(pj_stun_client_tsx *tsx);
+PJ_DECL(pj_status_t) pj_stun_client_tsx_stop(pj_stun_client_tsx *tsx);
 
 

pjproject/trunk/pjnath/include/pjnath/turn_session.h

@@ -418 +418 @@
  * @param af            Address family of the client connection. Currently
  *                      pj_AF_INET() and pj_AF_INET6() are supported.
- * @param conn_type     Connection type to the TURN server. 
+ * @param conn_type     Connection type to the TURN server.
+ * @param grp_lock      Optional group lock object to be used by this session.
+ *                      If this value is NULL, the session will create
+ *                      a group lock internally.
  * @param cb            Callback to receive events from the TURN session.
  * @param options       Option flags, currently this value must be zero.
@@ -433 +436 @@
                                             int af,
                                             pj_turn_tp_type conn_type,
+                                            pj_grp_lock_t *grp_lock,
                                             const pj_turn_session_cb *cb,
                                             unsigned options,

pjproject/trunk/pjnath/include/pjnath/turn_sock.h

@@ -109 +109 @@
 typedef struct pj_turn_sock_cfg
 {
+    /**
+     * The group lock to be used by the STUN socket. If NULL, the STUN socket
+     * will create one internally.
+     *
+     * Default: NULL
+     */
+    pj_grp_lock_t *grp_lock;
+
     /**
      * Packet buffer size.

pjproject/trunk/pjnath/src/pjnath-test/ice_test.c

@@ -30 +30 @@
 #define NODELAY         0xFFFFFFFF
 #define SRV_DOMAIN      "pjsip.lab.domain"
-
+#define MAX_THREADS     16
+
+#define THIS_FILE       "ice_test.c"
 #define INDENT          "    "
 
@@ -49 +51 @@
 };
 
-
+/*  Role    comp#   host?   stun?   turn?   flag?  ans_del snd_del des_del */
 /* Test session configuration */
 struct test_cfg
@@ -61 +63 @@
 
     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 +82 @@
 };
 
+/* 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 +100 @@
     pj_dns_resolver     *resolver;
 
+    struct sess_param   *param;
+
     test_server         *server;
+
+    pj_thread_t         *worker_threads[MAX_THREADS];
 
     unsigned             server_flag;
@@ -191 +208 @@
                        struct test_cfg *caller_cfg,
                        struct test_cfg *callee_cfg,
+                       struct sess_param *test_param,
                        struct test_sess **p_sess)
 {
@@ -205 +223 @@
     sess->pool = pool;
     sess->stun_cfg = stun_cfg;
+    sess->param = test_param;
 
     pj_memcpy(&sess->caller.cfg, caller_cfg, sizeof(*caller_cfg));
@@ -262 +281 @@
 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 +291 @@
         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 +354 @@
         ept->result.nego_status = status;
         break;
+    case PJ_ICE_STRANS_OP_KEEP_ALIVE:
+        /* keep alive failed? */
+        break;
     default:
         pj_assert(!"Unknown op");
@@ -385 +415 @@
         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 +422 @@
         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 +428 @@
 
         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 +439 @@
             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 +446 @@
             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 +467 @@
                                         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 +511 @@
 
     /* 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 +537 @@
         goto on_return;
     }
-
     /* Init ICE on caller */
     rc = pj_ice_strans_init_ice(sess->caller.ice, sess->caller.cfg.role, 
@@ -508 +554 @@
         return -110;
     }
-
     /* Start ICE on callee */
     rc = start_ice(&sess->callee, &sess->caller);
@@ -515 +560 @@
         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 +569 @@
     }
 
-    /* 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 +629 @@
 
     /* Looks like everything is okay */
+on_destroy:
 
     /* Destroy ICE stream transports first to let it de-allocate
@@ -579 +647 @@
 on_return:
     /* Wait.. */
-    poll_events(stun_cfg, 500, PJ_FALSE);
+    poll_events(stun_cfg, 200, PJ_FALSE);
 
     /* Now destroy everything */
@@ -592 +660 @@
     }
 
-    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 +762 @@
             goto on_return;
     }
-
+    
     /* Simple test first with srflx candidate */
     if (1) {
@@ -745 +826 @@
         };
 
-        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 +867 @@
     }
 
+
     /* STUN failure, testing TURN deallocation */
     if (1) {
@@ -793 +875 @@
             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 +901 @@
         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 +959 @@
 }
 
+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;
+}

pjproject/trunk/pjnath/src/pjnath-test/sess_auth.c

@@ -248 +248 @@
     sess_cb.on_rx_request = &server_on_rx_request;
     sess_cb.on_send_msg = &server_send_msg;
-    status = pj_stun_session_create(&stun_cfg, "server", &sess_cb, PJ_FALSE, &server->sess);
+    status = pj_stun_session_create(&stun_cfg, "server", &sess_cb, PJ_FALSE, NULL, &server->sess);
     if (status != PJ_SUCCESS) {
         destroy_server();
@@ -480 +480 @@
     sess_cb.on_request_complete = &client_on_request_complete;
     sess_cb.on_send_msg = &client_send_msg;
-    status = pj_stun_session_create(&stun_cfg, "client", &sess_cb, PJ_FALSE, &client->sess);
+    status = pj_stun_session_create(&stun_cfg, "client", &sess_cb, PJ_FALSE, NULL, &client->sess);
     if (status != PJ_SUCCESS) {
         destroy_client_server();

pjproject/trunk/pjnath/src/pjnath-test/stun_sock_test.c

@@ -299 +299 @@
     pj_str_t srv_addr;
     pj_time_val timeout, t;
-    int ret = 0;
+    int i, ret = 0;
     pj_status_t status;
 
@@ -360 +360 @@
     destroy_server(srv);
     destroy_client(client);
+    for (i=0; i<7; ++i)
+        handle_events(cfg, 50);
     return ret;
 }
@@ -374 +376 @@
     pj_str_t srv_addr;
     pj_time_val timeout, t;
-    int ret = 0;
+    int i, ret = 0;
     pj_status_t status;
 
@@ -427 +429 @@
     destroy_server(srv);
     destroy_client(client);
+    for (i=0; i<7; ++i)
+        handle_events(cfg, 50);
     return ret;
 }
@@ -441 +445 @@
     pj_str_t srv_addr;
     pj_time_val timeout, t;
-    int ret = 0;
+    int i, ret = 0;
     pj_status_t status;
 
@@ -792 +796 @@
     destroy_server(srv);
     destroy_client(client);
+    for (i=0; i<7; ++i)
+        handle_events(cfg, 50);
     return ret;
 }

pjproject/trunk/pjnath/src/pjnath-test/test.c

@@ -35 +35 @@
     pj_ioqueue_t *ioqueue;
     pj_timer_heap_t *timer_heap;
+    pj_lock_t *lock;
     pj_status_t status;
 
@@ -50 +51 @@
     }
 
+    pj_lock_create_recursive_mutex(pool, NULL, &lock);
+    pj_timer_heap_set_lock(timer_heap, lock, PJ_TRUE);
+
     pj_stun_config_init(stun_cfg, mem, 0, ioqueue, timer_heap);
 
@@ -106 +110 @@
     st->timer_cnt = pj_timer_heap_count(cfg->timer_heap);
     
-    cp = (pj_caching_pool*)mem;
+    cp = (pj_caching_pool*)cfg->pf;
     st->pool_used_cnt = cp->used_count;
 }
@@ -121 +125 @@
         PJ_LOG(3,("", "    error: possibly leaking timer"));
         rc |= ERR_TIMER_LEAK;
+
+#if PJ_TIMER_DEBUG
+        pj_timer_heap_dump(cfg->timer_heap);
+#endif
     }
 
@@ -149 +157 @@
                       PJ_LOG_HAS_MICRO_SEC;
 
+pj_log_func *orig_log_func;
+FILE *log_file;
+
+static void test_log_func(int level, const char *data, int len)
+{
+    if (log_file) {
+        fwrite(data, len, 1, log_file);
+    }
+    if (level <= 3)
+        orig_log_func(level, data, len);
+}
+
 static int test_inner(void)
 {
@@ -159 +179 @@
     pj_log_set_level(3);
     pj_log_set_decor(param_log_decor);
+#elif 1
+    log_file = fopen("pjnath-test.log", "wt");
+    pj_log_set_level(5);
+    orig_log_func = pj_log_get_log_func();
+    pj_log_set_log_func(&test_log_func);
 #endif
 
@@ -190 +215 @@
 #endif
 
+#if INCLUDE_CONCUR_TEST
+    DO_TEST(concur_test());
+#endif
+
 on_return:
+    if (log_file)
+        fclose(log_file);
     return rc;
 }

pjproject/trunk/pjnath/src/pjnath-test/test.h

@@ -26 +26 @@
 #define INCLUDE_STUN_SOCK_TEST      1
 #define INCLUDE_TURN_SOCK_TEST      1
+#define INCLUDE_CONCUR_TEST         1
 
 int stun_test(void);
@@ -32 +33 @@
 int turn_sock_test(void);
 int ice_test(void);
+int concur_test(void);
 int test_main(void);
 
 extern void app_perror(const char *title, pj_status_t rc);
 extern pj_pool_factory *mem;
+
+int ice_one_conc_test(pj_stun_config *stun_cfg, int err_quit);
 
 ////////////////////////////////////

pjproject/trunk/pjnath/src/pjnath/ice_session.c

@@ -98 +98 @@
 };
 
+#define THIS_FILE               "ice_session.c"
 #define CHECK_NAME_LEN          128
 #define LOG4(expr)              PJ_LOG(4,expr)
@@ -135 +136 @@
 static void on_ice_complete(pj_ice_sess *ice, pj_status_t status);
 static void ice_keep_alive(pj_ice_sess *ice, pj_bool_t send_now);
+static void ice_on_destroy(void *obj);
 static void destroy_ice(pj_ice_sess *ice,
                         pj_status_t reason);
@@ -289 +291 @@
     status = pj_stun_session_create(&ice->stun_cfg, NULL, 
                                     &sess_cb, PJ_TRUE,
+                                    ice->grp_lock,
                                     &comp->stun_sess);
     if (status != PJ_SUCCESS)
@@ -333 +336 @@
                                        const pj_str_t *local_ufrag,
                                        const pj_str_t *local_passwd,
+                                       pj_grp_lock_t *grp_lock,
                                        pj_ice_sess **p_ice)
 {
@@ -360 +364 @@
                      name, ice);
 
-    status = pj_mutex_create_recursive(pool, ice->obj_name, 
-                                       &ice->mutex);
-    if (status != PJ_SUCCESS) {
-        destroy_ice(ice, status);
-        return status;
-    }
+    if (grp_lock) {
+        ice->grp_lock = grp_lock;
+    } else {
+        status = pj_grp_lock_create(pool, NULL, &ice->grp_lock);
+        if (status != PJ_SUCCESS) {
+            pj_pool_release(pool);
+            return status;
+        }
+    }
+
+    pj_grp_lock_add_ref(ice->grp_lock);
+    pj_grp_lock_add_handler(ice->grp_lock, pool, ice,
+                            &ice_on_destroy);
 
     pj_memcpy(&ice->cb, cb, sizeof(*cb));
@@ -445 +456 @@
 
 /*
+ * Callback to really destroy the session
+ */
+static void ice_on_destroy(void *obj)
+{
+    pj_ice_sess *ice = (pj_ice_sess*) obj;
+
+    if (ice->pool) {
+        pj_pool_t *pool = ice->pool;
+        ice->pool = NULL;
+        pj_pool_release(pool);
+    }
+    LOG4((THIS_FILE, "ICE session %p destroyed", ice));
+}
+
+/*
  * Destroy
  */
@@ -453 +479 @@
 
     if (reason == PJ_SUCCESS) {
-        LOG4((ice->obj_name, "Destroying ICE session"));
+        LOG4((ice->obj_name, "Destroying ICE session %p", ice));
+    }
+
+    pj_grp_lock_acquire(ice->grp_lock);
+
+    if (ice->is_destroying) {
+        pj_grp_lock_release(ice->grp_lock);
+        return;
     }
 
     ice->is_destroying = PJ_TRUE;
 
-    /* Let other callbacks finish */
-    if (ice->mutex) {
-        pj_mutex_lock(ice->mutex);
-        pj_mutex_unlock(ice->mutex);
-    }
-
-    if (ice->timer.id) {
-        pj_timer_heap_cancel(ice->stun_cfg.timer_heap, 
-                             &ice->timer);
-        ice->timer.id = PJ_FALSE;
-    }
+    pj_timer_heap_cancel_if_active(ice->stun_cfg.timer_heap,
+                                   &ice->timer, PJ_FALSE);
 
     for (i=0; i<ice->comp_cnt; ++i) {
@@ -477 +501 @@
     }
 
-    if (ice->clist.timer.id) {
-        pj_timer_heap_cancel(ice->stun_cfg.timer_heap, &ice->clist.timer);
-        ice->clist.timer.id = PJ_FALSE;
-    }
-
-    if (ice->mutex) {
-        pj_mutex_destroy(ice->mutex);
-        ice->mutex = NULL;
-    }
-
-    if (ice->pool) {
-        pj_pool_t *pool = ice->pool;
-        ice->pool = NULL;
-        pj_pool_release(pool);
-    }
+    pj_timer_heap_cancel_if_active(ice->stun_cfg.timer_heap,
+                                   &ice->clist.timer,
+                                   PJ_FALSE);
+
+    pj_grp_lock_dec_ref(ice->grp_lock);
+    pj_grp_lock_release(ice->grp_lock);
 }
 
@@ -710 +725 @@
     PJ_ASSERT_RETURN(comp_id <= ice->comp_cnt, PJ_EINVAL);
 
-    pj_mutex_lock(ice->mutex);
+    pj_grp_lock_acquire(ice->grp_lock);
 
     if (ice->lcand_cnt >= PJ_ARRAY_SIZE(ice->lcand)) {
@@ -750 +765 @@
 
 on_error:
-    pj_mutex_unlock(ice->mutex);
+    pj_grp_lock_release(ice->grp_lock);
     return status;
 }
@@ -767 +782 @@
     *cand_id = -1;
 
-    pj_mutex_lock(ice->mutex);
+    pj_grp_lock_acquire(ice->grp_lock);
 
     /* First find in valid list if we have nominated pair */
@@ -775 +790 @@
         if (check->lcand->comp_id == comp_id) {
             *cand_id = GET_LCAND_ID(check->lcand);
-            pj_mutex_unlock(ice->mutex);
+            pj_grp_lock_release(ice->grp_lock);
             return PJ_SUCCESS;
         }
@@ -787 +802 @@
         {
             *cand_id = GET_LCAND_ID(lcand);
-            pj_mutex_unlock(ice->mutex);
+            pj_grp_lock_release(ice->grp_lock);
             return PJ_SUCCESS;
         }
@@ -800 +815 @@
         {
             *cand_id = GET_LCAND_ID(lcand);
-            pj_mutex_unlock(ice->mutex);
+            pj_grp_lock_release(ice->grp_lock);
             return PJ_SUCCESS;
         }
@@ -812 +827 @@
         {
             *cand_id = GET_LCAND_ID(lcand);
-            pj_mutex_unlock(ice->mutex);
+            pj_grp_lock_release(ice->grp_lock);
             return PJ_SUCCESS;
         }
@@ -818 +833 @@
 
     /* Still no candidate is found! :( */
-    pj_mutex_unlock(ice->mutex);
+    pj_grp_lock_release(ice->grp_lock);
 
     pj_assert(!"Should have a candidate by now");
@@ -1128 +1143 @@
     pj_ice_sess *ice = (pj_ice_sess*) te->user_data;
     enum timer_type type = (enum timer_type)te->id;
-    pj_bool_t has_mutex = PJ_TRUE;
 
     PJ_UNUSED_ARG(th);
 
-    pj_mutex_lock(ice->mutex);
+    pj_grp_lock_acquire(ice->grp_lock);
 
     te->id = TIMER_NONE;
+
+    if (ice->is_destroying) {
+        /* Stray timer, could happen when destroy is invoked while callback
+         * is pending. */
+        pj_grp_lock_release(ice->grp_lock);
+        return;
+    }
 
     switch (type) {
@@ -1158 +1179 @@
             ice_status = ice->ice_status;
             on_ice_complete = ice->cb.on_ice_complete;
-            has_mutex = PJ_FALSE;
-            pj_mutex_unlock(ice->mutex);
 
             /* Notify app about ICE completion*/
@@ -1177 +1196 @@
     }
 
-    if (has_mutex)
-        pj_mutex_unlock(ice->mutex);
+    pj_grp_lock_release(ice->grp_lock);
 }
 
@@ -1236 +1254 @@
         pj_time_val_normalize(&delay);
 
-        ice->timer.id = TIMER_KEEP_ALIVE;
-        pj_timer_heap_schedule(ice->stun_cfg.timer_heap, &ice->timer, &delay);
+        pj_timer_heap_schedule_w_grp_lock(ice->stun_cfg.timer_heap,
+                                          &ice->timer, &delay,
+                                          TIMER_KEEP_ALIVE,
+                                          ice->grp_lock);
 
     } else {
@@ -1251 +1271 @@
         ice->ice_status = status;
     
-        if (ice->timer.id != TIMER_NONE) {
-            pj_timer_heap_cancel(ice->stun_cfg.timer_heap, &ice->timer);
-            ice->timer.id = TIMER_NONE;
-        }
+        pj_timer_heap_cancel_if_active(ice->stun_cfg.timer_heap, &ice->timer,
+                                       TIMER_NONE);
 
         /* Log message */
@@ -1267 +1285 @@
             pj_time_val delay = {0, 0};
 
-            ice->timer.id = TIMER_COMPLETION_CALLBACK;
-            pj_timer_heap_schedule(ice->stun_cfg.timer_heap, 
-                                   &ice->timer, &delay);
+            pj_timer_heap_schedule_w_grp_lock(ice->stun_cfg.timer_heap,
+                                              &ice->timer, &delay,
+                                              TIMER_COMPLETION_CALLBACK,
+                                              ice->grp_lock);
         }
     }
@@ -1497 +1516 @@
                     pj_time_val_normalize(&delay);
 
-                    ice->timer.id = TIMER_CONTROLLED_WAIT_NOM;
-                    pj_timer_heap_schedule(ice->stun_cfg.timer_heap, 
-                                           &ice->timer,
-                                           &delay);
+                    pj_timer_heap_schedule_w_grp_lock(
+                                        ice->stun_cfg.timer_heap,
+                                        &ice->timer, &delay,
+                                        TIMER_CONTROLLED_WAIT_NOM,
+                                        ice->grp_lock);
 
                     LOG5((ice->obj_name, 
@@ -1576 +1596 @@
               ice->opt.nominated_check_delay));
 
-        if (ice->timer.id != TIMER_NONE) {
-            pj_timer_heap_cancel(ice->stun_cfg.timer_heap, &ice->timer);
-            ice->timer.id = TIMER_NONE;
-        }
+        pj_timer_heap_cancel_if_active(ice->stun_cfg.timer_heap, &ice->timer,
+                                       TIMER_NONE);
 
         /* All components have valid pair. Let connectivity checks run for
@@ -1588 +1606 @@
         pj_time_val_normalize(&delay);
 
-        ice->timer.id = TIMER_START_NOMINATED_CHECK;
-        pj_timer_heap_schedule(ice->stun_cfg.timer_heap, &ice->timer, &delay);
+        pj_timer_heap_schedule_w_grp_lock(ice->stun_cfg.timer_heap,
+                                          &ice->timer, &delay,
+                                          TIMER_START_NOMINATED_CHECK,
+                                          ice->grp_lock);
         return PJ_FALSE;
     }
@@ -1619 +1639 @@
                      PJ_ETOOMANY);
 
-    pj_mutex_lock(ice->mutex);
+    pj_grp_lock_acquire(ice->grp_lock);
 
     /* Save credentials */
@@ -1667 +1687 @@
 
             if (clist->count >= PJ_ICE_MAX_CHECKS) {
-                pj_mutex_unlock(ice->mutex);
+                pj_grp_lock_release(ice->grp_lock);
                 return PJ_ETOOMANY;
             } 
@@ -1695 +1715 @@
     if (clist->count == 0) {
         LOG4((ice->obj_name,  "Error: no checklist can be created"));
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
         return PJ_ENOTFOUND;
     }
@@ -1705 +1725 @@
     status = prune_checklist(ice, clist);
     if (status != PJ_SUCCESS) {
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
         return status;
     }
@@ -1732 +1752 @@
     dump_checklist("Checklist created:", ice, clist);
 
-    pj_mutex_unlock(ice->mutex);
+    pj_grp_lock_release(ice->grp_lock);
 
     return PJ_SUCCESS;
@@ -1851 +1871 @@
     clist = td->clist;
 
-    if (ice->is_destroying)
-        return PJ_SUCCESS;
-
-    pj_mutex_lock(ice->mutex);
+    pj_grp_lock_acquire(ice->grp_lock);
 
     if (ice->is_destroying) {
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
         return PJ_SUCCESS;
     }
@@ -1879 +1896 @@
             status = perform_check(ice, clist, i, ice->is_nominating);
             if (status != PJ_SUCCESS) {
-                pj_mutex_unlock(ice->mutex);
+                pj_grp_lock_release(ice->grp_lock);
                 pj_log_pop_indent();
                 return status;
@@ -1899 +1916 @@
                 status = perform_check(ice, clist, i, ice->is_nominating);
                 if (status != PJ_SUCCESS) {
-                    pj_mutex_unlock(ice->mutex);
+                    pj_grp_lock_release(ice->grp_lock);
                     pj_log_pop_indent();
                     return status;
@@ -1916 +1933 @@
         pj_time_val timeout = {0, PJ_ICE_TA_VAL};
 
-        te->id = PJ_TRUE;
         pj_time_val_normalize(&timeout);
-        pj_timer_heap_schedule(th, te, &timeout);
-    }
-
-    pj_mutex_unlock(ice->mutex);
+        pj_timer_heap_schedule_w_grp_lock(th, te, &timeout, PJ_TRUE,
+                                          ice->grp_lock);
+    }
+
+    pj_grp_lock_release(ice->grp_lock);
     pj_log_pop_indent();
     return PJ_SUCCESS;
@@ -1941 +1958 @@
     /* Stop our timer if it's active */
     if (ice->timer.id == TIMER_START_NOMINATED_CHECK) {
-        pj_timer_heap_cancel(ice->stun_cfg.timer_heap, &ice->timer);
-        ice->timer.id = TIMER_NONE;
+        pj_timer_heap_cancel_if_active(ice->stun_cfg.timer_heap, &ice->timer,
+                                       TIMER_NONE);
     }
 
@@ -1970 +1987 @@
 
     /* And (re)start the periodic check */
-    if (ice->clist.timer.id) {
-        pj_timer_heap_cancel(ice->stun_cfg.timer_heap, &ice->clist.timer);
-        ice->clist.timer.id = PJ_FALSE;
-    }
-
-    ice->clist.timer.id = PJ_TRUE;
+    pj_timer_heap_cancel_if_active(ice->stun_cfg.timer_heap,
+                                   &ice->clist.timer, PJ_FALSE);
+
     delay.sec = delay.msec = 0;
-    status = pj_timer_heap_schedule(ice->stun_cfg.timer_heap, 
-                                    &ice->clist.timer, &delay);
-    if (status != PJ_SUCCESS) {
-        ice->clist.timer.id = PJ_FALSE;
-    } else {
+    status = pj_timer_heap_schedule_w_grp_lock(ice->stun_cfg.timer_heap,
+                                               &ice->clist.timer, &delay,
+                                               PJ_TRUE,
+                                               ice->grp_lock);
+    if (status == PJ_SUCCESS) {
         LOG5((ice->obj_name, "Periodic timer rescheduled.."));
     }
@@ -2031 +2045 @@
 
     /* Lock session */
-    pj_mutex_lock(ice->mutex);
+    pj_grp_lock_acquire(ice->grp_lock);
 
     LOG4((ice->obj_name, "Starting ICE check.."));
@@ -2061 +2075 @@
     if (i == clist->count) {
         pj_assert(!"Unable to find checklist for component 1");
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
         pj_log_pop_indent();
         return PJNATH_EICEINCOMPID;
@@ -2115 +2129 @@
      * return start_periodic_check(ice->stun_cfg.timer_heap, &clist->timer);
      */
-    clist->timer.id = PJ_TRUE;
     delay.sec = delay.msec = 0;
-    status = pj_timer_heap_schedule(ice->stun_cfg.timer_heap, 
-                                    &clist->timer, &delay);
+    status = pj_timer_heap_schedule_w_grp_lock(ice->stun_cfg.timer_heap,
+                                               &clist->timer, &delay,
+                                               PJ_TRUE, ice->grp_lock);
     if (status != PJ_SUCCESS) {
         clist->timer.id = PJ_FALSE;
     }
 
-    pj_mutex_unlock(ice->mutex);
+    pj_grp_lock_release(ice->grp_lock);
     pj_log_pop_indent();
     return status;
@@ -2144 +2158 @@
     pj_ice_sess *ice = sd->ice;
     pj_ice_msg_data *msg_data = (pj_ice_msg_data*) token;
+    pj_status_t status;
     
-    return (*ice->cb.on_tx_pkt)(ice, sd->comp_id, msg_data->transport_id,
-                                pkt, pkt_size, dst_addr, addr_len);
+    pj_grp_lock_acquire(ice->grp_lock);
+
+    if (ice->is_destroying) {
+        /* Stray retransmit timer that could happen while
+         * we're being destroyed */
+        pj_grp_lock_release(ice->grp_lock);
+        return PJ_EINVALIDOP;
+    }
+
+    status = (*ice->cb.on_tx_pkt)(ice, sd->comp_id, msg_data->transport_id,
+                                  pkt, pkt_size, dst_addr, addr_len);
+
+    pj_grp_lock_release(ice->grp_lock);
+    return status;
 }
 
@@ -2181 +2208 @@
     check->tdata = NULL;
 
-    pj_mutex_lock(ice->mutex);
+    pj_grp_lock_acquire(ice->grp_lock);
+
+    if (ice->is_destroying) {
+        /* Not sure if this is possible but just in case */
+        pj_grp_lock_release(ice->grp_lock);
+        return;
+    }
 
     /* Init lcand to NULL. lcand will be found from the mapped address
@@ -2232 +2265 @@
                           check->nominated || ice->is_nominating);
             pj_log_pop_indent();
-            pj_mutex_unlock(ice->mutex);
+            pj_grp_lock_release(ice->grp_lock);
             return;
         }
@@ -2247 +2280 @@
         on_check_complete(ice, check);
         pj_log_pop_indent();
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
         return;
     }
@@ -2271 +2304 @@
         on_check_complete(ice, check);
         pj_log_pop_indent();
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
         return;
     }
@@ -2304 +2337 @@
                         PJNATH_ESTUNNOMAPPEDADDR);
         on_check_complete(ice, check);
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
         return;
     }
@@ -2352 +2385 @@
                             status);
             on_check_complete(ice, check);
-            pj_mutex_unlock(ice->mutex);
+            pj_grp_lock_release(ice->grp_lock);
             return;
         }
@@ -2412 +2445 @@
     if (on_check_complete(ice, check)) {
         /* ICE complete! */
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
         return;
     }
 
-    pj_mutex_unlock(ice->mutex);
+    pj_grp_lock_release(ice->grp_lock);
 }
 
@@ -2457 +2490 @@
     ice = sd->ice;
 
-    pj_mutex_lock(ice->mutex);
+    pj_grp_lock_acquire(ice->grp_lock);
+
+    if (ice->is_destroying) {
+        pj_grp_lock_release(ice->grp_lock);
+        return PJ_EINVALIDOP;
+    }
 
     /*
@@ -2472 +2510 @@
     if (prio_attr == NULL) {
         LOG5((ice->obj_name, "Received Binding request with no PRIORITY"));
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
         return PJ_SUCCESS;
     }
@@ -2517 +2555 @@
                                     NULL, token, PJ_TRUE, 
                                     src_addr, src_addr_len);
-            pj_mutex_unlock(ice->mutex);
+            pj_grp_lock_release(ice->grp_lock);
             return PJ_SUCCESS;
         }
@@ -2529 +2567 @@
                                     NULL, token, PJ_TRUE, 
                                     src_addr, src_addr_len);
-            pj_mutex_unlock(ice->mutex);
+            pj_grp_lock_release(ice->grp_lock);
             return PJ_SUCCESS;
         } else {
@@ -2544 +2582 @@
     status = pj_stun_session_create_res(sess, rdata, 0, NULL, &tdata);
     if (status != PJ_SUCCESS) {
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
         return status;
     }
@@ -2596 +2634 @@
     }
 
-    pj_mutex_unlock(ice->mutex);
+    pj_grp_lock_release(ice->grp_lock);
     return PJ_SUCCESS;
 }
@@ -2885 +2923 @@
     }
 
-    pj_mutex_lock(ice->mutex);
+    pj_grp_lock_acquire(ice->grp_lock);
+
+    if (ice->is_destroying) {
+        pj_grp_lock_release(ice->grp_lock);
+        return PJ_EINVALIDOP;
+    }
 
     comp = find_comp(ice, comp_id);
     if (comp == NULL) {
         status = PJNATH_EICEINCOMPID;
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
         goto on_return;
     }
@@ -2896 +2939 @@
     if (comp->valid_check == NULL) {
         status = PJNATH_EICEINPROGRESS;
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
         goto on_return;
     }
@@ -2905 +2948 @@
 
     /* Release the mutex now to avoid deadlock (see ticket #1451). */
-    pj_mutex_unlock(ice->mutex);
+    pj_grp_lock_release(ice->grp_lock);
+
+    PJ_RACE_ME(5);
 
     status = (*ice->cb.on_tx_pkt)(ice, comp_id, transport_id, 
@@ -2932 +2977 @@
     PJ_ASSERT_RETURN(ice, PJ_EINVAL);
 
-    pj_mutex_lock(ice->mutex);
+    pj_grp_lock_acquire(ice->grp_lock);
+
+    if (ice->is_destroying) {
+        pj_grp_lock_release(ice->grp_lock);
+        return PJ_EINVALIDOP;
+    }
 
     comp = find_comp(ice, comp_id);
     if (comp == NULL) {
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
         return PJNATH_EICEINCOMPID;
     }
@@ -2949 +2999 @@
     if (msg_data == NULL) {
         pj_assert(!"Invalid transport ID");
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
         return PJ_EINVAL;
     }
@@ -2969 +3019 @@
                   ice->tmp.errmsg));
         }
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
     } else {
         /* Not a STUN packet. Call application's callback instead, but release
          * the mutex now or otherwise we may get deadlock.
          */
-        pj_mutex_unlock(ice->mutex);
+        pj_grp_lock_release(ice->grp_lock);
+
+        PJ_RACE_ME(5);
 
         (*ice->cb.on_rx_data)(ice, comp_id, transport_id, pkt, pkt_size, 

pjproject/trunk/pjnath/src/pjnath/ice_strans.c

@@ -127 +127 @@
 
 /* Forward decls */
+static void ice_st_on_destroy(void *obj);
 static void destroy_ice_st(pj_ice_strans *ice_st);
 #define ice_st_perror(ice_st,msg,rc) pjnath_perror(ice_st->obj_name,msg,rc)
 static void sess_init_update(pj_ice_strans *ice_st);
-
-static void sess_add_ref(pj_ice_strans *ice_st);
-static pj_bool_t sess_dec_ref(pj_ice_strans *ice_st);
 
 /**
@@ -173 +171 @@
     pj_ice_strans_cfg        cfg;       /**< Configuration.             */
     pj_ice_strans_cb         cb;        /**< Application callback.      */
-    pj_lock_t               *init_lock; /**< Initialization mutex.      */
+    pj_grp_lock_t           *grp_lock;  /**< Group lock.                */
 
     pj_ice_strans_state      state;     /**< Session state.             */
@@ -184 +182 @@
     pj_timer_entry           ka_timer;  /**< STUN keep-alive timer.     */
 
-    pj_atomic_t             *busy_cnt;  /**< To prevent destroy         */
     pj_bool_t                destroy_req;/**< Destroy has been called?  */
     pj_bool_t                cb_called; /**< Init error callback called?*/
@@ -552 +549 @@
     pj_log_push_indent();
 
-    pj_ice_strans_cfg_copy(pool, &ice_st->cfg, cfg);
-    pj_memcpy(&ice_st->cb, cb, sizeof(*cb));
-    
-    status = pj_atomic_create(pool, 0, &ice_st->busy_cnt);
+    status = pj_grp_lock_create(pool, NULL, &ice_st->grp_lock);
     if (status != PJ_SUCCESS) {
-        destroy_ice_st(ice_st);
-        return status;
-    }
-
-    status = pj_lock_create_recursive_mutex(pool, ice_st->obj_name, 
-                                            &ice_st->init_lock);
-    if (status != PJ_SUCCESS) {
-        destroy_ice_st(ice_st);
+        pj_pool_release(pool);
         pj_log_pop_indent();
         return status;
     }
+
+    pj_grp_lock_add_ref(ice_st->grp_lock);
+    pj_grp_lock_add_handler(ice_st->grp_lock, pool, ice_st,
+                            &ice_st_on_destroy);
+
+    pj_ice_strans_cfg_copy(pool, &ice_st->cfg, cfg);
+    ice_st->cfg.stun.cfg.grp_lock = ice_st->grp_lock;
+    ice_st->cfg.turn.cfg.grp_lock = ice_st->grp_lock;
+    pj_memcpy(&ice_st->cb, cb, sizeof(*cb));
 
     ice_st->comp_cnt = comp_cnt;
@@ -579 +575 @@
      * called before we finish initialization.
      */
-    pj_lock_acquire(ice_st->init_lock);
+    pj_grp_lock_acquire(ice_st->grp_lock);
 
     for (i=0; i<comp_cnt; ++i) {
         status = create_comp(ice_st, i+1);
         if (status != PJ_SUCCESS) {
-            pj_lock_release(ice_st->init_lock);
+            pj_grp_lock_release(ice_st->grp_lock);
             destroy_ice_st(ice_st);
             pj_log_pop_indent();
@@ -592 +588 @@
 
     /* Done with initialization */
-    pj_lock_release(ice_st->init_lock);
-
-    PJ_LOG(4,(ice_st->obj_name, "ICE stream transport created"));
+    pj_grp_lock_release(ice_st->grp_lock);
+
+    PJ_LOG(4,(ice_st->obj_name, "ICE stream transport %p created", ice_st));
 
     *p_ice_st = ice_st;
@@ -606 +602 @@
 }
 
+/* REALLY destroy ICE */
+static void ice_st_on_destroy(void *obj)
+{
+    pj_ice_strans *ice_st = (pj_ice_strans*)obj;
+
+    PJ_LOG(4,(ice_st->obj_name, "ICE stream transport %p destroyed", obj));
+
+    /* Done */
+    pj_pool_release(ice_st->pool);
+}
+
 /* Destroy ICE */
 static void destroy_ice_st(pj_ice_strans *ice_st)
@@ -611 +618 @@
     unsigned i;
 
-    PJ_LOG(5,(ice_st->obj_name, "ICE stream transport destroying.."));
+    PJ_LOG(5,(ice_st->obj_name, "ICE stream transport %p destroy request..",
+              ice_st));
     pj_log_push_indent();
+
+    pj_grp_lock_acquire(ice_st->grp_lock);
+
+    if (ice_st->destroy_req) {
+        pj_grp_lock_release(ice_st->grp_lock);
+        return;
+    }
+
+    ice_st->destroy_req = PJ_TRUE;
 
     /* Destroy ICE if we have ICE */
@@ -624 +641 @@
         if (ice_st->comp[i]) {
             if (ice_st->comp[i]->stun_sock) {
-                pj_stun_sock_set_user_data(ice_st->comp[i]->stun_sock, NULL);
                 pj_stun_sock_destroy(ice_st->comp[i]->stun_sock);
                 ice_st->comp[i]->stun_sock = NULL;
             }
             if (ice_st->comp[i]->turn_sock) {
-                pj_turn_sock_set_user_data(ice_st->comp[i]->turn_sock, NULL);
                 pj_turn_sock_destroy(ice_st->comp[i]->turn_sock);
                 ice_st->comp[i]->turn_sock = NULL;
@@ -635 +650 @@
         }
     }
-    ice_st->comp_cnt = 0;
-
-    /* Destroy mutex */
-    if (ice_st->init_lock) {
-        pj_lock_acquire(ice_st->init_lock);
-        pj_lock_release(ice_st->init_lock);
-        pj_lock_destroy(ice_st->init_lock);
-        ice_st->init_lock = NULL;
-    }
-
-    /* Destroy reference counter */
-    if (ice_st->busy_cnt) {
-        pj_assert(pj_atomic_get(ice_st->busy_cnt)==0);
-        pj_atomic_destroy(ice_st->busy_cnt);
-        ice_st->busy_cnt = NULL;
-    }
-
-    PJ_LOG(4,(ice_st->obj_name, "ICE stream transport destroyed"));
-
-    /* Done */
-    pj_pool_release(ice_st->pool);
+
+    pj_grp_lock_dec_ref(ice_st->grp_lock);
+    pj_grp_lock_release(ice_st->grp_lock);
+
     pj_log_pop_indent();
 }
@@ -740 +738 @@
 PJ_DEF(pj_status_t) pj_ice_strans_destroy(pj_ice_strans *ice_st)
 {
-    PJ_ASSERT_RETURN(ice_st, PJ_EINVAL);
-
-    sess_add_ref(ice_st);
-    ice_st->destroy_req = PJ_TRUE;
-    if (sess_dec_ref(ice_st)) {
-        PJ_LOG(5,(ice_st->obj_name, 
-                  "ICE strans object is busy, will destroy later"));
-        return PJ_EPENDING;
-    }
-
+    destroy_ice_st(ice_st);
     return PJ_SUCCESS;
 }
 
-
-/*
- * Increment busy counter.
- */
-static void sess_add_ref(pj_ice_strans *ice_st)
-{
-    pj_atomic_inc(ice_st->busy_cnt);
-}
-
-/*
- * Decrement busy counter. If the counter has reached zero and destroy
- * has been requested, destroy the object and return FALSE.
- */
-static pj_bool_t sess_dec_ref(pj_ice_strans *ice_st)
-{
-    int count = pj_atomic_dec_and_get(ice_st->busy_cnt);
-    pj_assert(count >= 0);
-    if (count==0 && ice_st->destroy_req) {
-        destroy_ice_st(ice_st);
-        return PJ_FALSE;
-    } else {
-        return PJ_TRUE;
-    }
-}
 
 /*
@@ -841 +806 @@
     status = pj_ice_sess_create(&ice_st->cfg.stun_cfg, ice_st->obj_name, role,
                                 ice_st->comp_cnt, &ice_cb, 
-                                local_ufrag, local_passwd, &ice_st->ice);
+                                local_ufrag, local_passwd,
+                                ice_st->grp_lock,
+                                &ice_st->ice);
     if (status != PJ_SUCCESS)
         return status;
@@ -1256 +1223 @@
     unsigned msec;
 
-    sess_add_ref(ice_st);
+    pj_grp_lock_add_ref(ice_st->grp_lock);
 
     pj_gettimeofday(&t);
@@ -1338 +1305 @@
     }
 
-    sess_dec_ref(ice_st);
+    pj_grp_lock_dec_ref(ice_st->grp_lock);
 }
 
@@ -1427 +1394 @@
     ice_st = comp->ice_st;
 
-    sess_add_ref(ice_st);
+    pj_grp_lock_add_ref(ice_st->grp_lock);
 
     if (ice_st->ice == NULL) {
@@ -1452 +1419 @@
     }
 
-    return sess_dec_ref(ice_st);
+    return pj_grp_lock_dec_ref(ice_st->grp_lock) ? PJ_FALSE : PJ_TRUE;
 }
 
@@ -1483 +1450 @@
     ice_st = comp->ice_st;
 
-    sess_add_ref(ice_st);
+    pj_grp_lock_add_ref(ice_st->grp_lock);
 
     /* Wait until initialization completes */
-    pj_lock_acquire(ice_st->init_lock);
+    pj_grp_lock_acquire(ice_st->grp_lock);
 
     /* Find the srflx cancidate */
@@ -1496 +1463 @@
     }
 
-    pj_lock_release(ice_st->init_lock);
+    pj_grp_lock_release(ice_st->grp_lock);
 
     /* It is possible that we don't have srflx candidate even though this
@@ -1503 +1470 @@
      */
     if (cand == NULL) {
-        return sess_dec_ref(ice_st);
+        return pj_grp_lock_dec_ref(ice_st->grp_lock) ? PJ_FALSE : PJ_TRUE;
     }
 
@@ -1619 +1586 @@
     }
 
-    return sess_dec_ref(ice_st);
+    return pj_grp_lock_dec_ref(ice_st->grp_lock)? PJ_FALSE : PJ_TRUE;
 }
 
@@ -1638 +1605 @@
     }
 
-    sess_add_ref(comp->ice_st);
+    pj_grp_lock_add_ref(comp->ice_st->grp_lock);
 
     if (comp->ice_st->ice == NULL) {
@@ -1665 +1632 @@
     }
 
-    sess_dec_ref(comp->ice_st);
+    pj_grp_lock_dec_ref(comp->ice_st->grp_lock);
 }
 
@@ -1687 +1654 @@
     pj_log_push_indent();
 
-    sess_add_ref(comp->ice_st);
+    pj_grp_lock_add_ref(comp->ice_st->grp_lock);
 
     if (new_state == PJ_TURN_STATE_READY) {
@@ -1701 +1668 @@
 
         /* Wait until initialization completes */
-        pj_lock_acquire(comp->ice_st->init_lock);
+        pj_grp_lock_acquire(comp->ice_st->grp_lock);
 
         /* Find relayed candidate in the component */
@@ -1712 +1679 @@
         pj_assert(cand != NULL);
 
-        pj_lock_release(comp->ice_st->init_lock);
+        pj_grp_lock_release(comp->ice_st->grp_lock);
 
         /* Update candidate */
@@ -1745 +1712 @@
         comp->turn_sock = NULL;
 
-        /* Set session to fail if we're still initializing */
-        if (comp->ice_st->state < PJ_ICE_STRANS_STATE_READY) {
-            sess_fail(comp->ice_st, PJ_ICE_STRANS_OP_INIT,
-                      "TURN allocation failed", info.last_status);
-        } else if (comp->turn_err_cnt > 1) {
-            sess_fail(comp->ice_st, PJ_ICE_STRANS_OP_KEEP_ALIVE,
-                      "TURN refresh failed", info.last_status);
-        } else {
-            PJ_PERROR(4,(comp->ice_st->obj_name, info.last_status,
-                      "Comp %d: TURN allocation failed, retrying",
-                      comp->comp_id));
-            add_update_turn(comp->ice_st, comp);
-        }
-    }
-
-    sess_dec_ref(comp->ice_st);
+        /* Set session to fail on error. last_status PJ_SUCCESS means normal
+         * deallocation, which should not trigger sess_fail as it may have
+         * been initiated by ICE destroy
+         */
+        if (info.last_status != PJ_SUCCESS) {
+            if (comp->ice_st->state < PJ_ICE_STRANS_STATE_READY) {
+                sess_fail(comp->ice_st, PJ_ICE_STRANS_OP_INIT,
+                          "TURN allocation failed", info.last_status);
+            } else if (comp->turn_err_cnt > 1) {
+                sess_fail(comp->ice_st, PJ_ICE_STRANS_OP_KEEP_ALIVE,
+                          "TURN refresh failed", info.last_status);
+            } else {
+                PJ_PERROR(4,(comp->ice_st->obj_name, info.last_status,
+                          "Comp %d: TURN allocation failed, retrying",
+                          comp->comp_id));
+                add_update_turn(comp->ice_st, comp);
+            }
+        }
+    }
+
+    pj_grp_lock_dec_ref(comp->ice_st->grp_lock);
 
     pj_log_pop_indent();

pjproject/trunk/pjnath/src/pjnath/nat_detect.c

@@ -308 +308 @@
     sess_cb.on_send_msg = &on_send_msg;
     status = pj_stun_session_create(stun_cfg, pool->obj_name, &sess_cb,
-                                    PJ_FALSE, &sess->stun_sess);
+                                    PJ_FALSE, NULL, &sess->stun_sess);
     if (status != PJ_SUCCESS)
         goto on_error;

pjproject/trunk/pjnath/src/pjnath/stun_session.c

@@ -26 +26 @@
     pj_stun_config      *cfg;
     pj_pool_t           *pool;
-    pj_lock_t           *lock;
-    pj_bool_t            delete_lock;
+    pj_grp_lock_t       *grp_lock;
     pj_stun_session_cb   cb;
     void                *user_data;
-
-    pj_atomic_t         *busy;
-    pj_bool_t            destroy_request;
+    pj_bool_t            is_destroying;
 
     pj_bool_t            use_fingerprint;
@@ -56 +53 @@
 
 #define SNAME(s_)                   ((s_)->pool->obj_name)
-
-#if PJ_LOG_MAX_LEVEL >= 5
+#define THIS_FILE                   "stun_session.c"
+
+#if 1
 #   define TRACE_(expr)             PJ_LOG(5,expr)
 #else
@@ -63 +61 @@
 #endif
 
-#define LOG_ERR_(sess,title,rc) pjnath_perror(sess->pool->obj_name,title,rc)
+#define LOG_ERR_(sess,title,rc) PJ_PERROR(3,(sess->pool->obj_name,rc,title))
 
 #define TDATA_POOL_SIZE             PJNATH_POOL_LEN_STUN_TDATA
@@ -78 +76 @@
                                         pj_size_t pkt_size);
 static void stun_tsx_on_destroy(pj_stun_client_tsx *tsx);
+static void stun_sess_on_destroy(void *comp);
 
 static pj_stun_tsx_cb tsx_cb = 
@@ -149 +148 @@
 
     tdata = (pj_stun_tx_data*) pj_stun_client_tsx_get_data(tsx);
-    tsx_erase(tdata->sess, tdata);
-
-    pj_stun_client_tsx_destroy(tsx);
-    pj_pool_release(tdata->pool);
+    pj_stun_client_tsx_stop(tsx);
+    if (tdata) {
+        tsx_erase(tdata->sess, tdata);
+        pj_pool_release(tdata->pool);
+    }
+
+    TRACE_((THIS_FILE, "STUN transaction %p destroyed", tsx));
 }
 
 static void destroy_tdata(pj_stun_tx_data *tdata, pj_bool_t force)
 {
+    TRACE_((THIS_FILE, "tdata %p destroy request, force=%d, tsx=%p", tdata,
+            force, tdata->client_tsx));
+
     if (tdata->res_timer.id != PJ_FALSE) {
-        pj_timer_heap_cancel(tdata->sess->cfg->timer_heap, 
-                             &tdata->res_timer);
-        tdata->res_timer.id = PJ_FALSE;
+        pj_timer_heap_cancel_if_active(tdata->sess->cfg->timer_heap,
+                                       &tdata->res_timer, PJ_FALSE);
         pj_list_erase(tdata);
     }
 
     if (force) {
+        pj_list_erase(tdata);
         if (tdata->client_tsx) {
-            tsx_erase(tdata->sess, tdata);
-            pj_stun_client_tsx_destroy(tdata->client_tsx);
+            pj_stun_client_tsx_stop(tdata->client_tsx);
+            pj_stun_client_tsx_set_data(tdata->client_tsx, NULL);
         }
         pj_pool_release(tdata->pool);
@@ -173 +178 @@
     } else {
         if (tdata->client_tsx) {
-            pj_time_val delay = {2, 0};
+            /* "Probably" this is to absorb retransmission */
+            pj_time_val delay = {0, 300};
             pj_stun_client_tsx_schedule_destroy(tdata->client_tsx, &delay);
 
@@ -207 +213 @@
 
     pj_list_erase(tdata);
-    pj_stun_msg_destroy_tdata(tdata->sess, tdata);
+    destroy_tdata(tdata, PJ_FALSE);
 }
 
@@ -420 +426 @@
 
     /* Lock the session and prevent user from destroying us in the callback */
-    pj_atomic_inc(sess->busy);
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
+    if (sess->is_destroying) {
+        pj_stun_msg_destroy_tdata(sess, tdata);
+        pj_grp_lock_release(sess->grp_lock);
+        return;
+    }
 
     /* Handle authentication challenge */
@@ -435 +445 @@
      * from the pending list too. 
      */
-    pj_stun_msg_destroy_tdata(sess, tdata);
+    if (status == PJNATH_ESTUNTIMEDOUT)
+        destroy_tdata(tdata, PJ_TRUE);
+    else
+        destroy_tdata(tdata, PJ_FALSE);
     tdata = NULL;
 
-    pj_lock_release(sess->lock);
-
-    if (pj_atomic_dec_and_get(sess->busy)==0 && sess->destroy_request) {
-        pj_stun_session_destroy(sess);
-        return;
-    }
+    pj_grp_lock_release(sess->grp_lock);
 }
 
@@ -458 +466 @@
 
     /* Lock the session and prevent user from destroying us in the callback */
-    pj_atomic_inc(sess->busy);
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
     
+    if (sess->is_destroying) {
+        /* Stray timer */
+        pj_grp_lock_release(sess->grp_lock);
+        return PJ_EINVALIDOP;
+    }
+
     status = sess->cb.on_send_msg(tdata->sess, tdata->token, stun_pkt, 
                                   pkt_size, tdata->dst_addr, 
                                   tdata->addr_len);
-    pj_lock_release(sess->lock);
-
-    if (pj_atomic_dec_and_get(sess->busy)==0 && sess->destroy_request) {
-        pj_stun_session_destroy(sess);
-        return PJNATH_ESTUNDESTROYED;
-    } else {
-        return status;
-    }
+    if (pj_grp_lock_release(sess->grp_lock))
+        return PJ_EGONE;
+
+    return status;
 }
 
@@ -480 +489 @@
                                             const pj_stun_session_cb *cb,
                                             pj_bool_t fingerprint,
+                                            pj_grp_lock_t *grp_lock,
                                             pj_stun_session **p_sess)
 {
@@ -502 +512 @@
     sess->log_flag = 0xFFFF;
 
+    if (grp_lock) {
+        sess->grp_lock = grp_lock;
+    } else {
+        status = pj_grp_lock_create(pool, NULL, &sess->grp_lock);
+        if (status != PJ_SUCCESS) {
+            pj_pool_release(pool);
+            return status;
+        }
+    }
+
+    pj_grp_lock_add_ref(sess->grp_lock);
+    pj_grp_lock_add_handler(sess->grp_lock, pool, sess,
+                            &stun_sess_on_destroy);
+
     pj_stun_session_set_software_name(sess, &cfg->software_name);
 
-    sess->rx_pool = pj_pool_create(sess->cfg->pf, name, 
-                                   PJNATH_POOL_LEN_STUN_TDATA, 
+    sess->rx_pool = pj_pool_create(sess->cfg->pf, name,
+                                   PJNATH_POOL_LEN_STUN_TDATA,
                                    PJNATH_POOL_INC_STUN_TDATA, NULL);
 
@@ -511 +535 @@
     pj_list_init(&sess->cached_response_list);
 
-    status = pj_lock_create_recursive_mutex(pool, name, &sess->lock);
-    if (status != PJ_SUCCESS) {
-        pj_pool_release(pool);
-        return status;
-    }
-    sess->delete_lock = PJ_TRUE;
-
-    status = pj_atomic_create(pool, 0, &sess->busy);
-    if (status != PJ_SUCCESS) {
-        pj_lock_destroy(sess->lock);
-        pj_pool_release(pool);
-        return status;
-    }
-
     *p_sess = sess;
 
@@ -530 +540 @@
 }
 
-PJ_DEF(pj_status_t) pj_stun_session_destroy(pj_stun_session *sess)
-{
-    PJ_ASSERT_RETURN(sess, PJ_EINVAL);
-
-    pj_lock_acquire(sess->lock);
-
-    /* Can't destroy if we're in a callback */
-    sess->destroy_request = PJ_TRUE;
-    if (pj_atomic_get(sess->busy)) {
-        pj_lock_release(sess->lock);
-        return PJ_EPENDING;
-    }
+static void stun_sess_on_destroy(void *comp)
+{
+    pj_stun_session *sess = (pj_stun_session*)comp;
 
     while (!pj_list_empty(&sess->pending_request_list)) {
@@ -552 +553 @@
         destroy_tdata(tdata, PJ_TRUE);
     }
-    pj_lock_release(sess->lock);
-
-    if (sess->delete_lock) {
-        pj_lock_destroy(sess->lock);
-    }
 
     if (sess->rx_pool) {
@@ -565 +561 @@
     pj_pool_release(sess->pool);
 
+    TRACE_((THIS_FILE, "STUN session %p destroyed", sess));
+}
+
+PJ_DEF(pj_status_t) pj_stun_session_destroy(pj_stun_session *sess)
+{
+    pj_stun_tx_data *tdata;
+
+    PJ_ASSERT_RETURN(sess, PJ_EINVAL);
+
+    TRACE_((SNAME(sess), "STUN session %p destroy request, ref_cnt=%d",
+             sess, pj_grp_lock_get_ref(sess->grp_lock)));
+
+    pj_grp_lock_acquire(sess->grp_lock);
+
+    if (sess->is_destroying) {
+        /* Prevent from decrementing the ref counter more than once */
+        pj_grp_lock_release(sess->grp_lock);
+        return PJ_EINVALIDOP;
+    }
+
+    sess->is_destroying = PJ_TRUE;
+
+    /* We need to stop transactions and cached response because they are
+     * holding the group lock's reference counter while retransmitting.
+     */
+    tdata = sess->pending_request_list.next;
+    while (tdata != &sess->pending_request_list) {
+        if (tdata->client_tsx)
+            pj_stun_client_tsx_stop(tdata->client_tsx);
+        tdata = tdata->next;
+    }
+
+    tdata = sess->cached_response_list.next;
+    while (tdata != &sess->cached_response_list) {
+        pj_timer_heap_cancel_if_active(tdata->sess->cfg->timer_heap,
+                                       &tdata->res_timer, PJ_FALSE);
+        tdata = tdata->next;
+    }
+
+    pj_grp_lock_dec_ref(sess->grp_lock);
+    pj_grp_lock_release(sess->grp_lock);
     return PJ_SUCCESS;
 }
@@ -573 +610 @@
 {
     PJ_ASSERT_RETURN(sess, PJ_EINVAL);
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
     sess->user_data = user_data;
-    pj_lock_release(sess->lock);
+    pj_grp_lock_release(sess->grp_lock);
     return PJ_SUCCESS;
 }
@@ -585 +622 @@
 }
 
-PJ_DEF(pj_status_t) pj_stun_session_set_lock( pj_stun_session *sess,
-                                              pj_lock_t *lock,
-                                              pj_bool_t auto_del)
-{
-    pj_lock_t *old_lock = sess->lock;
-    pj_bool_t old_del;
-
-    PJ_ASSERT_RETURN(sess && lock, PJ_EINVAL);
-
-    pj_lock_acquire(old_lock);
-    sess->lock = lock;
-    old_del = sess->delete_lock;
-    sess->delete_lock = auto_del;
-    pj_lock_release(old_lock);
-
-    if (old_lock)
-        pj_lock_destroy(old_lock);
-
-    return PJ_SUCCESS;
-}
-
 PJ_DEF(pj_status_t) pj_stun_session_set_software_name(pj_stun_session *sess,
                                                       const pj_str_t *sw)
 {
     PJ_ASSERT_RETURN(sess, PJ_EINVAL);
+    pj_grp_lock_acquire(sess->grp_lock);
     if (sw && sw->slen)
         pj_strdup(sess->pool, &sess->srv_name, sw);
     else
         sess->srv_name.slen = 0;
+    pj_grp_lock_release(sess->grp_lock);
     return PJ_SUCCESS;
 }
@@ -623 +641 @@
     PJ_ASSERT_RETURN(sess, PJ_EINVAL);
 
+    pj_grp_lock_acquire(sess->grp_lock);
     sess->auth_type = auth_type;
     if (cred) {
@@ -630 +649 @@
         pj_bzero(&sess->cred, sizeof(sess->cred));
     }
+    pj_grp_lock_release(sess->grp_lock);
 
     return PJ_SUCCESS;
@@ -706 +726 @@
     PJ_ASSERT_RETURN(sess && p_tdata, PJ_EINVAL);
 
+    pj_grp_lock_acquire(sess->grp_lock);
+    if (sess->is_destroying) {
+        pj_grp_lock_release(sess->grp_lock);
+        return PJ_EINVALIDOP;
+    }
+
     status = create_tdata(sess, &tdata);
     if (status != PJ_SUCCESS)
-        return status;
+        goto on_error;
 
     /* Create STUN message */
     status = pj_stun_msg_create(tdata->pool, method,  magic, 
                                 tsx_id, &tdata->msg);
-    if (status != PJ_SUCCESS) {
-        pj_pool_release(tdata->pool);
-        return status;
-    }
+    if (status != PJ_SUCCESS)
+        goto on_error;
 
     /* copy the request's transaction ID as the transaction key. */
@@ -732 +756 @@
         /* MUST put authentication in request */
         status = get_auth(sess, tdata);
-        if (status != PJ_SUCCESS) {
-            pj_pool_release(tdata->pool);
-            return status;
-        }
+        if (status != PJ_SUCCESS)
+            goto on_error;
 
     } else if (sess->auth_type == PJ_STUN_AUTH_LONG_TERM) {
@@ -743 +765 @@
         if (sess->next_nonce.slen != 0) {
             status = get_auth(sess, tdata);
-            if (status != PJ_SUCCESS) {
-                pj_pool_release(tdata->pool);
-                return status;
-            }
+            if (status != PJ_SUCCESS)
+                goto on_error;
             tdata->auth_info.nonce = sess->next_nonce;
             tdata->auth_info.realm = sess->server_realm;
@@ -753 +773 @@
     } else {
         pj_assert(!"Invalid authentication type");
+        status = PJ_EBUG;
+        goto on_error;
+    }
+
+    *p_tdata = tdata;
+    pj_grp_lock_release(sess->grp_lock);
+    return PJ_SUCCESS;
+
+on_error:
+    if (tdata)
         pj_pool_release(tdata->pool);
-        return PJ_EBUG;
-    }
-
-    *p_tdata = tdata;
-    return PJ_SUCCESS;
+    pj_grp_lock_release(sess->grp_lock);
+    return status;
 }
 
@@ -770 +797 @@
     PJ_ASSERT_RETURN(sess && p_tdata, PJ_EINVAL);
 
+    pj_grp_lock_acquire(sess->grp_lock);
+    if (sess->is_destroying) {
+        pj_grp_lock_release(sess->grp_lock);
+        return PJ_EINVALIDOP;
+    }
+
     status = create_tdata(sess, &tdata);
-    if (status != PJ_SUCCESS)
+    if (status != PJ_SUCCESS) {
+        pj_grp_lock_release(sess->grp_lock);
         return status;
+    }
 
     /* Create STUN message */
@@ -780 +815 @@
     if (status != PJ_SUCCESS) {
         pj_pool_release(tdata->pool);
+        pj_grp_lock_release(sess->grp_lock);
         return status;
     }
 
     *p_tdata = tdata;
+
+    pj_grp_lock_release(sess->grp_lock);
     return PJ_SUCCESS;
 }
@@ -799 +837 @@
     pj_stun_tx_data *tdata = NULL;
 
+    pj_grp_lock_acquire(sess->grp_lock);
+    if (sess->is_destroying) {
+        pj_grp_lock_release(sess->grp_lock);
+        return PJ_EINVALIDOP;
+    }
+
     status = create_tdata(sess, &tdata);
-    if (status != PJ_SUCCESS)
+    if (status != PJ_SUCCESS) {
+        pj_grp_lock_release(sess->grp_lock);
         return status;
+    }
 
     /* Create STUN response message */
@@ -808 +854 @@
     if (status != PJ_SUCCESS) {
         pj_pool_release(tdata->pool);
+        pj_grp_lock_release(sess->grp_lock);
         return status;
     }
@@ -821 +868 @@
 
     *p_tdata = tdata;
+
+    pj_grp_lock_release(sess->grp_lock);
 
     return PJ_SUCCESS;
@@ -868 +917 @@
     PJ_ASSERT_RETURN(sess && addr_len && server && tdata, PJ_EINVAL);
 
+    /* Lock the session and prevent user from destroying us in the callback */
+    pj_grp_lock_acquire(sess->grp_lock);
+    if (sess->is_destroying) {
+        pj_grp_lock_release(sess->grp_lock);
+        return PJ_EINVALIDOP;
+    }
+
     pj_log_push_indent();
 
@@ -876 +932 @@
     tdata->token = token;
     tdata->retransmit = retransmit;
-
-    /* Lock the session and prevent user from destroying us in the callback */
-    pj_atomic_inc(sess->busy);
-    pj_lock_acquire(sess->lock);
 
     /* Apply options */
@@ -910 +962 @@
 
         /* Create STUN client transaction */
-        status = pj_stun_client_tsx_create(sess->cfg, tdata->pool, 
+        status = pj_stun_client_tsx_create(sess->cfg, tdata->pool,
+                                           sess->grp_lock,
                                            &tsx_cb, &tdata->client_tsx);
         PJ_ASSERT_RETURN(status==PJ_SUCCESS, status);
@@ -940 +993 @@
             
             pj_memset(&tdata->res_timer, 0, sizeof(tdata->res_timer));
-            pj_timer_entry_init(&tdata->res_timer, PJ_TRUE, tdata, 
+            pj_timer_entry_init(&tdata->res_timer, PJ_FALSE, tdata,
                                 &on_cache_timeout);
 
@@ -946 +999 @@
             timeout.msec = sess->cfg->res_cache_msec % 1000;
 
-            status = pj_timer_heap_schedule(sess->cfg->timer_heap, 
-                                            &tdata->res_timer,
-                                            &timeout);
+            status = pj_timer_heap_schedule_w_grp_lock(sess->cfg->timer_heap,
+                                                       &tdata->res_timer,
+                                                       &timeout, PJ_TRUE,
+                                                       sess->grp_lock);
             if (status != PJ_SUCCESS) {
-                tdata->res_timer.id = PJ_FALSE;
                 pj_stun_msg_destroy_tdata(sess, tdata);
                 LOG_ERR_(sess, "Error scheduling response timer", status);
@@ -976 +1029 @@
 
 on_return:
-    pj_lock_release(sess->lock);
-
     pj_log_pop_indent();
 
-    /* Check if application has called destroy() in the callback */
-    if (pj_atomic_dec_and_get(sess->busy)==0 && sess->destroy_request) {
-        pj_stun_session_destroy(sess);
-        return PJNATH_ESTUNDESTROYED;
-    }
+    if (pj_grp_lock_release(sess->grp_lock))
+        return PJ_EGONE;
 
     return status;
@@ -1006 +1054 @@
     pj_stun_tx_data *tdata;
 
+    pj_grp_lock_acquire(sess->grp_lock);
+    if (sess->is_destroying) {
+        pj_grp_lock_release(sess->grp_lock);
+        return PJ_EINVALIDOP;
+    }
+
     status = pj_stun_session_create_res(sess, rdata, code, 
                                         (errmsg?pj_cstr(&reason,errmsg):NULL), 
                                         &tdata);
-    if (status != PJ_SUCCESS)
+    if (status != PJ_SUCCESS) {
+        pj_grp_lock_release(sess->grp_lock);
         return status;
-
-    return pj_stun_session_send_msg(sess, token, cache, PJ_FALSE,
-                                    dst_addr,  addr_len, tdata);
+    }
+
+    status = pj_stun_session_send_msg(sess, token, cache, PJ_FALSE,
+                                      dst_addr,  addr_len, tdata);
+
+    pj_grp_lock_release(sess->grp_lock);
+    return status;
 }
 
@@ -1030 +1089 @@
 
     /* Lock the session and prevent user from destroying us in the callback */
-    pj_atomic_inc(sess->busy);
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
+    if (sess->is_destroying) {
+        pj_grp_lock_release(sess->grp_lock);
+        return PJ_EINVALIDOP;
+    }
 
     if (notify) {
@@ -1041 +1103 @@
     pj_stun_msg_destroy_tdata(sess, tdata);
 
-    pj_lock_release(sess->lock);
-
-    if (pj_atomic_dec_and_get(sess->busy)==0 && sess->destroy_request) {
-        pj_stun_session_destroy(sess);
-        return PJNATH_ESTUNDESTROYED;
-    }
+    pj_grp_lock_release(sess->grp_lock);
 
     return PJ_SUCCESS;
@@ -1064 +1121 @@
 
     /* Lock the session and prevent user from destroying us in the callback */
-    pj_atomic_inc(sess->busy);
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
+    if (sess->is_destroying) {
+        pj_grp_lock_release(sess->grp_lock);
+        return PJ_EINVALIDOP;
+    }
 
     status = pj_stun_client_tsx_retransmit(tdata->client_tsx, mod_count);
 
-    pj_lock_release(sess->lock);
-
-    if (pj_atomic_dec_and_get(sess->busy)==0 && sess->destroy_request) {
-        pj_stun_session_destroy(sess);
-        return PJNATH_ESTUNDESTROYED;
-    }
+    pj_grp_lock_release(sess->grp_lock);
 
     return status;
@@ -1362 +1417 @@
     PJ_ASSERT_RETURN(sess && packet && pkt_size, PJ_EINVAL);
 
+    /* Lock the session and prevent user from destroying us in the callback */
+    pj_grp_lock_acquire(sess->grp_lock);
+
+    if (sess->is_destroying) {
+        pj_grp_lock_release(sess->grp_lock);
+        return PJ_EINVALIDOP;
+    }
+
     pj_log_push_indent();
-
-    /* Lock the session and prevent user from destroying us in the callback */
-    pj_atomic_inc(sess->busy);
-    pj_lock_acquire(sess->lock);
 
     /* Reset pool */
@@ -1419 +1478 @@
 
 on_return:
-    pj_lock_release(sess->lock);
-
     pj_log_pop_indent();
 
-    /* If we've received destroy request while we're on the callback,
-     * destroy the session now.
-     */
-    if (pj_atomic_dec_and_get(sess->busy)==0 && sess->destroy_request) {
-        pj_stun_session_destroy(sess);
-        return PJNATH_ESTUNDESTROYED;
-    }
+    if (pj_grp_lock_release(sess->grp_lock))
+        return PJ_EGONE;
 
     return status;

pjproject/trunk/pjnath/src/pjnath/stun_sock.c

@@ -29 +29 @@
 #include <pj/ip_helper.h>
 #include <pj/log.h>
+#include <pj/os.h>
 #include <pj/pool.h>
 #include <pj/rand.h>
 
+#if 1
+#  define TRACE_(x)     PJ_LOG(5,x)
+#else
+#  define TRACE_(x)
+#endif
 
 enum { MAX_BIND_RETRY = 100 };
@@ -40 +46 @@
     pj_pool_t           *pool;          /* Pool                     */
     void                *user_data;     /* Application user data    */
-
+    pj_bool_t            is_destroying; /* Destroy already called   */
     int                  af;            /* Address family           */
     pj_stun_config       stun_cfg;      /* STUN config (ioqueue etc)*/
@@ -59 +65 @@
     pj_uint16_t          tsx_id[6];     /* .. to match STUN msg     */
     pj_stun_session     *stun_sess;     /* STUN session             */
-
+    pj_grp_lock_t       *grp_lock;      /* Session group lock       */
 };
 
@@ -65 +71 @@
  * Prototypes for static functions 
  */
+
+/* Destructor for group lock */
+static void stun_sock_destructor(void *obj);
 
 /* This callback is called by the STUN session to send packet */
@@ -203 +212 @@
         stun_sock->ka_interval = PJ_STUN_KEEP_ALIVE_SEC;
 
+    if (cfg && cfg->grp_lock) {
+        stun_sock->grp_lock = cfg->grp_lock;
+    } else {
+        status = pj_grp_lock_create(pool, NULL, &stun_sock->grp_lock);
+        if (status != PJ_SUCCESS) {
+            pj_pool_release(pool);
+            return status;
+        }
+    }
+
+    pj_grp_lock_add_ref(stun_sock->grp_lock);
+    pj_grp_lock_add_handler(stun_sock->grp_lock, pool, stun_sock,
+                            &stun_sock_destructor);
+
     /* Create socket and bind socket */
     status = pj_sock_socket(af, pj_SOCK_DGRAM(), 0, &stun_sock->sock_fd);
@@ -253 +276 @@
 
         pj_activesock_cfg_default(&activesock_cfg);
+        activesock_cfg.grp_lock = stun_sock->grp_lock;
         activesock_cfg.async_cnt = cfg->async_cnt;
         activesock_cfg.concurrency = 0;
@@ -291 +315 @@
                                         stun_sock->obj_name,
                                         &sess_cb, PJ_FALSE, 
+                                        stun_sock->grp_lock,
                                         &stun_sock->stun_sess);
         if (status != PJ_SUCCESS)
@@ -333 +358 @@
     PJ_ASSERT_RETURN(stun_sock && domain && default_port, PJ_EINVAL);
 
+    pj_grp_lock_acquire(stun_sock->grp_lock);
+
     /* Check whether the domain contains IP address */
     stun_sock->srv_addr.addr.sa_family = (pj_uint16_t)stun_sock->af;
@@ -361 +388 @@
 
         /* Processing will resume when the DNS SRV callback is called */
-        return status;
 
     } else {
@@ -379 +405 @@
 
         /* Start sending Binding request */
-        return get_mapped_addr(stun_sock);
-    }
-}
-
-/* Destroy */
-PJ_DEF(pj_status_t) pj_stun_sock_destroy(pj_stun_sock *stun_sock)
-{
+        status = get_mapped_addr(stun_sock);
+    }
+
+    pj_grp_lock_release(stun_sock->grp_lock);
+    return status;
+}
+
+/* Destructor */
+static void stun_sock_destructor(void *obj)
+{
+    pj_stun_sock *stun_sock = (pj_stun_sock*)obj;
+
     if (stun_sock->q) {
         pj_dns_srv_cancel_query(stun_sock->q, PJ_FALSE);
@@ -391 +422 @@
     }
 
-    if (stun_sock->stun_sess) {
-        pj_stun_session_set_user_data(stun_sock->stun_sess, NULL);
-    }
-    
-    /* Destroy the active socket first just in case we'll get
-     * stray callback.
-     */
-    if (stun_sock->active_sock != NULL) {
-        pj_activesock_t *asock = stun_sock->active_sock;
-        stun_sock->active_sock = NULL;
-        stun_sock->sock_fd = PJ_INVALID_SOCKET;
-        pj_activesock_set_user_data(asock, NULL);
-        pj_activesock_close(asock);
-    } else if (stun_sock->sock_fd != PJ_INVALID_SOCKET) {
-        pj_sock_close(stun_sock->sock_fd);
-        stun_sock->sock_fd = PJ_INVALID_SOCKET;
-    }
-
-    if (stun_sock->ka_timer.id != 0) {
-        pj_timer_heap_cancel(stun_sock->stun_cfg.timer_heap, 
-                             &stun_sock->ka_timer);
-        stun_sock->ka_timer.id = 0;
-    }
-
+    /*
     if (stun_sock->stun_sess) {
         pj_stun_session_destroy(stun_sock->stun_sess);
         stun_sock->stun_sess = NULL;
     }
+    */
 
     if (stun_sock->pool) {
@@ -426 +435 @@
     }
 
+    TRACE_(("", "STUN sock %p destroyed", stun_sock));
+
+}
+
+/* Destroy */
+PJ_DEF(pj_status_t) pj_stun_sock_destroy(pj_stun_sock *stun_sock)
+{
+    TRACE_((stun_sock->obj_name, "STUN sock %p request, ref_cnt=%d",
+            stun_sock, pj_grp_lock_get_ref(stun_sock->grp_lock)));
+
+    pj_grp_lock_acquire(stun_sock->grp_lock);
+    if (stun_sock->is_destroying) {
+        /* Destroy already called */
+        pj_grp_lock_release(stun_sock->grp_lock);
+        return PJ_EINVALIDOP;
+    }
+
+    stun_sock->is_destroying = PJ_TRUE;
+    pj_timer_heap_cancel_if_active(stun_sock->stun_cfg.timer_heap,
+                                   &stun_sock->ka_timer, 0);
+
+    if (stun_sock->active_sock != NULL) {
+        stun_sock->sock_fd = PJ_INVALID_SOCKET;
+        pj_activesock_close(stun_sock->active_sock);
+    } else if (stun_sock->sock_fd != PJ_INVALID_SOCKET) {
+        pj_sock_close(stun_sock->sock_fd);
+        stun_sock->sock_fd = PJ_INVALID_SOCKET;
+    }
+
+    if (stun_sock->stun_sess) {
+        pj_stun_session_destroy(stun_sock->stun_sess);
+    }
+    pj_grp_lock_dec_ref(stun_sock->grp_lock);
+    pj_grp_lock_release(stun_sock->grp_lock);
     return PJ_SUCCESS;
 }
@@ -469 +512 @@
     pj_stun_sock *stun_sock = (pj_stun_sock*) user_data;
 
+    pj_grp_lock_acquire(stun_sock->grp_lock);
+
     /* Clear query */
     stun_sock->q = NULL;
@@ -475 +520 @@
     if (status != PJ_SUCCESS) {
         sess_fail(stun_sock, PJ_STUN_SOCK_DNS_OP, status);
+        pj_grp_lock_release(stun_sock->grp_lock);
         return;
     }
@@ -491 +537 @@
     /* Start sending Binding request */
     get_mapped_addr(stun_sock);
+
+    pj_grp_lock_release(stun_sock->grp_lock);
 }
 
@@ -534 +582 @@
     PJ_ASSERT_RETURN(stun_sock && info, PJ_EINVAL);
 
+    pj_grp_lock_acquire(stun_sock->grp_lock);
+
     /* Copy STUN server address and mapped address */
     pj_memcpy(&info->srv_addr, &stun_sock->srv_addr,
@@ -544 +594 @@
     status = pj_sock_getsockname(stun_sock->sock_fd, &info->bound_addr,
                                  &addr_len);
-    if (status != PJ_SUCCESS)
+    if (status != PJ_SUCCESS) {
+        pj_grp_lock_release(stun_sock->grp_lock);
         return status;
+    }
 
     /* If socket is bound to a specific interface, then only put that
@@ -561 +613 @@
         /* Get the default address */
         status = pj_gethostip(stun_sock->af, &def_addr);
-        if (status != PJ_SUCCESS)
+        if (status != PJ_SUCCESS) {
+            pj_grp_lock_release(stun_sock->grp_lock);
             return status;
+        }
         
         pj_sockaddr_set_port(&def_addr, port);
@@ -570 +624 @@
         status = pj_enum_ip_interface(stun_sock->af, &info->alias_cnt, 
                                       info->aliases);
-        if (status != PJ_SUCCESS)
+        if (status != PJ_SUCCESS) {
+            pj_grp_lock_release(stun_sock->grp_lock);
             return status;
+        }
 
         /* Set the port number for each address.
@@ -591 +647 @@
     }
 
+    pj_grp_lock_release(stun_sock->grp_lock);
     return PJ_SUCCESS;
 }
@@ -604 +661 @@
 {
     pj_ssize_t size;
+    pj_status_t status;
+
     PJ_ASSERT_RETURN(stun_sock && pkt && dst_addr && addr_len, PJ_EINVAL);
     
+    pj_grp_lock_acquire(stun_sock->grp_lock);
+
+    if (!stun_sock->active_sock) {
+        /* We have been shutdown, but this callback may still get called
+         * by retransmit timer.
+         */
+        pj_grp_lock_release(stun_sock->grp_lock);
+        return PJ_EINVALIDOP;
+    }
+
     if (send_key==NULL)
         send_key = &stun_sock->send_key;
 
     size = pkt_len;
-    return pj_activesock_sendto(stun_sock->active_sock, send_key, 
-                                pkt, &size, flag, dst_addr, addr_len);
+    status = pj_activesock_sendto(stun_sock->active_sock, send_key,
+                                  pkt, &size, flag, dst_addr, addr_len);
+
+    pj_grp_lock_release(stun_sock->grp_lock);
+    return status;
 }
 
@@ -626 +698 @@
 
     stun_sock = (pj_stun_sock *) pj_stun_session_get_user_data(sess);
-    if (!stun_sock || !stun_sock->active_sock)
+    if (!stun_sock || !stun_sock->active_sock) {
+        /* We have been shutdown, but this callback may still get called
+         * by retransmit timer.
+         */
         return PJ_EINVALIDOP;
+    }
 
     pj_assert(token==INTERNAL_MSG_TOKEN);
@@ -633 +709 @@
 
     size = pkt_size;
-    return pj_activesock_sendto(stun_sock->active_sock, 
+    return pj_activesock_sendto(stun_sock->active_sock,
                                 &stun_sock->int_send_key,
                                 pkt, &size, 0, dst_addr, addr_len);
@@ -727 +803 @@
 static void start_ka_timer(pj_stun_sock *stun_sock)
 {
-    if (stun_sock->ka_timer.id != 0) {
-        pj_timer_heap_cancel(stun_sock->stun_cfg.timer_heap, 
-                             &stun_sock->ka_timer);
-        stun_sock->ka_timer.id = 0;
-    }
+    pj_timer_heap_cancel_if_active(stun_sock->stun_cfg.timer_heap,
+                                   &stun_sock->ka_timer, 0);
 
     pj_assert(stun_sock->ka_interval != 0);
-    if (stun_sock->ka_interval > 0) {
+    if (stun_sock->ka_interval > 0 && !stun_sock->is_destroying) {
         pj_time_val delay;
 
@@ -740 +813 @@
         delay.msec = 0;
 
-        if (pj_timer_heap_schedule(stun_sock->stun_cfg.timer_heap, 
-                                   &stun_sock->ka_timer, 
-                                   &delay) == PJ_SUCCESS)
-        {
-            stun_sock->ka_timer.id = PJ_TRUE;
-        }
+        pj_timer_heap_schedule_w_grp_lock(stun_sock->stun_cfg.timer_heap,
+                                          &stun_sock->ka_timer,
+                                          &delay, PJ_TRUE,
+                                          stun_sock->grp_lock);
     }
 }
@@ -757 +828 @@
 
     PJ_UNUSED_ARG(th);
+    pj_grp_lock_acquire(stun_sock->grp_lock);
 
     /* Time to send STUN Binding request */
-    if (get_mapped_addr(stun_sock) != PJ_SUCCESS)
+    if (get_mapped_addr(stun_sock) != PJ_SUCCESS) {
+        pj_grp_lock_release(stun_sock->grp_lock);
         return;
+    }
 
     /* Next keep-alive timer will be scheduled once the request
      * is complete.
      */
+    pj_grp_lock_release(stun_sock->grp_lock);
 }
 
@@ -788 +863 @@
         return PJ_TRUE;
     }
+
+    pj_grp_lock_acquire(stun_sock->grp_lock);
 
     /* Check that this is STUN message */
@@ -824 +901 @@
                                        PJ_STUN_IS_DATAGRAM, NULL, NULL,
                                        src_addr, addr_len);
-    return status!=PJNATH_ESTUNDESTROYED ? PJ_TRUE : PJ_FALSE;
+
+    status = pj_grp_lock_release(stun_sock->grp_lock);
+
+    return status!=PJ_EGONE ? PJ_TRUE : PJ_FALSE;
 
 process_app_data:
@@ -832 +912 @@
         ret = (*stun_sock->cb.on_rx_data)(stun_sock, data, size,
                                           src_addr, addr_len);
-        return ret;
-    }
-
-    return PJ_TRUE;
+        status = pj_grp_lock_release(stun_sock->grp_lock);
+        return status!=PJ_EGONE ? PJ_TRUE : PJ_FALSE;
+    }
+
+    status = pj_grp_lock_release(stun_sock->grp_lock);
+    return status!=PJ_EGONE ? PJ_TRUE : PJ_FALSE;
 }
 
@@ -857 +939 @@
     if (stun_sock->cb.on_data_sent) {
         pj_bool_t ret;
+
+        pj_grp_lock_acquire(stun_sock->grp_lock);
 
         /* If app gives NULL send_key in sendto() function, then give
@@ -867 +951 @@
         ret = (*stun_sock->cb.on_data_sent)(stun_sock, send_key, sent);
 
+        pj_grp_lock_release(stun_sock->grp_lock);
         return ret;
     }

pjproject/trunk/pjnath/src/pjnath/stun_transaction.c

@@ -27 +27 @@
 
 
+#define THIS_FILE               "stun_transaction.c"
+#define TIMER_INACTIVE          0
 #define TIMER_ACTIVE            1
 
@@ -35 +37 @@
     pj_stun_tsx_cb       cb;
     void                *user_data;
+    pj_grp_lock_t       *grp_lock;
 
     pj_bool_t            complete;
@@ -52 +55 @@
 
 
+#if 1
+#   define TRACE_(expr)             PJ_LOG(5,expr)
+#else
+#   define TRACE_(expr)
+#endif
+
+
 static void retransmit_timer_callback(pj_timer_heap_t *timer_heap, 
                                       pj_timer_entry *timer);
@@ -57 +67 @@
                                    pj_timer_entry *timer);
 
-#define stun_perror(tsx,msg,rc) pjnath_perror(tsx->obj_name, msg, rc)
-
 /*
  * Create a STUN client transaction.
@@ -64 +72 @@
 PJ_DEF(pj_status_t) pj_stun_client_tsx_create(pj_stun_config *cfg,
                                               pj_pool_t *pool,
+                                              pj_grp_lock_t *grp_lock,
                                               const pj_stun_tsx_cb *cb,
                                               pj_stun_client_tsx **p_tsx)
@@ -75 +84 @@
     tsx->rto_msec = cfg->rto_msec;
     tsx->timer_heap = cfg->timer_heap;
+    tsx->grp_lock = grp_lock;
     pj_memcpy(&tsx->cb, cb, sizeof(*cb));
 
@@ -83 +93 @@
     tsx->destroy_timer.user_data = tsx;
 
-    pj_ansi_snprintf(tsx->obj_name, sizeof(tsx->obj_name), "stuntsx%p", tsx);
+    pj_ansi_snprintf(tsx->obj_name, sizeof(tsx->obj_name), "utsx%p", tsx);
 
     *p_tsx = tsx;
@@ -101 +111 @@
     PJ_ASSERT_RETURN(tsx->cb.on_destroy, PJ_EINVAL);
 
+    pj_grp_lock_acquire(tsx->grp_lock);
+
     /* Cancel previously registered timer */
-    if (tsx->destroy_timer.id != 0) {
-        pj_timer_heap_cancel(tsx->timer_heap, &tsx->destroy_timer);
-        tsx->destroy_timer.id = 0;
-    }
+    pj_timer_heap_cancel_if_active(tsx->timer_heap, &tsx->destroy_timer,
+                                   TIMER_INACTIVE);
 
     /* Stop retransmission, just in case */
-    if (tsx->retransmit_timer.id != 0) {
-        pj_timer_heap_cancel(tsx->timer_heap, &tsx->retransmit_timer);
-        tsx->retransmit_timer.id = 0;
-    }
-
-    status = pj_timer_heap_schedule(tsx->timer_heap,
-                                    &tsx->destroy_timer, delay);
-    if (status != PJ_SUCCESS)
+    pj_timer_heap_cancel_if_active(tsx->timer_heap, &tsx->retransmit_timer,
+                                   TIMER_INACTIVE);
+
+    status = pj_timer_heap_schedule_w_grp_lock(tsx->timer_heap,
+                                               &tsx->destroy_timer, delay,
+                                               TIMER_ACTIVE, tsx->grp_lock);
+    if (status != PJ_SUCCESS) {
+        pj_grp_lock_release(tsx->grp_lock);
         return status;
-
-    tsx->destroy_timer.id = TIMER_ACTIVE;
+    }
+
     tsx->cb.on_complete = NULL;
 
+    pj_grp_lock_release(tsx->grp_lock);
+
+    TRACE_((tsx->obj_name, "STUN transaction %p schedule destroy", tsx));
+
     return PJ_SUCCESS;
 }
@@ -128 +142 @@
  * Destroy transaction immediately.
  */
-PJ_DEF(pj_status_t) pj_stun_client_tsx_destroy(pj_stun_client_tsx *tsx)
+PJ_DEF(pj_status_t) pj_stun_client_tsx_stop(pj_stun_client_tsx *tsx)
 {
     PJ_ASSERT_RETURN(tsx, PJ_EINVAL);
 
-    if (tsx->retransmit_timer.id != 0) {
-        pj_timer_heap_cancel(tsx->timer_heap, &tsx->retransmit_timer);
-        tsx->retransmit_timer.id = 0;
-    }
-    if (tsx->destroy_timer.id != 0) {
-        pj_timer_heap_cancel(tsx->timer_heap, &tsx->destroy_timer);
-        tsx->destroy_timer.id = 0;
-    }
-
-    PJ_LOG(5,(tsx->obj_name, "STUN client transaction destroyed"));
+    /* Don't call grp_lock_acquire() because we might be called on
+     * group lock's destructor.
+     */
+    pj_timer_heap_cancel_if_active(tsx->timer_heap, &tsx->retransmit_timer,
+                                   TIMER_INACTIVE);
+    pj_timer_heap_cancel_if_active(tsx->timer_heap, &tsx->destroy_timer,
+                                   TIMER_INACTIVE);
+
+    PJ_LOG(5,(tsx->obj_name, "STUN client transaction %p stopped, ref_cnt=%d",
+              tsx, pj_grp_lock_get_ref(tsx->grp_lock)));
+
     return PJ_SUCCESS;
 }
@@ -186 +201 @@
     pj_status_t status;
 
-    PJ_ASSERT_RETURN(tsx->retransmit_timer.id == 0 ||
+    PJ_ASSERT_RETURN(tsx->retransmit_timer.id == TIMER_INACTIVE ||
                      !tsx->require_retransmit, PJ_EBUSY);
 
@@ -212 +227 @@
          * cancel transmission).
          */;
-        status = pj_timer_heap_schedule(tsx->timer_heap, 
-                                        &tsx->retransmit_timer,
-                                        &tsx->retransmit_time);
+        status = pj_timer_heap_schedule_w_grp_lock(tsx->timer_heap,
+                                                   &tsx->retransmit_timer,
+                                                   &tsx->retransmit_time,
+                                                   TIMER_ACTIVE,
+                                                   tsx->grp_lock);
         if (status != PJ_SUCCESS) {
-            tsx->retransmit_timer.id = 0;
+            tsx->retransmit_timer.id = TIMER_INACTIVE;
             return status;
         }
-        tsx->retransmit_timer.id = TIMER_ACTIVE;
     }
 
@@ -236 +252 @@
         /* We've been destroyed, don't access the object. */
     } else if (status != PJ_SUCCESS) {
-        if (tsx->retransmit_timer.id != 0 && mod_count) {
-            pj_timer_heap_cancel(tsx->timer_heap, 
-                                 &tsx->retransmit_timer);
-            tsx->retransmit_timer.id = 0;
-        }
-        stun_perror(tsx, "STUN error sending message", status);
+        if (mod_count) {
+                pj_timer_heap_cancel_if_active( tsx->timer_heap,
+                                                &tsx->retransmit_timer,
+                                                TIMER_INACTIVE);
+        }
+        PJ_PERROR(4, (tsx->obj_name, status, "STUN error sending message"));
     }
 
@@ -261 +277 @@
     PJ_ASSERT_RETURN(tsx && pkt && pkt_len, PJ_EINVAL);
     PJ_ASSERT_RETURN(tsx->retransmit_timer.id == 0, PJ_EBUSY);
+
+    pj_grp_lock_acquire(tsx->grp_lock);
 
     /* Encode message */
@@ -287 +305 @@
          * cancel transmission).
          */;
-        status = pj_timer_heap_schedule(tsx->timer_heap, 
-                                        &tsx->retransmit_timer,
-                                        &tsx->retransmit_time);
+        status = pj_timer_heap_schedule_w_grp_lock(tsx->timer_heap,
+                                                   &tsx->retransmit_timer,
+                                                   &tsx->retransmit_time,
+                                                   TIMER_ACTIVE,
+                                                   tsx->grp_lock);
         if (status != PJ_SUCCESS) {
-            tsx->retransmit_timer.id = 0;
+            tsx->retransmit_timer.id = TIMER_INACTIVE;
+            pj_grp_lock_release(tsx->grp_lock);
             return status;
         }
-        tsx->retransmit_timer.id = TIMER_ACTIVE;
     }
 
@@ -300 +320 @@
     status = tsx_transmit_msg(tsx, PJ_TRUE);
     if (status != PJ_SUCCESS) {
-        if (tsx->retransmit_timer.id != 0) {
-            pj_timer_heap_cancel(tsx->timer_heap, 
-                                 &tsx->retransmit_timer);
-            tsx->retransmit_timer.id = 0;
-        }
+        pj_timer_heap_cancel_if_active(tsx->timer_heap,
+                                       &tsx->retransmit_timer,
+                                       TIMER_INACTIVE);
+        pj_grp_lock_release(tsx->grp_lock);
         return status;
     }
 
+    pj_grp_lock_release(tsx->grp_lock);
     return PJ_SUCCESS;
 }
@@ -320 +340 @@
 
     PJ_UNUSED_ARG(timer_heap);
+    pj_grp_lock_acquire(tsx->grp_lock);
 
     if (tsx->transmit_count >= PJ_STUN_MAX_TRANSMIT_COUNT) {
@@ -332 +353 @@
             }
         }
+        pj_grp_lock_release(tsx->grp_lock);
         /* We might have been destroyed, don't try to access the object */
         pj_log_pop_indent();
@@ -339 +361 @@
     tsx->retransmit_timer.id = 0;
     status = tsx_transmit_msg(tsx, PJ_TRUE);
-    if (status == PJNATH_ESTUNDESTROYED) {
-        /* We've been destroyed, don't try to access the object */
-    } else if (status != PJ_SUCCESS) {
+    if (status != PJ_SUCCESS) {
         tsx->retransmit_timer.id = 0;
         if (!tsx->complete) {
@@ -349 +369 @@
             }
         }
-        /* We might have been destroyed, don't try to access the object */
-    }
+    }
+
+    pj_grp_lock_release(tsx->grp_lock);
+    /* We might have been destroyed, don't try to access the object */
 }
 
@@ -363 +385 @@
     }
 
-    if (tsx->retransmit_timer.id != 0) {
-        pj_timer_heap_cancel(tsx->timer_heap, &tsx->retransmit_timer);
-        tsx->retransmit_timer.id = 0;
-    }
+    pj_timer_heap_cancel_if_active(tsx->timer_heap, &tsx->retransmit_timer,
+                                   TIMER_INACTIVE);
 
     return tsx_transmit_msg(tsx, mod_count);
@@ -380 +400 @@
 
     tsx->destroy_timer.id = PJ_FALSE;
+
     tsx->cb.on_destroy(tsx);
     /* Don't access transaction after this */
@@ -409 +430 @@
      * We can cancel retransmit timer now.
      */
-    if (tsx->retransmit_timer.id) {
-        pj_timer_heap_cancel(tsx->timer_heap, &tsx->retransmit_timer);
-        tsx->retransmit_timer.id = 0;
-    }
+    pj_timer_heap_cancel_if_active(tsx->timer_heap, &tsx->retransmit_timer,
+                                   TIMER_INACTIVE);
 
     /* Find STUN error code attribute */

pjproject/trunk/pjnath/src/pjnath/turn_session.c

@@ -113 +113 @@
     void                *user_data;
     pj_stun_config       stun_cfg;
-
-    pj_lock_t           *lock;
+    pj_bool_t            is_destroying;
+
+    pj_grp_lock_t       *grp_lock;
     int                  busy;
 
@@ -162 +163 @@
 static void sess_shutdown(pj_turn_session *sess,
                           pj_status_t status);
+static void turn_sess_on_destroy(void *comp);
 static void do_destroy(pj_turn_session *sess);
 static void send_refresh(pj_turn_session *sess, int lifetime);
@@ -237 +239 @@
                                             int af,
                                             pj_turn_tp_type conn_type,
+                                            pj_grp_lock_t *grp_lock,
                                             const pj_turn_session_cb *cb,
                                             unsigned options,
@@ -245 +248 @@
     pj_turn_session *sess;
     pj_stun_session_cb stun_cb;
-    pj_lock_t *null_lock;
     pj_status_t status;
 
@@ -282 +284 @@
 
     /* Session lock */
-    status = pj_lock_create_recursive_mutex(pool, sess->obj_name, 
-                                            &sess->lock);
-    if (status != PJ_SUCCESS) {
-        do_destroy(sess);
-        return status;
-    }
+    if (grp_lock) {
+        sess->grp_lock = grp_lock;
+    } else {
+        status = pj_grp_lock_create(pool, NULL, &sess->grp_lock);
+        if (status != PJ_SUCCESS) {
+            pj_pool_release(pool);
+            return status;
+        }
+    }
+
+    pj_grp_lock_add_ref(sess->grp_lock);
+    pj_grp_lock_add_handler(sess->grp_lock, pool, sess,
+                            &turn_sess_on_destroy);
 
     /* Timer */
@@ -298 +307 @@
     stun_cb.on_rx_indication = &stun_on_rx_indication;
     status = pj_stun_session_create(&sess->stun_cfg, sess->obj_name, &stun_cb,
-                                    PJ_FALSE, &sess->stun);
+                                    PJ_FALSE, sess->grp_lock, &sess->stun);
     if (status != PJ_SUCCESS) {
         do_destroy(sess);
@@ -307 +316 @@
     pj_stun_session_set_user_data(sess->stun, sess);
 
-    /* Replace mutex in STUN session with a NULL mutex, since access to
-     * STUN session is serialized.
-     */
-    status = pj_lock_create_null_mutex(pool, name, &null_lock);
-    if (status != PJ_SUCCESS) {
-        do_destroy(sess);
-        return status;
-    }
-    pj_stun_session_set_lock(sess->stun, null_lock, PJ_TRUE);
-
     /* Done */
 
@@ -326 +325 @@
 
 
-/* Destroy */
-static void do_destroy(pj_turn_session *sess)
-{
-    /* Lock session */
-    if (sess->lock) {
-        pj_lock_acquire(sess->lock);
-    }
-
-    /* Cancel pending timer, if any */
-    if (sess->timer.id != TIMER_NONE) {
-        pj_timer_heap_cancel(sess->timer_heap, &sess->timer);
-        sess->timer.id = TIMER_NONE;
-    }
-
-    /* Destroy STUN session */
-    if (sess->stun) {
-        pj_stun_session_destroy(sess->stun);
-        sess->stun = NULL;
-    }
-
-    /* Destroy lock */
-    if (sess->lock) {
-        pj_lock_release(sess->lock);
-        pj_lock_destroy(sess->lock);
-        sess->lock = NULL;
-    }
+static void turn_sess_on_destroy(void *comp)
+{
+    pj_turn_session *sess = (pj_turn_session*) comp;
 
     /* Destroy pool */
@@ -362 +338 @@
         pj_pool_release(pool);
     }
+}
+
+/* Destroy */
+static void do_destroy(pj_turn_session *sess)
+{
+    PJ_LOG(4,(sess->obj_name, "TURN session destroy request, ref_cnt=%d",
+              pj_grp_lock_get_ref(sess->grp_lock)));
+
+    pj_grp_lock_acquire(sess->grp_lock);
+    if (sess->is_destroying) {
+        pj_grp_lock_release(sess->grp_lock);
+        return;
+    }
+
+    sess->is_destroying = PJ_TRUE;
+    pj_timer_heap_cancel_if_active(sess->timer_heap, &sess->timer, TIMER_NONE);
+    pj_stun_session_destroy(sess->stun);
+
+    pj_grp_lock_dec_ref(sess->grp_lock);
+    pj_grp_lock_release(sess->grp_lock);
 }
 
@@ -438 +434 @@
         set_state(sess, PJ_TURN_STATE_DESTROYING);
 
-        if (sess->timer.id != TIMER_NONE) {
-            pj_timer_heap_cancel(sess->timer_heap, &sess->timer);
-            sess->timer.id = TIMER_NONE;
-        }
-
-        sess->timer.id = TIMER_DESTROY;
-        pj_timer_heap_schedule(sess->timer_heap, &sess->timer, &delay);
+        pj_timer_heap_cancel_if_active(sess->timer_heap, &sess->timer,
+                                       TIMER_NONE);
+        pj_timer_heap_schedule_w_grp_lock(sess->timer_heap, &sess->timer,
+                                          &delay, TIMER_DESTROY,
+                                          sess->grp_lock);
     }
 }
@@ -456 +450 @@
     PJ_ASSERT_RETURN(sess, PJ_EINVAL);
 
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
 
     sess_shutdown(sess, PJ_SUCCESS);
 
-    pj_lock_release(sess->lock);
+    pj_grp_lock_release(sess->grp_lock);
 
     return PJ_SUCCESS;
@@ -554 +548 @@
     pj_status_t status;
 
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
     status = pj_stun_session_set_software_name(sess->stun, sw);
-    pj_lock_release(sess->lock);
+    pj_grp_lock_release(sess->grp_lock);
 
     return status;
@@ -577 +571 @@
     PJ_ASSERT_RETURN(sess->state == PJ_TURN_STATE_NULL, PJ_EINVALIDOP);
 
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
 
     /* See if "domain" contains just IP address */
@@ -677 +671 @@
 
 on_return:
-    pj_lock_release(sess->lock);
+    pj_grp_lock_release(sess->grp_lock);
     return status;
 }
@@ -691 +685 @@
     PJ_ASSERT_RETURN(sess->stun, PJ_EINVALIDOP);
 
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
 
     pj_stun_session_set_credential(sess->stun, PJ_STUN_AUTH_LONG_TERM, cred);
 
-    pj_lock_release(sess->lock);
+    pj_grp_lock_release(sess->grp_lock);
 
     return PJ_SUCCESS;
@@ -716 +710 @@
                      PJ_EINVALIDOP);
 
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
 
     if (param && param != &sess->alloc_param) 
@@ -727 +721 @@
                   state_names[sess->state]));
 
-        pj_lock_release(sess->lock);
+        pj_grp_lock_release(sess->grp_lock);
         return PJ_SUCCESS;
 
@@ -739 +733 @@
                                         PJ_STUN_MAGIC, NULL, &tdata);
     if (status != PJ_SUCCESS) {
-        pj_lock_release(sess->lock);
+        pj_grp_lock_release(sess->grp_lock);
         return status;
     }
@@ -779 +773 @@
     }
 
-    pj_lock_release(sess->lock);
+    pj_grp_lock_release(sess->grp_lock);
     return status;
 }
@@ -800 +794 @@
     PJ_ASSERT_RETURN(sess && addr_cnt && addr, PJ_EINVAL);
 
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
 
     /* Create a bare CreatePermission request */
@@ -807 +801 @@
                                         PJ_STUN_MAGIC, NULL, &tdata);
     if (status != PJ_SUCCESS) {
-        pj_lock_release(sess->lock);
+        pj_grp_lock_release(sess->grp_lock);
         return status;
     }
@@ -858 +852 @@
     }
 
-    pj_lock_release(sess->lock);
+    pj_grp_lock_release(sess->grp_lock);
     return PJ_SUCCESS;
 
@@ -875 +869 @@
             invalidate_perm(sess, perm);
     }
-    pj_lock_release(sess->lock);
+    pj_grp_lock_release(sess->grp_lock);
     return status;
 }
@@ -946 +940 @@
 
     /* Lock session now */
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
 
     /* Lookup permission first */
@@ -961 +955 @@
                                           0);
         if (status != PJ_SUCCESS) {
-            pj_lock_release(sess->lock);
+            pj_grp_lock_release(sess->grp_lock);
             return status;
         }
@@ -1036 +1030 @@
 
 on_return:
-    pj_lock_release(sess->lock);
+    pj_grp_lock_release(sess->grp_lock);
     return status;
 }
@@ -1056 +1050 @@
     PJ_ASSERT_RETURN(sess->state == PJ_TURN_STATE_READY, PJ_EINVALIDOP);
 
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
 
     /* Create blank ChannelBind request */
@@ -1099 +1093 @@
 
 on_return:
-    pj_lock_release(sess->lock);
+    pj_grp_lock_release(sess->grp_lock);
     return status;
 }
@@ -1122 +1116 @@
 
     /* Start locking the session */
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
 
     is_datagram = (sess->conn_type==PJ_TURN_TP_UDP);
@@ -1194 +1188 @@
 
 on_return:
-    pj_lock_release(sess->lock);
+    pj_grp_lock_release(sess->grp_lock);
     return status;
 }
@@ -1386 +1380 @@
     /* Cancel existing keep-alive timer, if any */
     pj_assert(sess->timer.id != TIMER_DESTROY);
-
-    if (sess->timer.id != TIMER_NONE) {
-        pj_timer_heap_cancel(sess->timer_heap, &sess->timer);
-        sess->timer.id = TIMER_NONE;
+    if (sess->timer.id == TIMER_KEEP_ALIVE) {
+        pj_timer_heap_cancel_if_active(sess->timer_heap, &sess->timer,
+                                       TIMER_NONE);
     }
 
     /* Start keep-alive timer once allocation succeeds */
-    timeout.sec = sess->ka_interval;
-    timeout.msec = 0;
-
-    sess->timer.id = TIMER_KEEP_ALIVE;
-    pj_timer_heap_schedule(sess->timer_heap, &sess->timer, &timeout);
-
-    set_state(sess, PJ_TURN_STATE_READY);
+    if (sess->state < PJ_TURN_STATE_DEALLOCATING) {
+        timeout.sec = sess->ka_interval;
+        timeout.msec = 0;
+
+        pj_timer_heap_schedule_w_grp_lock(sess->timer_heap, &sess->timer,
+                                          &timeout, TIMER_KEEP_ALIVE,
+                                          sess->grp_lock);
+
+        set_state(sess, PJ_TURN_STATE_READY);
+    }
 }
 
@@ -1949 +1945 @@
     PJ_UNUSED_ARG(th);
 
-    pj_lock_acquire(sess->lock);
+    pj_grp_lock_acquire(sess->grp_lock);
 
     eid = (enum timer_id_t) e->id;
@@ -2026 +2022 @@
             delay.msec = 0;
 
-            sess->timer.id = TIMER_KEEP_ALIVE;
-            pj_timer_heap_schedule(sess->timer_heap, &sess->timer, &delay);
-        }
-
-        pj_lock_release(sess->lock);
+            pj_timer_heap_schedule_w_grp_lock(sess->timer_heap, &sess->timer,
+                                              &delay, TIMER_KEEP_ALIVE,
+                                              sess->grp_lock);
+        }
 
     } else if (eid == TIMER_DESTROY) {
         /* Time to destroy */
-        pj_lock_release(sess->lock);
         do_destroy(sess);
     } else {
         pj_assert(!"Unknown timer event");
-        pj_lock_release(sess->lock);
-    }
-}
-
+    }
+
+    pj_grp_lock_release(sess->grp_lock);
+}
+

pjproject/trunk/pjnath/src/pjnath/turn_sock.c

@@ -47 +47 @@
     void                *user_data;
 
-    pj_lock_t           *lock;
+    pj_bool_t            is_destroying;
+    pj_grp_lock_t       *grp_lock;
 
     pj_turn_alloc_param  alloc_param;
@@ -53 +54 @@
     pj_turn_sock_cfg     setting;
 
-    pj_bool_t            destroy_request;
     pj_timer_entry       timer;
 
@@ -94 +94 @@
 
 
+static void turn_sock_on_destroy(void *comp);
 static void destroy(pj_turn_sock *turn_sock);
 static void timer_cb(pj_timer_heap_t *th, pj_timer_entry *e);
@@ -169 +170 @@
     }
 
-    /* Create lock */
-    status = pj_lock_create_recursive_mutex(pool, turn_sock->obj_name, 
-                                            &turn_sock->lock);
-    if (status != PJ_SUCCESS) {
-        destroy(turn_sock);
-        return status;
-    }
+    /* Session lock */
+    if (setting && setting->grp_lock) {
+        turn_sock->grp_lock = setting->grp_lock;
+    } else {
+        status = pj_grp_lock_create(pool, NULL, &turn_sock->grp_lock);
+        if (status != PJ_SUCCESS) {
+            pj_pool_release(pool);
+            return status;
+        }
+    }
+
+    pj_grp_lock_add_ref(turn_sock->grp_lock);
+    pj_grp_lock_add_handler(turn_sock->grp_lock, pool, turn_sock,
+                            &turn_sock_on_destroy);
 
     /* Init timer */
@@ -187 +195 @@
     sess_cb.on_state = &turn_on_state;
     status = pj_turn_session_create(cfg, pool->obj_name, af, conn_type,
-                                    &sess_cb, 0, turn_sock, &turn_sock->sess);
+                                    turn_sock->grp_lock, &sess_cb, 0,
+                                    turn_sock, &turn_sock->sess);
     if (status != PJ_SUCCESS) {
         destroy(turn_sock);
@@ -204 +213 @@
  * Destroy.
  */
-static void destroy(pj_turn_sock *turn_sock)
-{
-    if (turn_sock->lock) {
-        pj_lock_acquire(turn_sock->lock);
-    }
-
-    if (turn_sock->sess) {
-        pj_turn_session_set_user_data(turn_sock->sess, NULL);
-        pj_turn_session_shutdown(turn_sock->sess);
-        turn_sock->sess = NULL;
-    }
-
-    if (turn_sock->active_sock) {
-        pj_activesock_set_user_data(turn_sock->active_sock, NULL);
-        pj_activesock_close(turn_sock->active_sock);
-        turn_sock->active_sock = NULL;
-    }
-
-    if (turn_sock->lock) {
-        pj_lock_release(turn_sock->lock);
-        pj_lock_destroy(turn_sock->lock);
-        turn_sock->lock = NULL;
-    }
+static void turn_sock_on_destroy(void *comp)
+{
+    pj_turn_sock *turn_sock = (pj_turn_sock*) comp;
 
     if (turn_sock->pool) {
         pj_pool_t *pool = turn_sock->pool;
+        PJ_LOG(4,(turn_sock->obj_name, "TURN socket destroyed"));
         turn_sock->pool = NULL;
         pj_pool_release(pool);
@@ -235 +225 @@
 }
 
+static void destroy(pj_turn_sock *turn_sock)
+{
+    PJ_LOG(4,(turn_sock->obj_name, "TURN socket destroy request, ref_cnt=%d",
+              pj_grp_lock_get_ref(turn_sock->grp_lock)));
+
+    pj_grp_lock_acquire(turn_sock->grp_lock);
+    if (turn_sock->is_destroying) {
+        pj_grp_lock_release(turn_sock->grp_lock);
+        return;
+    }
+
+    turn_sock->is_destroying = PJ_TRUE;
+    if (turn_sock->sess)
+        pj_turn_session_shutdown(turn_sock->sess);
+    if (turn_sock->active_sock)
+        pj_activesock_close(turn_sock->active_sock);
+    pj_grp_lock_dec_ref(turn_sock->grp_lock);
+    pj_grp_lock_release(turn_sock->grp_lock);
+}
 
 PJ_DEF(void) pj_turn_sock_destroy(pj_turn_sock *turn_sock)
 {
-    pj_lock_acquire(turn_sock->lock);
-    turn_sock->destroy_request = PJ_TRUE;
+    pj_grp_lock_acquire(turn_sock->grp_lock);
+    if (turn_sock->is_destroying) {
+        pj_grp_lock_release(turn_sock->grp_lock);
+        return;
+    }
 
     if (turn_sock->sess) {
@@ -247 +259 @@
          * destroy ourselves.
          */
-        pj_lock_release(turn_sock->lock);
     } else {
-        pj_lock_release(turn_sock->lock);
         destroy(turn_sock);
     }
 
+    pj_grp_lock_release(turn_sock->grp_lock);
 }
 
@@ -268 +279 @@
     switch (eid) {
     case TIMER_DESTROY:
-        PJ_LOG(5,(turn_sock->obj_name, "Destroying TURN"));
         destroy(turn_sock);
         break;
@@ -338 +348 @@
 PJ_DEF(pj_status_t) pj_turn_sock_lock(pj_turn_sock *turn_sock)
 {
-    return pj_lock_acquire(turn_sock->lock);
+    return pj_grp_lock_acquire(turn_sock->grp_lock);
 }
 
@@ -346 +356 @@
 PJ_DEF(pj_status_t) pj_turn_sock_unlock(pj_turn_sock *turn_sock)
 {
-    return pj_lock_release(turn_sock->lock);
+    return pj_grp_lock_release(turn_sock->grp_lock);
 }
 
@@ -381 +391 @@
     PJ_ASSERT_RETURN(turn_sock && domain, PJ_EINVAL);
     PJ_ASSERT_RETURN(turn_sock->sess, PJ_EINVALIDOP);
+
+    pj_grp_lock_acquire(turn_sock->grp_lock);
 
     /* Copy alloc param. We will call session_alloc() only after the 
@@ -396 +408 @@
         if (status != PJ_SUCCESS) {
             sess_fail(turn_sock, "Error setting credential", status);
+            pj_grp_lock_release(turn_sock->grp_lock);
             return status;
         }
@@ -405 +418 @@
     if (status != PJ_SUCCESS) {
         sess_fail(turn_sock, "Error setting TURN server", status);
+        pj_grp_lock_release(turn_sock->grp_lock);
         return status;
     }
@@ -411 +425 @@
      * to RESOLVED state.
      */
-
+    pj_grp_lock_release(turn_sock->grp_lock);
     return PJ_SUCCESS;
 }
@@ -473 +487 @@
         return PJ_FALSE;
 
+    pj_grp_lock_acquire(turn_sock->grp_lock);
+
     /* TURN session may have already been destroyed here.
      * See ticket #1557 (http://trac.pjsip.org/repos/ticket/1557).
@@ -478 +494 @@
     if (!turn_sock->sess) {
         sess_fail(turn_sock, "TURN session already destroyed", status);
+        pj_grp_lock_release(turn_sock->grp_lock);
         return PJ_FALSE;
     }
@@ -483 +500 @@
     if (status != PJ_SUCCESS) {
         sess_fail(turn_sock, "TCP connect() error", status);
+        pj_grp_lock_release(turn_sock->grp_lock);
         return PJ_FALSE;
     }
@@ -501 +519 @@
     if (status != PJ_SUCCESS) {
         sess_fail(turn_sock, "Error sending ALLOCATE", status);
+        pj_grp_lock_release(turn_sock->grp_lock);
         return PJ_FALSE;
     }
 
+    pj_grp_lock_release(turn_sock->grp_lock);
     return PJ_TRUE;
 }
@@ -563 +583 @@
 
     turn_sock = (pj_turn_sock*) pj_activesock_get_user_data(asock);
-    pj_lock_acquire(turn_sock->lock);
-
-    if (status == PJ_SUCCESS && turn_sock->sess) {
+    pj_grp_lock_acquire(turn_sock->grp_lock);
+
+    if (status == PJ_SUCCESS && turn_sock->sess && !turn_sock->is_destroying) {
         /* Report incoming packet to TURN session, repeat while we have
          * "packet" in the buffer (required for stream-oriented transports)
@@ -615 +635 @@
 
 on_return:
-    pj_lock_release(turn_sock->lock);
+    pj_grp_lock_release(turn_sock->grp_lock);
 
     return ret;
@@ -635 +655 @@
     pj_status_t status;
 
-    if (turn_sock == NULL) {
+    if (turn_sock == NULL || turn_sock->is_destroying) {
         /* We've been destroyed */
         // https://trac.pjsip.org/repos/ticket/1316
@@ -681 +701 @@
     pj_turn_sock *turn_sock = (pj_turn_sock*) 
                            pj_turn_session_get_user_data(sess);
-    if (turn_sock == NULL) {
+    if (turn_sock == NULL || turn_sock->is_destroying) {
         /* We've been destroyed */
         return;
@@ -730 +750 @@
         int sock_type;
         pj_sock_t sock;
+        pj_activesock_cfg asock_cfg;
         pj_activesock_cb asock_cb;
         pj_sockaddr bound_addr, *cfg_bind_addr;
@@ -791 +812 @@
 
         /* Create active socket */
+        pj_activesock_cfg_default(&asock_cfg);
+        asock_cfg.grp_lock = turn_sock->grp_lock;
+
         pj_bzero(&asock_cb, sizeof(asock_cb));
         asock_cb.on_data_read = &on_data_read;
         asock_cb.on_connect_complete = &on_connect_complete;
         status = pj_activesock_create(turn_sock->pool, sock,
-                                      sock_type, NULL,
+                                      sock_type, &asock_cfg,
                                       turn_sock->cfg.ioqueue, &asock_cb, 
                                       turn_sock,
@@ -836 +860 @@
         pj_turn_session_set_user_data(sess, NULL);
 
-        if (turn_sock->timer.id) {
-            pj_timer_heap_cancel(turn_sock->cfg.timer_heap, &turn_sock->timer);
-            turn_sock->timer.id = 0;
-        }
-
-        turn_sock->timer.id = TIMER_DESTROY;
-        pj_timer_heap_schedule(turn_sock->cfg.timer_heap, &turn_sock->timer, 
-                               &delay);
-    }
-}
-
-
+        pj_timer_heap_cancel_if_active(turn_sock->cfg.timer_heap,
+                                       &turn_sock->timer, 0);
+        pj_timer_heap_schedule_w_grp_lock(turn_sock->cfg.timer_heap,
+                                          &turn_sock->timer,
+                                          &delay, TIMER_DESTROY,
+                                          turn_sock->grp_lock);
+    }
+}
+
+

pjproject/trunk/pjnath/src/pjturn-srv/allocation.c

@@ -339 +339 @@
     sess_cb.on_rx_indication = &stun_on_rx_indication;
     status = pj_stun_session_create(&srv->core.stun_cfg, alloc->obj_name,
-                                    &sess_cb, PJ_FALSE, &alloc->sess);
+                                    &sess_cb, PJ_FALSE, NULL, &alloc->sess);
     if (status != PJ_SUCCESS) {
         goto on_error;

pjproject/trunk/pjnath/src/pjturn-srv/server.c

@@ -156 +156 @@
 
     status = pj_stun_session_create(&srv->core.stun_cfg, srv->obj_name,
-                                    &sess_cb, PJ_FALSE, &srv->core.stun_sess);
+                                    &sess_cb, PJ_FALSE, NULL,
+                                    &srv->core.stun_sess);
     if (status != PJ_SUCCESS) {
         goto on_error;