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

Timestamp:

Feb 13, 2008 3:17:28 PM (17 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_winnt.c (modified) (23 diffs)

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

@@ -115 +115 @@
     enum handle_type    hnd_type;
     pj_ioqueue_callback cb;
+    pj_bool_t           allow_concurrent;
 
 #if PJ_HAS_TCP
@@ -124 +125 @@
     pj_bool_t           closing;
     pj_time_val         free_time;
+    pj_mutex_t         *mutex;
 #endif
 
@@ -136 +138 @@
     pj_lock_t        *lock;
     pj_bool_t         auto_delete_lock;
+    pj_bool_t         default_concurrency;
 
 #if PJ_IOQUEUE_HAS_SAFE_UNREG
@@ -152 +155 @@
 #endif
 };
+
+
+#if PJ_IOQUEUE_HAS_SAFE_UNREG
+/* Prototype */
+static void scan_closing_keys(pj_ioqueue_t *ioqueue);
+#endif
 
 
@@ -316 +325 @@
 
     /* Create IOCP mutex */
-    rc = pj_lock_create_simple_mutex(pool, NULL, &ioqueue->lock);
+    rc = pj_lock_create_recursive_mutex(pool, NULL, &ioqueue->lock);
     if (rc != PJ_SUCCESS) {
         CloseHandle(ioqueue->iocp);
@@ -323 +332 @@
 
     ioqueue->auto_delete_lock = PJ_TRUE;
+    ioqueue->default_concurrency = PJ_IOQUEUE_DEFAULT_ALLOW_CONCURRENCY;
 
 #if PJ_IOQUEUE_HAS_SAFE_UNREG
@@ -345 +355 @@
             while (key != &ioqueue->free_list) {
                 pj_atomic_destroy(key->ref_count);
+                pj_mutex_destroy(key->mutex);
                 key = key->next;
             }
@@ -351 +362 @@
         }
 
+        rc = pj_mutex_create_recursive(pool, "ioqkey", &key->mutex);
+        if (rc != PJ_SUCCESS) {
+            pj_atomic_destroy(key->ref_count);
+            key = ioqueue->free_list.next;
+            while (key != &ioqueue->free_list) {
+                pj_atomic_destroy(key->ref_count);
+                pj_mutex_destroy(key->mutex);
+                key = key->next;
+            }
+            CloseHandle(ioqueue->iocp);
+            return rc;
+        }
+
         pj_list_push_back(&ioqueue->free_list, key);
-
     }
 #endif
@@ -393 +416 @@
     while (key != &ioqueue->active_list) {
         pj_atomic_destroy(key->ref_count);
+        pj_mutex_destroy(key->mutex);
         key = key->next;
     }
@@ -399 +423 @@
     while (key != &ioqueue->closing_list) {
         pj_atomic_destroy(key->ref_count);
+        pj_mutex_destroy(key->mutex);
         key = key->next;
     }
@@ -405 +430 @@
     while (key != &ioqueue->free_list) {
         pj_atomic_destroy(key->ref_count);
+        pj_mutex_destroy(key->mutex);
         key = key->next;
     }
@@ -412 +438 @@
         pj_lock_destroy(ioqueue->lock);
 
+    return PJ_SUCCESS;
+}
+
+
+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;
 }
@@ -454 +489 @@
 
 #if PJ_IOQUEUE_HAS_SAFE_UNREG
+    /* Scan closing list first to release unused keys.
+     * Must do this with lock acquired.
+     */
+    scan_closing_keys(ioqueue);
+
     /* If safe unregistration is used, then get the key record from
      * the free list.
@@ -482 +522 @@
     pj_memcpy(&rec->cb, cb, sizeof(pj_ioqueue_callback));
 
+    /* Set concurrency for this handle */
+    rc = pj_ioqueue_set_concurrency(rec, ioqueue->default_concurrency);
+    if (rc != PJ_SUCCESS) {
+        pj_lock_release(ioqueue->lock);
+        return rc;
+    }
+
 #if PJ_HAS_TCP
     rec->connecting = 0;
@@ -586 +633 @@
      */
     if (pOv) {
+        pj_bool_t has_lock;
+
         /* Event was dequeued for either successfull or failed I/O */
         key = (pj_ioqueue_key_t*)dwKey;
@@ -601 +650 @@
             return PJ_TRUE;
 
+        /* If concurrency is disabled, lock the key 
+         * (and save the lock status to local var since app may change
+         * concurrency setting while in the callback) */
+        if (key->allow_concurrent == PJ_FALSE) {
+            pj_mutex_lock(key->mutex);
+            has_lock = PJ_TRUE;
+        } else {
+            has_lock = PJ_FALSE;
+        }
+
+        /* Now that we get the lock, check again that key is not closing */
+        if (key->closing) {
+            if (has_lock) {
+                pj_mutex_unlock(key->mutex);
+            }
+            return PJ_TRUE;
+        }
+
         /* Increment reference counter to prevent this key from being
          * deleted
          */
         pj_atomic_inc(key->ref_count);
+#else
+        PJ_UNUSED_ARG(has_lock);
 #endif
 
@@ -655 +724 @@
 #if PJ_IOQUEUE_HAS_SAFE_UNREG
         decrement_counter(key);
+        if (has_lock)
+            pj_mutex_unlock(key->mutex);
 #endif
 
@@ -670 +741 @@
 {
     unsigned i;
+    pj_bool_t has_lock;
     enum { RETRY = 10 };
 
@@ -697 +769 @@
     /* Mark key as closing before closing handle. */
     key->closing = 1;
+
+    /* If concurrency is disabled, wait until the key has finished
+     * processing the callback
+     */
+    if (key->allow_concurrent == PJ_FALSE) {
+        pj_mutex_lock(key->mutex);
+        has_lock = PJ_TRUE;
+    } else {
+        has_lock = PJ_FALSE;
+    }
+#else
+    PJ_UNUSED_ARG(has_lock);
 #endif
     
@@ -718 +802 @@
      * Forcing context switch seems to have fixed that, but this is quite
      * an ugly solution..
+     *
+     * Update 2008/02/13:
+     *  This should not happen if concurrency is disallowed for the key.
+     *  So at least application has a solution for this (i.e. by disallowing
+     *  concurrency in the key).
      */
     //This will loop forever if unregistration is done on the callback.
@@ -729 +818 @@
     /* Decrement reference counter to destroy the key. */
     decrement_counter(key);
+
+    if (has_lock)
+        pj_mutex_unlock(key->mutex);
 #endif
 
     return PJ_SUCCESS;
 }
+
+#if PJ_IOQUEUE_HAS_SAFE_UNREG
+/* Scan the closing list, and put pending closing keys to free list.
+ * Must do this with ioqueue mutex held.
+ */
+static void scan_closing_keys(pj_ioqueue_t *ioqueue)
+{
+    if (!pj_list_empty(&ioqueue->closing_list)) {
+        pj_time_val now;
+        pj_ioqueue_key_t *key;
+
+        pj_gettimeofday(&now);
+        
+        /* Move closing keys to free list when they've finished the closing
+         * idle time.
+         */
+        key = ioqueue->closing_list.next;
+        while (key != &ioqueue->closing_list) {
+            pj_ioqueue_key_t *next = key->next;
+
+            pj_assert(key->closing != 0);
+
+            if (PJ_TIME_VAL_GTE(now, key->free_time)) {
+                pj_list_erase(key);
+                pj_list_push_back(&ioqueue->free_list, key);
+            }
+            key = next;
+        }
+    }
+}
+#endif
 
 /*
@@ -767 +890 @@
     /* Check the closing keys only when there's no activity and when there are
      * pending closing keys.
-     * blp:
-     *  no, always check the list. Otherwise on busy activity, this will cause
-     *  ioqueue to reject new registration.
-     */
-    if (/*event_count == 0 &&*/ !pj_list_empty(&ioqueue->closing_list)) {
-        pj_time_val now;
-        pj_ioqueue_key_t *key;
-
-        pj_gettimeofday(&now);
-        
-        /* Move closing keys to free list when they've finished the closing
-         * idle time.
-         */
+     */
+    if (event_count == 0 && !pj_list_empty(&ioqueue->closing_list)) {
         pj_lock_acquire(ioqueue->lock);
-        key = ioqueue->closing_list.next;
-        while (key != &ioqueue->closing_list) {
-            pj_ioqueue_key_t *next = key->next;
-
-            pj_assert(key->closing != 0);
-
-            if (PJ_TIME_VAL_GTE(now, key->free_time)) {
-                pj_list_erase(key);
-                pj_list_push_back(&ioqueue->free_list, key);
-            }
-            key = next;
-        }
+        scan_closing_keys(ioqueue);
         pj_lock_release(ioqueue->lock);
     }
@@ -1269 +1370 @@
 }
 
+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)
+{
+#if PJ_IOQUEUE_HAS_SAFE_UNREG
+    return pj_mutex_lock(key->mutex);
+#else
+    PJ_ASSERT_RETURN(!"PJ_IOQUEUE_HAS_SAFE_UNREG is disabled", PJ_EINVALIDOP);
+#endif
+}
+
+PJ_DEF(pj_status_t) pj_ioqueue_unlock_key(pj_ioqueue_key_t *key)
+{
+#if PJ_IOQUEUE_HAS_SAFE_UNREG
+    return pj_mutex_unlock(key->mutex);
+#else
+    PJ_ASSERT_RETURN(!"PJ_IOQUEUE_HAS_SAFE_UNREG is disabled", PJ_EINVALIDOP);
+#endif
+}
+