Changeset 4359 for pjproject/trunk/pjlib/src/pj/ioqueue_select.c

Timestamp:

Feb 21, 2013 11:18:36 AM (12 years ago)

Author:

bennylp

Message:

Fixed #1616: Implementation of Group lock and other foundation in PJLIB for fixing synchronization issues

File:

• pjproject/trunk/pjlib/src/pj/ioqueue_select.c (modified) (15 diffs)

pjproject/trunk/pjlib/src/pj/ioqueue_select.c

@@ -40 +40 @@
 #include <pj/sock_qos.h>
 #include <pj/errno.h>
+#include <pj/rand.h>
 
 /* Now that we have access to OS'es <sys/select>, lets check again that
@@ -238 +239 @@
         key = PJ_POOL_ALLOC_T(pool, pj_ioqueue_key_t);
         key->ref_count = 0;
-        rc = pj_mutex_create_recursive(pool, NULL, &key->mutex);
+        rc = pj_lock_create_recursive_mutex(pool, NULL, &key->lock);
         if (rc != PJ_SUCCESS) {
             key = ioqueue->free_list.next;
             while (key != &ioqueue->free_list) {
-                pj_mutex_destroy(key->mutex);
+                pj_lock_destroy(key->lock);
                 key = key->next;
             }
@@ -285 +286 @@
     key = ioqueue->active_list.next;
     while (key != &ioqueue->active_list) {
-        pj_mutex_destroy(key->mutex);
+        pj_lock_destroy(key->lock);
         key = key->next;
     }
@@ -291 +292 @@
     key = ioqueue->closing_list.next;
     while (key != &ioqueue->closing_list) {
-        pj_mutex_destroy(key->mutex);
+        pj_lock_destroy(key->lock);
         key = key->next;
     }
@@ -297 +298 @@
     key = ioqueue->free_list.next;
     while (key != &ioqueue->free_list) {
-        pj_mutex_destroy(key->mutex);
+        pj_lock_destroy(key->lock);
         key = key->next;
     }
@@ -313 +314 @@
  * Register socket handle to ioqueue.
  */
-PJ_DEF(pj_status_t) pj_ioqueue_register_sock( pj_pool_t *pool,
+PJ_DEF(pj_status_t) pj_ioqueue_register_sock2(pj_pool_t *pool,
                                               pj_ioqueue_t *ioqueue,
                                               pj_sock_t sock,
+                                              pj_grp_lock_t *grp_lock,
                                               void *user_data,
                                               const pj_ioqueue_callback *cb,
@@ -359 +361 @@
 #endif
 
-    rc = ioqueue_init_key(pool, ioqueue, key, sock, user_data, cb);
+    rc = ioqueue_init_key(pool, ioqueue, key, sock, grp_lock, user_data, cb);
     if (rc != PJ_SUCCESS) {
         key = NULL;
@@ -387 +389 @@
 on_return:
     /* On error, socket may be left in non-blocking mode. */
+    if (rc != PJ_SUCCESS) {
+        if (key->grp_lock)
+            pj_grp_lock_dec_ref_dbg(key->grp_lock, "ioqueue", 0);
+    }
     *p_key = key;
     pj_lock_release(ioqueue->lock);
@@ -393 +399 @@
 }
 
+PJ_DEF(pj_status_t) pj_ioqueue_register_sock( pj_pool_t *pool,
+                                              pj_ioqueue_t *ioqueue,
+                                              pj_sock_t sock,
+                                              void *user_data,
+                                              const pj_ioqueue_callback *cb,
+                                              pj_ioqueue_key_t **p_key)
+{
+    return pj_ioqueue_register_sock2(pool, ioqueue, sock, NULL, user_data,
+                                     cb, p_key);
+}
+
 #if PJ_IOQUEUE_HAS_SAFE_UNREG
 /* Increment key's reference counter */
@@ -447 +464 @@
      * deadlock.
      */
-    pj_mutex_lock(key->mutex);
+    pj_ioqueue_lock_key(key);
 
     /* Also lock ioqueue */
@@ -486 +503 @@
 
     /* Done. */
-    pj_mutex_unlock(key->mutex);
+    if (key->grp_lock) {
+        /* just dec_ref and unlock. we will set grp_lock to NULL
+         * elsewhere */
+        pj_grp_lock_t *grp_lock = key->grp_lock;
+        // Don't set grp_lock to NULL otherwise the other thread
+        // will crash. Just leave it as dangling pointer, but this
+        // should be safe
+        //key->grp_lock = NULL;
+        pj_grp_lock_dec_ref_dbg(grp_lock, "ioqueue", 0);
+        pj_grp_lock_release(grp_lock);
+    } else {
+        pj_ioqueue_unlock_key(key);
+    }
 #else
-    pj_mutex_destroy(key->mutex);
+    if (key->grp_lock) {
+        /* set grp_lock to NULL and unlock */
+        pj_grp_lock_t *grp_lock = key->grp_lock;
+        // Don't set grp_lock to NULL otherwise the other thread
+        // will crash. Just leave it as dangling pointer, but this
+        // should be safe
+        //key->grp_lock = NULL;
+        pj_grp_lock_dec_ref_dbg(grp_lock, "ioqueue", 0);
+        pj_grp_lock_release(grp_lock);
+    } else {
+        pj_ioqueue_unlock_key(key);
+    }
+
+    pj_lock_destroy(key->lock);
 #endif
 
@@ -621 +663 @@
         if (PJ_TIME_VAL_GTE(now, h->free_time)) {
             pj_list_erase(h);
+            // Don't set grp_lock to NULL otherwise the other thread
+            // will crash. Just leave it as dangling pointer, but this
+            // should be safe
+            //h->grp_lock = NULL;
             pj_list_push_back(&ioqueue->free_list, h);
         }
@@ -782 +828 @@
 {
     pj_fd_set_t rfdset, wfdset, xfdset;
-    int count, counter;
+    int count, i, counter;
     pj_ioqueue_key_t *h;
     struct event
@@ -893 +939 @@
     }
 
+    for (i=0; i<counter; ++i) {
+        if (event[i].key->grp_lock)
+            pj_grp_lock_add_ref_dbg(event[i].key->grp_lock, "ioqueue", 0);
+    }
+
+    PJ_RACE_ME(5);
+
     pj_lock_release(ioqueue->lock);
+
+    PJ_RACE_ME(5);
 
     count = counter;
@@ -919 +974 @@
         decrement_counter(event[counter].key);
 #endif
+
+        if (event[counter].key->grp_lock)
+            pj_grp_lock_dec_ref_dbg(event[counter].key->grp_lock,
+                                    "ioqueue", 0);
     }