Changeset 1879 for pjproject/trunk/pjnath/src/pjnath/turn_udp.c

Timestamp:

Mar 20, 2008 4:32:06 PM (15 years ago)

Author:

bennylp

Message:

More ticket #485: client and server self tested

File:

• pjproject/trunk/pjnath/src/pjnath/turn_udp.c (modified) (12 diffs)

pjproject/trunk/pjnath/src/pjnath/turn_udp.c

@@ -20 +20 @@
 #include <pj/assert.h>
 #include <pj/errno.h>
+#include <pj/lock.h>
 #include <pj/log.h>
 #include <pj/pool.h>
 #include <pj/ioqueue.h>
+
+enum
+{
+    TIMER_NONE,
+    TIMER_DESTROY
+};
 
 struct pj_turn_udp
@@ -30 +37 @@
     pj_turn_udp_cb       cb;
     void                *user_data;
+
+    pj_lock_t           *lock;
+
+    pj_bool_t            destroy_request;
+    pj_timer_heap_t     *timer_heap;
+    pj_timer_entry       timer;
 
     pj_sock_t            sock;
@@ -65 +78 @@
 
 
+static void destroy(pj_turn_udp *udp_rel);
+static void timer_cb(pj_timer_heap_t *th, pj_timer_entry *e);
+
+
 /*
  * Create.
@@ -92 +109 @@
     if (cb) {
         pj_memcpy(&udp_rel->cb, cb, sizeof(*cb));
+    }
+
+    /* Create lock */
+    status = pj_lock_create_recursive_mutex(pool, pool->obj_name, 
+                                            &udp_rel->lock);
+    if (status != PJ_SUCCESS) {
+        destroy(udp_rel);
+        return status;
+    }
+
+    /* Init timer */
+    udp_rel->timer_heap = cfg->timer_heap;
+    pj_timer_entry_init(&udp_rel->timer, TIMER_NONE, udp_rel, &timer_cb);
+
+    /* Init socket */
+    status = pj_sock_socket(af, pj_SOCK_DGRAM(), 0, &udp_rel->sock);
+    if (status != PJ_SUCCESS) {
+        destroy(udp_rel);
+        return status;
+    }
+
+    /* Bind to any */
+    pj_sockaddr_init(af, &udp_rel->src_addr, NULL, 0);
+    status = pj_sock_bind(udp_rel->sock, &udp_rel->src_addr, 
+                          pj_sockaddr_get_len(&udp_rel->src_addr));
+    if (status != PJ_SUCCESS) {
+        destroy(udp_rel);
+        return status;
+    }
+
+    /* Register to ioqeuue */
+    pj_bzero(&ioq_cb, sizeof(ioq_cb));
+    ioq_cb.on_read_complete = &on_read_complete;
+    status = pj_ioqueue_register_sock(udp_rel->pool, cfg->ioqueue, 
+                                      udp_rel->sock, udp_rel, 
+                                      &ioq_cb, &udp_rel->key);
+    if (status != PJ_SUCCESS) {
+        destroy(udp_rel);
+        return status;
     }
 
@@ -103 +159 @@
                                     &sess_cb, udp_rel, 0, &udp_rel->sess);
     if (status != PJ_SUCCESS) {
-        pj_turn_udp_destroy(udp_rel);
-        return status;
-    }
-
-    /* Init socket */
-    status = pj_sock_socket(af, pj_SOCK_DGRAM(), 0, &udp_rel->sock);
-    if (status != PJ_SUCCESS) {
-        pj_turn_udp_destroy(udp_rel);
-        return status;
-    }
-
-    /* Bind to any */
-    pj_sockaddr_init(af, &udp_rel->src_addr, NULL, 0);
-    status = pj_sock_bind(udp_rel->sock, &udp_rel->src_addr, 
-                          pj_sockaddr_get_len(&udp_rel->src_addr));
-    if (status != PJ_SUCCESS) {
-        pj_turn_udp_destroy(udp_rel);
-        return status;
-    }
-
-    /* Register to ioqeuue */
-    pj_bzero(&ioq_cb, sizeof(ioq_cb));
-    ioq_cb.on_read_complete = &on_read_complete;
-    status = pj_ioqueue_register_sock(udp_rel->pool, cfg->ioqueue, 
-                                      udp_rel->sock, udp_rel, 
-                                      &ioq_cb, &udp_rel->key);
-    if (status != PJ_SUCCESS) {
-        pj_turn_udp_destroy(udp_rel);
+        destroy(udp_rel);
         return status;
     }
@@ -145 +174 @@
  * Destroy.
  */
-PJ_DEF(void) pj_turn_udp_destroy(pj_turn_udp *udp_rel)
-{
+static void destroy(pj_turn_udp *udp_rel)
+{
+    if (udp_rel->lock) {
+        pj_lock_acquire(udp_rel->lock);
+    }
+
     if (udp_rel->sess) {
+        pj_turn_session_set_user_data(udp_rel->sess, NULL);
         pj_turn_session_destroy(udp_rel->sess);
         udp_rel->sess = NULL;
+    }
+
+    if (udp_rel->key) {
+        pj_ioqueue_unregister(udp_rel->key);
+        udp_rel->key = NULL;
+        udp_rel->sock = 0;
+    } else if (udp_rel->sock) {
+        pj_sock_close(udp_rel->sock);
+        udp_rel->sock = 0;
+    }
+
+    if (udp_rel->lock) {
+        pj_lock_release(udp_rel->lock);
+        pj_lock_destroy(udp_rel->lock);
+        udp_rel->lock = NULL;
     }
 
@@ -158 +207 @@
     }
 }
+
+PJ_DEF(void) pj_turn_udp_destroy(pj_turn_udp *udp_rel)
+{
+    pj_lock_acquire(udp_rel->lock);
+    udp_rel->destroy_request = PJ_TRUE;
+
+    if (udp_rel->sess) {
+        pj_turn_session_destroy(udp_rel->sess);
+        /* This will ultimately call our state callback, and when
+         * session state is DESTROYING we will schedule a timer to
+         * destroy ourselves.
+         */
+        pj_lock_release(udp_rel->lock);
+    } else {
+        pj_lock_release(udp_rel->lock);
+        destroy(udp_rel);
+    }
+
+}
+
+/* Timer callback */
+static void timer_cb(pj_timer_heap_t *th, pj_timer_entry *e)
+{
+    pj_turn_udp *udp_rel = (pj_turn_udp*)e->user_data;
+    int eid = e->id;
+
+    PJ_UNUSED_ARG(th);
+
+    e->id = TIMER_NONE;
+
+    switch (eid) {
+    case TIMER_DESTROY:
+        destroy(udp_rel);
+        break;
+    default:
+        pj_assert(!"Invalid timer id");
+        break;
+    }
+}
+
 
 /*
@@ -272 +361 @@
 
     udp_rel = (pj_turn_udp*) pj_ioqueue_get_user_data(key);
+    pj_lock_acquire(udp_rel->lock);
 
     do {
@@ -301 +391 @@
              ++retry < MAX_RETRY);
 
+    pj_lock_release(udp_rel->lock);
 }
 
@@ -317 +408 @@
     pj_ssize_t len = pkt_len;
 
+    if (udp_rel == NULL) {
+        /* We've been destroyed */
+        pj_assert(!"We should shutdown gracefully");
+        return PJ_EINVALIDOP;
+    }
+
     return pj_sock_sendto(udp_rel->sock, pkt, &len, 0,
                           dst_addr, dst_addr_len);
@@ -348 +445 @@
     pj_turn_udp *udp_rel = (pj_turn_udp*) 
                            pj_turn_session_get_user_data(sess);
+    if (udp_rel == NULL) {
+        /* We've been destroyed */
+        return;
+    }
+
     if (udp_rel->cb.on_rx_data) {
         (*udp_rel->cb.on_rx_data)(udp_rel, pkt, pkt_len, 
@@ -364 +466 @@
     pj_turn_udp *udp_rel = (pj_turn_udp*) 
                            pj_turn_session_get_user_data(sess);
+    if (udp_rel == NULL) {
+        /* We've been destroyed */
+        return;
+    }
+
     if (udp_rel->cb.on_state) {
         (*udp_rel->cb.on_state)(udp_rel, old_state, new_state);
     }
 
-    if (new_state > PJ_TURN_STATE_READY) {
-        udp_rel->sess = NULL;
-    }
-}
-
-
+    if (new_state >= PJ_TURN_STATE_DESTROYING && udp_rel->sess) {
+        if (udp_rel->destroy_request) {
+            pj_time_val delay = {0, 0};
+
+            pj_turn_session_set_user_data(udp_rel->sess, NULL);
+
+            udp_rel->timer.id = TIMER_DESTROY;
+            pj_timer_heap_schedule(udp_rel->timer_heap, &udp_rel->timer, 
+                                   &delay);
+        } else {
+            udp_rel->sess = NULL;
+        }
+    }
+}
+
+