Changeset 1789 for pjproject/trunk/pjlib/src/pj/ioqueue_common_abs.c

Timestamp:

Feb 13, 2008 3:17:28 PM (16 years ago)

Author:

bennylp

Message:

Ticket #474: option in ioqueue to control concurrency (to allow/disallow simultaneous/multiple callback calls)

File:

• pjproject/trunk/pjlib/src/pj/ioqueue_common_abs.c (modified) (16 diffs)

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

@@ -97 +97 @@
 #endif
 
+    rc = pj_ioqueue_set_concurrency(key, ioqueue->default_concurrency);
+    if (rc != PJ_SUCCESS)
+        return rc;
+
     /* Get socket type. When socket type is datagram, some optimization
      * will be performed during send to allow parallel send operations.
@@ -194 +198 @@
         /* Completion of connect() operation */
         pj_ssize_t bytes_transfered;
+        pj_bool_t has_lock;
 
         /* Clear operation. */
@@ -247 +252 @@
 #endif
 
-        /* Unlock; from this point we don't need to hold key's mutex. */
-        pj_mutex_unlock(h->mutex);
+        /* Unlock; from this point we don't need to hold key's mutex
+         * (unless concurrency is disabled, which in this case we should
+         * hold the mutex while calling the callback) */
+        if (h->allow_concurrent) {
+            /* concurrency may be changed while we're in the callback, so
+             * save it to a flag.
+             */
+            has_lock = PJ_FALSE;
+            pj_mutex_unlock(h->mutex);
+        } else {
+            has_lock = PJ_TRUE;
+        }
 
         /* Call callback. */
         if (h->cb.on_connect_complete && !IS_CLOSING(h))
             (*h->cb.on_connect_complete)(h, bytes_transfered);
+
+        /* Unlock if we still hold the lock */
+        if (has_lock) {
+            pj_mutex_unlock(h->mutex);
+        }
 
         /* Done. */
@@ -318 +338 @@
             h->fd_type == pj_SOCK_DGRAM()) 
         {
+            pj_bool_t has_lock;
 
             write_op->op = PJ_IOQUEUE_OP_NONE;
@@ -331 +352 @@
             }
 
-            /* No need to hold mutex anymore */
-            pj_mutex_unlock(h->mutex);
+            /* Unlock; from this point we don't need to hold key's mutex
+             * (unless concurrency is disabled, which in this case we should
+             * hold the mutex while calling the callback) */
+            if (h->allow_concurrent) {
+                /* concurrency may be changed while we're in the callback, so
+                 * save it to a flag.
+                 */
+                has_lock = PJ_FALSE;
+                pj_mutex_unlock(h->mutex);
+            } else {
+                has_lock = PJ_TRUE;
+            }
 
             /* Call callback. */
@@ -340 +371 @@
                                            write_op->written);
             }
+
+            if (has_lock) {
+                pj_mutex_unlock(h->mutex);
+            }
 
         } else {
@@ -372 +407 @@
 
         struct accept_operation *accept_op;
+        pj_bool_t has_lock;
         
         /* Get one accept operation from the list. */
@@ -390 +426 @@
         }
 
-        /* Unlock; from this point we don't need to hold key's mutex. */
-        pj_mutex_unlock(h->mutex);
+        /* Unlock; from this point we don't need to hold key's mutex
+         * (unless concurrency is disabled, which in this case we should
+         * hold the mutex while calling the callback) */
+        if (h->allow_concurrent) {
+            /* concurrency may be changed while we're in the callback, so
+             * save it to a flag.
+             */
+            has_lock = PJ_FALSE;
+            pj_mutex_unlock(h->mutex);
+        } else {
+            has_lock = PJ_TRUE;
+        }
 
         /* Call callback. */
@@ -400 +446 @@
         }
 
+        if (has_lock) {
+            pj_mutex_unlock(h->mutex);
+        }
     }
     else
@@ -406 +455 @@
         struct read_operation *read_op;
         pj_ssize_t bytes_read;
+        pj_bool_t has_lock;
 
         /* Get one pending read operation from the list. */
@@ -480 +530 @@
         }
 
-        /* Unlock; from this point we don't need to hold key's mutex. */
-        pj_mutex_unlock(h->mutex);
+        /* Unlock; from this point we don't need to hold key's mutex
+         * (unless concurrency is disabled, which in this case we should
+         * hold the mutex while calling the callback) */
+        if (h->allow_concurrent) {
+            /* concurrency may be changed while we're in the callback, so
+             * save it to a flag.
+             */
+            has_lock = PJ_FALSE;
+            pj_mutex_unlock(h->mutex);
+        } else {
+            has_lock = PJ_TRUE;
+        }
 
         /* Call callback. */
@@ -489 +549 @@
                                       bytes_read);
         }
+
+        if (has_lock) {
+            pj_mutex_unlock(h->mutex);
+        }
 
     } else {
@@ -504 +568 @@
                                        pj_ioqueue_key_t *h )
 {
+    pj_bool_t has_lock;
+
     pj_mutex_lock(h->mutex);
 
@@ -526 +592 @@
     ioqueue_remove_from_set(ioqueue, h, EXCEPTION_EVENT);
 
-    pj_mutex_unlock(h->mutex);
+    /* Unlock; from this point we don't need to hold key's mutex
+     * (unless concurrency is disabled, which in this case we should
+     * hold the mutex while calling the callback) */
+    if (h->allow_concurrent) {
+        /* concurrency may be changed while we're in the callback, so
+         * save it to a flag.
+         */
+        has_lock = PJ_FALSE;
+        pj_mutex_unlock(h->mutex);
+    } else {
+        has_lock = PJ_TRUE;
+    }
 
     /* Call callback. */
@@ -542 +619 @@
 
         (*h->cb.on_connect_complete)(h, status);
+    }
+
+    if (has_lock) {
+        pj_mutex_unlock(h->mutex);
     }
 }
@@ -1097 +1178 @@
 }
 
+PJ_DEF(pj_status_t) pj_ioqueue_set_default_concurrency( pj_ioqueue_t *ioqueue,
+                                                        pj_bool_t allow)
+{
+    PJ_ASSERT_RETURN(ioqueue != NULL, PJ_EINVAL);
+    ioqueue->default_concurrency = allow;
+    return PJ_SUCCESS;
+}
+
+
+PJ_DEF(pj_status_t) pj_ioqueue_set_concurrency(pj_ioqueue_key_t *key,
+                                               pj_bool_t allow)
+{
+    PJ_ASSERT_RETURN(key, PJ_EINVAL);
+
+    /* PJ_IOQUEUE_HAS_SAFE_UNREG must be enabled if concurrency is
+     * disabled.
+     */
+    PJ_ASSERT_RETURN(allow || PJ_IOQUEUE_HAS_SAFE_UNREG, PJ_EINVAL);
+
+    key->allow_concurrent = allow;
+    return PJ_SUCCESS;
+}
+
+PJ_DEF(pj_status_t) pj_ioqueue_lock_key(pj_ioqueue_key_t *key)
+{
+    return pj_mutex_lock(key->mutex);
+}
+
+PJ_DEF(pj_status_t) pj_ioqueue_unlock_key(pj_ioqueue_key_t *key)
+{
+    return pj_mutex_unlock(key->mutex);
+}
+