Fix lock order inversion in ioqueue connect/write callbacks and handle clear_key() from callback

pjlib-util/src/pjlib-util/http_client.c

@@ -1148,6 +1148,31 @@ static pj_status_t auth_respond_basic(pj_http_req *hreq)
     pj_status_t status;
     int len;
 
+    /* Ensure the combined length of username, colon, and password doesn't
+     * exceed buffer size.
+     */
+    if (hreq->param.auth_cred.username.slen < 0 ||
+        hreq->param.auth_cred.data.slen < 0)
+    {
+        TRACE_((THIS_FILE,
+                "Error: username and/or password length is negative"));
+        return PJ_ETOOBIG;
+    }
+
+    do {
+        pj_size_t user_len =
+            (pj_size_t)hreq->param.auth_cred.username.slen;
+        pj_size_t pass_len =
+            (pj_size_t)hreq->param.auth_cred.data.slen;
+
+        if (user_len >= BUF_SIZE ||
+            pass_len > (pj_size_t)BUF_SIZE - 1 - user_len)
+        {
+            TRACE_((THIS_FILE,
+                    "Error: username and password exceed buffer size"));
+            return PJ_ETOOBIG;
+        }
+    } while (0);
     /* Use send buffer to store userid ":" password */
     user_pass.ptr = hreq->buffer.ptr;
     pj_strcpy(&user_pass, &hreq->param.auth_cred.username);

pjlib/src/pj/ioqueue_common_abs.c

@@ -94,7 +94,9 @@ static pj_status_t ioqueue_init_key( pj_pool_t *pool,
     key->connecting = 0;
 #endif
     pj_list_init(&key->read_cb_list);
+    pj_list_init(&key->write_cb_list);
     key->read_callback_thread = NULL;
+    key->write_callback_thread = NULL;
     key->closing = 0;
 
     /* Save callback. */
@@ -191,6 +193,52 @@ PJ_INLINE(int) key_has_pending_connect(pj_ioqueue_key_t *key)
 #define IS_CLOSING(key) (key->closing)
 
 
+#if PJ_IOQUEUE_CALLBACK_NO_LOCK
+static unsigned ioqueue_dispatch_write_event_no_lock(pj_ioqueue_key_t* h,
+                                                     unsigned max_event)
+{
+    unsigned event_cnt = 0;
+
+    while (1) {
+        struct write_operation *write_op = NULL;
+        void (*on_write_complete)(pj_ioqueue_key_t *key,
+                                  pj_ioqueue_op_key_t *op_key,
+                                  pj_ssize_t bytes_sent);
+
+        /* Check if there is any pending write callback for this key. */
+        pj_ioqueue_lock_key(h);
+
+        if (!IS_CLOSING(h) && !pj_list_empty(&h->write_cb_list) &&
+            (max_event == 0 || event_cnt < max_event))
+        {
+            write_op = h->write_cb_list.next;
+            pj_list_erase(write_op);
+            on_write_complete = h->cb.on_write_complete;
+        } else {
+            /* No more pending callback or maximum event number is reached.
+             * Clear the callback thread and return.
+             */
+            h->write_callback_thread = NULL;
+            on_write_complete = NULL;
+        }
+
+        pj_ioqueue_unlock_key(h);
+        PJ_RACE_ME(5);
+
+        /* Invoke the callback or return */
+        if (on_write_complete) {
+            (*on_write_complete)(h, (pj_ioqueue_op_key_t*)write_op,
+                                write_op->written);
+            ++event_cnt;
+        } else {
+            break;
+        }
+    }
+
+    return event_cnt;
+}
+#endif
+
 /*
  * ioqueue_dispatch_event()
  *
@@ -279,7 +327,16 @@ static pj_bool_t ioqueue_dispatch_write_event( pj_ioqueue_t *ioqueue,
             has_lock = PJ_FALSE;
             pj_ioqueue_unlock_key(h);
         } else {
+#if PJ_IOQUEUE_CALLBACK_NO_LOCK
+            /* Do not hold mutex while invoking callback to avoid
+             * lock order inversion with application locks.
+             */
+            has_lock = PJ_FALSE;
+            pj_ioqueue_unlock_key(h);
+            PJ_RACE_ME(5);
+#else
             has_lock = PJ_TRUE;
+#endif
         }
 
         /* Call callback. */
@@ -398,7 +455,33 @@ static pj_bool_t ioqueue_dispatch_write_event( pj_ioqueue_t *ioqueue,
                 pj_ioqueue_unlock_key(h);
                 PJ_RACE_ME(5);
             } else {
+#if PJ_IOQUEUE_CALLBACK_NO_LOCK
+                /* If we're not allowing concurrency, we must prevent
+                 * re-entrancy in the callback.
+                 */
+                if (h->write_callback_thread) {
+                    /* Another thread is in the write callback for this key,
+                     * just queue this write_op, that thread will invoke the
+                     * callback later.
+                     */
+                    pj_list_push_back(&h->write_cb_list, write_op);
+                    pj_ioqueue_unlock_key(h);
+                    return PJ_TRUE;
+                }
+
+                /* Save the thread invoking the write callback.
+                 * Note that when threading is disabled or concurrency is allowed,
+                 * this will always be NULL.
+                 */
+                h->write_callback_thread = pj_thread_this();
+
+                /* Do not hold mutex while invoking callback */
+                has_lock = PJ_FALSE;
+                pj_ioqueue_unlock_key(h);
+                PJ_RACE_ME(5);
+#else
                 has_lock = PJ_TRUE;
+#endif
             }
 
             /* Call callback. */
@@ -412,6 +495,11 @@ static pj_bool_t ioqueue_dispatch_write_event( pj_ioqueue_t *ioqueue,
                 pj_ioqueue_unlock_key(h);
             }
 
+#if PJ_IOQUEUE_CALLBACK_NO_LOCK
+            /* If we have more pending write callback, process it now */
+            ioqueue_dispatch_write_event_no_lock(h, 0);
+#endif
+
         } else {
             pj_ioqueue_unlock_key(h);
         }
@@ -754,7 +842,16 @@ static pj_bool_t ioqueue_dispatch_exception_event( pj_ioqueue_t *ioqueue,
         pj_ioqueue_unlock_key(h);
         PJ_RACE_ME(5);
     } else {
+#if PJ_IOQUEUE_CALLBACK_NO_LOCK
+        /* Do not hold mutex while invoking callback to avoid
+         * lock order inversion with application locks.
+         */
+        has_lock = PJ_FALSE;
+        pj_ioqueue_unlock_key(h);
+        PJ_RACE_ME(5);
+#else
         has_lock = PJ_TRUE;
+#endif
     }
 
     /* Call callback. */
@@ -1459,8 +1556,69 @@ PJ_DEF(pj_status_t) pj_ioqueue_clear_key( pj_ioqueue_key_t *key )
     pj_list_init(&key->write_list);
     pj_list_init(&key->accept_list);
     pj_list_init(&key->read_cb_list);
+<<<<<<< HEAD
+    pj_list_init(&key->write_cb_list);
+=======
+>>>>>>> 9b0a4e83fb1663a0c05f2186c147aff9e063a4b5
+
+#if PJ_IOQUEUE_CALLBACK_NO_LOCK
+    /* Wait until any read callback is finished */
+    do {
+        unsigned counter = 0;
+
+        while (key->read_callback_thread &&
+               key->read_callback_thread != pj_thread_this())
+        {
+            /* Callback is running, unlock while waiting, since the callback
+             * may need the lock.
+             */
+            pj_ioqueue_unlock_key(key);
+            pj_thread_sleep(10);
+            pj_ioqueue_lock_key(key);
+
+            /* Clear read pending list again */
+            pj_list_init(&key->read_list);
+
+            /* Timeout after ~1 second */
+            if (++counter > 100) {
+                PJ_LOG(1,(THIS_FILE, "Timeout waiting for read callback "
+                                     "to finish on socket=%ld", key->fd));
+                break;
+            }
+        }
+    } while (0);
+<<<<<<< HEAD
+
+    /* Wait until any write callback is finished */
+    do {
+        unsigned counter = 0;
+
+        while (key->write_callback_thread &&
+               key->write_callback_thread != pj_thread_this())
+        {
+            /* Callback is running, unlock while waiting, since the callback
+             * may need the lock.
+             */
+            pj_ioqueue_unlock_key(key);
+            pj_thread_sleep(10);
+            pj_ioqueue_lock_key(key);
+
+            /* Clear write pending list again */
+            pj_list_init(&key->write_list);
+
+            /* Timeout after ~1 second */
+            if (++counter > 100) {
+                PJ_LOG(1,(THIS_FILE, "Timeout waiting for write callback "
+                                     "to finish on socket=%ld", key->fd));
+                break;
+            }
+        }
+    } while (0);
+=======
+>>>>>>> 9b0a4e83fb1663a0c05f2186c147aff9e063a4b5
+#endif
+
     key->connecting = 0;
-    key->read_callback_thread = NULL;
 
     /* Remove key from sets */
     ioqueue_remove_from_set2(key->ioqueue, key,

pjlib/src/pj/ioqueue_common_abs.h

@@ -105,6 +105,7 @@ union operation_key
     pj_lock_t              *lock;                   \
     pj_bool_t               closing;                \
     pj_thread_t            *read_callback_thread;   \
+    pj_thread_t            *write_callback_thread;  \
     pj_bool_t               destroy_requested;      \
     pj_bool_t               allow_concurrent;       \
     pj_sock_t               fd;                     \
@@ -116,6 +117,7 @@ union operation_key
     struct write_operation  write_list;             \
     struct accept_operation accept_list;            \
     struct read_operation   read_cb_list;           \
+    struct write_operation  write_cb_list;          \
     UNREG_FIELDS