Index: /pjproject/trunk/pjlib/include/pj/config.h
===================================================================
--- /pjproject/trunk/pjlib/include/pj/config.h	(revision 1788)
+++ /pjproject/trunk/pjlib/include/pj/config.h	(revision 1789)
@@ -286,4 +286,6 @@
 #   undef PJ_EXCEPTION_USE_WIN32_SEH
 #   undef PJ_HAS_ERROR_STRING
+
+#   define PJ_HAS_IPV6	1
 #endif
 
@@ -510,4 +512,30 @@
 #ifndef PJ_IOQUEUE_HAS_SAFE_UNREG
 #   define PJ_IOQUEUE_HAS_SAFE_UNREG	1
+#endif
+
+
+/**
+ * Default concurrency setting for sockets/handles registered to ioqueue.
+ * This controls whether the ioqueue is allowed to call the key's callback
+ * concurrently/in parallel. The default is yes, which means that if there
+ * are more than one pending operations complete simultaneously, more
+ * than one threads may call the key's callback at the same time. This
+ * generally would promote good scalability for application, at the 
+ * expense of more complexity to manage the concurrent accesses.
+ *
+ * Please see the ioqueue documentation for more info.
+ */
+#ifndef PJ_IOQUEUE_DEFAULT_ALLOW_CONCURRENCY
+#   define PJ_IOQUEUE_DEFAULT_ALLOW_CONCURRENCY   1
+#endif
+
+
+/* Sanity check:
+ *  if ioqueue concurrency is disallowed, PJ_IOQUEUE_HAS_SAFE_UNREG
+ *  must be enabled.
+ */
+#if (PJ_IOQUEUE_DEFAULT_ALLOW_CONCURRENCY==0) && (PJ_IOQUEUE_HAS_SAFE_UNREG==0)
+#   error PJ_IOQUEUE_HAS_SAFE_UNREG must be enabled if ioqueue concurrency \
+	  is disabled
 #endif
 
Index: /pjproject/trunk/pjlib/include/pj/ioqueue.h
===================================================================
--- /pjproject/trunk/pjlib/include/pj/ioqueue.h	(revision 1788)
+++ /pjproject/trunk/pjlib/include/pj/ioqueue.h	(revision 1789)
@@ -102,13 +102,69 @@
  * \section pj_ioqueue_concurrency_sec Concurrency Rules
  *
- * The items below describe rules that must be obeyed when using the I/O 
- * queue, with regard to concurrency:
- *  - simultaneous operations (by different threads) to different key is safe.
- *  - simultaneous operations to the same key is also safe, except
- *    <b>unregistration</b>, which is described below.
- *  - <b>care must be taken when unregistering a key</b> from the
+ * The ioqueue has been fine tuned to allow multiple threads to poll the
+ * handles simultaneously, to maximize scalability when the application is
+ * running on multiprocessor systems. When more than one threads are polling
+ * the ioqueue and there are more than one handles are signaled, more than
+ * one threads will execute the callback simultaneously to serve the events.
+ * These parallel executions are completely safe when the events happen for
+ * two different handles.
+ *
+ * However, with multithreading, care must be taken when multiple events 
+ * happen on the same handle, or when event is happening on a handle (and 
+ * the callback is being executed) and application is performing 
+ * unregistration to the handle at the same time.
+ *
+ * The treatments of above scenario differ according to the concurrency
+ * setting that are applied to the handle.
+ *
+ * \subsection pj_ioq_concur_set Concurrency Settings for Handles
+ *
+ * Concurrency can be set on per handle (key) basis, by using
+ * #pj_ioqueue_set_concurrency() function. The default key concurrency value 
+ * for the handle is inherited from the key concurrency setting of the ioqueue, 
+ * and the key concurrency setting for the ioqueue can be changed by using
+ * #pj_ioqueue_set_default_concurrency(). The default key concurrency setting 
+ * for ioqueue itself is controlled by compile time setting
+ * PJ_IOQUEUE_DEFAULT_ALLOW_CONCURRENCY.
+ *
+ * Note that this key concurrency setting only controls whether multiple
+ * threads are allowed to operate <b>on the same key</b> at the same time. 
+ * The ioqueue itself always allows multiple threads to enter the ioqeuue at 
+ * the same time, and also simultaneous callback calls to <b>differrent 
+ * keys</b> is always allowed regardless to the key concurrency setting.
+ *
+ * \subsection pj_ioq_parallel Parallel Callback Executions for the Same Handle
+ *
+ * Note that when key concurrency is enabled (i.e. parallel callback calls on
+ * the same key is allowed; this is the default setting), the ioqueue will only
+ * perform simultaneous callback executions on the same key when the key has
+ * invoked multiple pending operations. This could be done for example by
+ * calling #pj_ioqueue_recvfrom() more than once on the same key, each with
+ * the same key but different operation key (pj_ioqueue_op_key_t). With this
+ * scenario, when multiple packets arrive on the key at the same time, more
+ * than one threads may execute the callback simultaneously, each with the
+ * same key but different operation key.
+ *
+ * When there is only one pending operation on the key (e.g. there is only one
+ * #pj_ioqueue_recvfrom() invoked on the key), then events occuring to the
+ * same key will be queued by the ioqueue, thus no simultaneous callback calls
+ * will be performed.
+ *
+ * \subsection pj_ioq_allow_concur Concurrency is Enabled (Default Value)
+ *
+ * The default setting for the ioqueue is to allow multiple threads to
+ * execute callbacks for the same handle/key. This setting is selected to
+ * promote good performance and scalability for application.
+ *
+ * However this setting has a major drawback with regard to synchronization,
+ * and application MUST carefully follow the following guidelines to ensure 
+ * that parallel access to the key does not cause problems:
+ *
+ *  - Always note that callback may be called simultaneously for the same
+ *    key.
+ *  - <b>Care must be taken when unregistering a key</b> from the
  *    ioqueue. Application must take care that when one thread is issuing
- *    an unregistration, other thread is not simultaneously invoking an
- *    operation <b>to the same key</b>.
+ *    an unregistration, other thread is not simultaneously invoking the
+ *    callback <b>to the same key</b>.
  *\n
  *    This happens because the ioqueue functions are working with a pointer
@@ -116,4 +172,15 @@
  *    has been rendered invalid by other threads before the ioqueue has a
  *    chance to acquire mutex on it.
+ *
+ * \subsection pj_ioq_disallow_concur Concurrency is Disabled
+ *
+ * Alternatively, application may disable key concurrency to make 
+ * synchronization easier. As noted above, there are three ways to control
+ * key concurrency setting:
+ *  - by controlling on per handle/key basis, with #pj_ioqueue_set_concurrency().
+ *  - by changing default key concurrency setting on the ioqueue, with
+ *    #pj_ioqueue_set_default_concurrency().
+ *  - by changing the default concurrency on compile time, by declaring
+ *    PJ_IOQUEUE_DEFAULT_ALLOW_CONCURRENCY macro to zero in your config_site.h
  *
  * \section pj_ioqeuue_examples_sec Examples
@@ -292,4 +359,22 @@
 
 /**
+ * Set default concurrency policy for this ioqueue. If this function is not
+ * called, the default concurrency policy for the ioqueue is controlled by 
+ * compile time setting PJ_IOQUEUE_DEFAULT_ALLOW_CONCURRENCY.
+ *
+ * Note that changing the concurrency setting to the ioqueue will only affect
+ * subsequent key registrations. To modify the concurrency setting for
+ * individual key, use #pj_ioqueue_set_concurrency().
+ *
+ * @param ioqueue	The ioqueue instance.
+ * @param allow		Non-zero to allow concurrent callback calls, or
+ *			PJ_FALSE to disallow it.
+ *
+ * @return		PJ_SUCCESS on success or the appropriate error code.
+ */
+PJ_DECL(pj_status_t) pj_ioqueue_set_default_concurrency(pj_ioqueue_t *ioqueue,
+							pj_bool_t allow);
+
+/**
  * Register a socket to the I/O queue framework. 
  * When a socket is registered to the IOQueue, it may be modified to use
@@ -367,4 +452,49 @@
                                                void **old_data);
 
+/**
+ * Configure whether the ioqueue is allowed to call the key's callback
+ * concurrently/in parallel. The default concurrency setting for the key
+ * is controlled by ioqueue's default concurrency value, which can be
+ * changed by calling #pj_ioqueue_set_default_concurrency().
+ *
+ * If concurrency is allowed for the key, it means that if there are more
+ * than one pending operations complete simultaneously, more than one
+ * threads may call the key's  callback at the same time. This generally
+ * would promote good scalability for application, at the expense of more
+ * complexity to manage the concurrent accesses in application's code.
+ *
+ * Alternatively application may disable the concurrent access by
+ * setting the \a allow flag to false. With concurrency disabled, only
+ * one thread can call the key's callback at one time.
+ *
+ * @param key	    The key that was previously obtained from registration.
+ * @param allow	    Set this to non-zero to allow concurrent callback calls
+ *		    and zero (PJ_FALSE) to disallow it.
+ *
+ * @return	    PJ_SUCCESS on success or the appropriate error code.
+ */
+PJ_DECL(pj_status_t) pj_ioqueue_set_concurrency(pj_ioqueue_key_t *key,
+						pj_bool_t allow);
+
+/**
+ * Acquire the key's mutex. When the key's concurrency is disabled, 
+ * application may call this function to synchronize its operation
+ * with the key's callback (i.e. this function will block until the
+ * key's callback returns).
+ *
+ * @param key	    The key that was previously obtained from registration.
+ *
+ * @return	    PJ_SUCCESS on success or the appropriate error code.
+ */
+PJ_DECL(pj_status_t) pj_ioqueue_lock_key(pj_ioqueue_key_t *key);
+
+/**
+ * Release the lock previously acquired with pj_ioqueue_lock_key().
+ *
+ * @param key	    The key that was previously obtained from registration.
+ *
+ * @return	    PJ_SUCCESS on success or the appropriate error code.
+ */
+PJ_DECL(pj_status_t) pj_ioqueue_unlock_key(pj_ioqueue_key_t *key);
 
 /**
Index: /pjproject/trunk/pjlib/src/pj/ioqueue_common_abs.c
===================================================================
--- /pjproject/trunk/pjlib/src/pj/ioqueue_common_abs.c	(revision 1788)
+++ /pjproject/trunk/pjlib/src/pj/ioqueue_common_abs.c	(revision 1789)
@@ -97,4 +97,8 @@
 #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,4 +198,5 @@
 	/* Completion of connect() operation */
 	pj_ssize_t bytes_transfered;
+	pj_bool_t has_lock;
 
 	/* Clear operation. */
@@ -247,10 +252,25 @@
 #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,4 +338,5 @@
             h->fd_type == pj_SOCK_DGRAM()) 
         {
+	    pj_bool_t has_lock;
 
 	    write_op->op = PJ_IOQUEUE_OP_NONE;
@@ -331,6 +352,16 @@
             }
 
-            /* 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,4 +371,8 @@
                                            write_op->written);
             }
+
+	    if (has_lock) {
+		pj_mutex_unlock(h->mutex);
+	    }
 
         } else {
@@ -372,4 +407,5 @@
 
         struct accept_operation *accept_op;
+	pj_bool_t has_lock;
 	
         /* Get one accept operation from the list. */
@@ -390,6 +426,16 @@
 	}
 
-        /* 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,4 +446,7 @@
 	}
 
+	if (has_lock) {
+	    pj_mutex_unlock(h->mutex);
+	}
     }
     else
@@ -406,4 +455,5 @@
         struct read_operation *read_op;
         pj_ssize_t bytes_read;
+	pj_bool_t has_lock;
 
         /* Get one pending read operation from the list. */
@@ -480,6 +530,16 @@
 	}
 
-        /* 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,4 +549,8 @@
                                       bytes_read);
         }
+
+	if (has_lock) {
+	    pj_mutex_unlock(h->mutex);
+	}
 
     } else {
@@ -504,4 +568,6 @@
                                        pj_ioqueue_key_t *h )
 {
+    pj_bool_t has_lock;
+
     pj_mutex_lock(h->mutex);
 
@@ -526,5 +592,16 @@
     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,4 +619,8 @@
 
 	(*h->cb.on_connect_complete)(h, status);
+    }
+
+    if (has_lock) {
+	pj_mutex_unlock(h->mutex);
     }
 }
@@ -1097,2 +1178,35 @@
 }
 
+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);
+}
+
Index: /pjproject/trunk/pjlib/src/pj/ioqueue_common_abs.h
===================================================================
--- /pjproject/trunk/pjlib/src/pj/ioqueue_common_abs.h	(revision 1788)
+++ /pjproject/trunk/pjlib/src/pj/ioqueue_common_abs.h	(revision 1789)
@@ -104,4 +104,5 @@
     pj_bool_t		    inside_callback;	    \
     pj_bool_t		    destroy_requested;	    \
+    pj_bool_t		    allow_concurrent;	    \
     pj_sock_t		    fd;                     \
     int                     fd_type;                \
@@ -117,5 +118,6 @@
 #define DECLARE_COMMON_IOQUEUE                      \
     pj_lock_t          *lock;                       \
-    pj_bool_t           auto_delete_lock;
+    pj_bool_t           auto_delete_lock;	    \
+    pj_bool_t		default_concurrency;
 
 
Index: /pjproject/trunk/pjlib/src/pj/ioqueue_winnt.c
===================================================================
--- /pjproject/trunk/pjlib/src/pj/ioqueue_winnt.c	(revision 1788)
+++ /pjproject/trunk/pjlib/src/pj/ioqueue_winnt.c	(revision 1789)
@@ -115,4 +115,5 @@
     enum handle_type    hnd_type;
     pj_ioqueue_callback	cb;
+    pj_bool_t		allow_concurrent;
 
 #if PJ_HAS_TCP
@@ -124,4 +125,5 @@
     pj_bool_t		closing;
     pj_time_val		free_time;
+    pj_mutex_t	       *mutex;
 #endif
 
@@ -136,4 +138,5 @@
     pj_lock_t        *lock;
     pj_bool_t         auto_delete_lock;
+    pj_bool_t	      default_concurrency;
 
 #if PJ_IOQUEUE_HAS_SAFE_UNREG
@@ -152,4 +155,10 @@
 #endif
 };
+
+
+#if PJ_IOQUEUE_HAS_SAFE_UNREG
+/* Prototype */
+static void scan_closing_keys(pj_ioqueue_t *ioqueue);
+#endif
 
 
@@ -316,5 +325,5 @@
 
     /* 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,4 +332,5 @@
 
     ioqueue->auto_delete_lock = PJ_TRUE;
+    ioqueue->default_concurrency = PJ_IOQUEUE_DEFAULT_ALLOW_CONCURRENCY;
 
 #if PJ_IOQUEUE_HAS_SAFE_UNREG
@@ -345,4 +355,5 @@
 	    while (key != &ioqueue->free_list) {
 		pj_atomic_destroy(key->ref_count);
+		pj_mutex_destroy(key->mutex);
 		key = key->next;
 	    }
@@ -351,6 +362,18 @@
 	}
 
+	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,4 +416,5 @@
     while (key != &ioqueue->active_list) {
 	pj_atomic_destroy(key->ref_count);
+	pj_mutex_destroy(key->mutex);
 	key = key->next;
     }
@@ -399,4 +423,5 @@
     while (key != &ioqueue->closing_list) {
 	pj_atomic_destroy(key->ref_count);
+	pj_mutex_destroy(key->mutex);
 	key = key->next;
     }
@@ -405,4 +430,5 @@
     while (key != &ioqueue->free_list) {
 	pj_atomic_destroy(key->ref_count);
+	pj_mutex_destroy(key->mutex);
 	key = key->next;
     }
@@ -412,4 +438,13 @@
         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,4 +489,9 @@
 
 #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,4 +522,11 @@
     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,4 +633,6 @@
      */
     if (pOv) {
+	pj_bool_t has_lock;
+
 	/* Event was dequeued for either successfull or failed I/O */
 	key = (pj_ioqueue_key_t*)dwKey;
@@ -601,8 +650,28 @@
 	    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,4 +724,6 @@
 #if PJ_IOQUEUE_HAS_SAFE_UNREG
 	decrement_counter(key);
+	if (has_lock)
+	    pj_mutex_unlock(key->mutex);
 #endif
 
@@ -670,4 +741,5 @@
 {
     unsigned i;
+    pj_bool_t has_lock;
     enum { RETRY = 10 };
 
@@ -697,4 +769,16 @@
     /* 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,4 +802,9 @@
      * 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,8 +818,42 @@
     /* 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,30 +890,8 @@
     /* 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,2 +1370,34 @@
 }
 
+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
+}
+
Index: /pjproject/trunk/pjlib/src/pjlib-test/ioq_perf.c
===================================================================
--- /pjproject/trunk/pjlib/src/pjlib-test/ioq_perf.c	(revision 1788)
+++ /pjproject/trunk/pjlib/src/pjlib-test/ioq_perf.c	(revision 1789)
@@ -222,5 +222,6 @@
  *    period of time.
  */
-static int perform_test(int sock_type, const char *type_name,
+static int perform_test(pj_bool_t allow_concur,
+			int sock_type, const char *type_name,
                         unsigned thread_cnt, unsigned sockpair_cnt,
                         pj_size_t buffer_size, 
@@ -261,4 +262,10 @@
     }
 
+    rc = pj_ioqueue_set_default_concurrency(ioqueue, allow_concur);
+    if (rc != PJ_SUCCESS) {
+	app_perror("...error: pj_ioqueue_set_default_concurrency()", rc);
+        return -16;
+    }
+
     /* Initialize each producer-consumer pair. */
     for (i=0; i<sockpair_cnt; ++i) {
@@ -438,8 +445,5 @@
 }
 
-/*
- * main test entry.
- */
-int ioqueue_perf_test(void)
+static int ioqueue_perf_test_imp(pj_bool_t allow_concur)
 {
     enum { BUF_SIZE = 512 };
@@ -501,4 +505,5 @@
 
     PJ_LOG(3,(THIS_FILE, "   Benchmarking %s ioqueue:", pj_ioqueue_name()));
+    PJ_LOG(3,(THIS_FILE, "   Testing with concurency=%d", allow_concur));
     PJ_LOG(3,(THIS_FILE, "   ======================================="));
     PJ_LOG(3,(THIS_FILE, "   Type  Threads  Skt.Pairs      Bandwidth"));
@@ -509,5 +514,6 @@
         pj_size_t bandwidth;
 
-        rc = perform_test(test_param[i].type, 
+        rc = perform_test(allow_concur,
+			  test_param[i].type, 
                           test_param[i].type_name,
                           test_param[i].thread_cnt, 
@@ -538,4 +544,22 @@
 }
 
+/*
+ * main test entry.
+ */
+int ioqueue_perf_test(void)
+{
+    int rc;
+
+    rc = ioqueue_perf_test_imp(PJ_TRUE);
+    if (rc != 0)
+	return rc;
+
+    rc = ioqueue_perf_test_imp(PJ_FALSE);
+    if (rc != 0)
+	return rc;
+
+    return 0;
+}
+
 #else
 /* To prevent warning about "translation unit is empty"
Index: /pjproject/trunk/pjlib/src/pjlib-test/ioq_tcp.c
===================================================================
--- /pjproject/trunk/pjlib/src/pjlib-test/ioq_tcp.c	(revision 1788)
+++ /pjproject/trunk/pjlib/src/pjlib-test/ioq_tcp.c	(revision 1789)
@@ -233,5 +233,5 @@
  * Compliance test for success scenario.
  */
-static int compliance_test_0(void)
+static int compliance_test_0(pj_bool_t allow_concur)
 {
     pj_sock_t ssock=-1, csock0=-1, csock1=-1;
@@ -291,4 +291,11 @@
         app_perror("...ERROR in pj_ioqueue_create()", rc);
 	status=-20; goto on_error;
+    }
+
+    // Concurrency
+    rc = pj_ioqueue_set_default_concurrency(ioque, allow_concur);
+    if (rc != PJ_SUCCESS) {
+        app_perror("...ERROR in pj_ioqueue_set_default_concurrency()", rc);
+	status=-21; goto on_error;
     }
 
@@ -459,5 +466,5 @@
  * In this case, the client connects to a non-existant service.
  */
-static int compliance_test_1(void)
+static int compliance_test_1(pj_bool_t allow_concur)
 {
     pj_sock_t csock1=PJ_INVALID_SOCKET;
@@ -478,4 +485,10 @@
     if (!ioque) {
 	status=-20; goto on_error;
+    }
+
+    // Concurrency
+    rc = pj_ioqueue_set_default_concurrency(ioque, allow_concur);
+    if (rc != PJ_SUCCESS) {
+	status=-21; goto on_error;
     }
 
@@ -582,5 +595,5 @@
  * Repeated connect/accept on the same listener socket.
  */
-static int compliance_test_2(void)
+static int compliance_test_2(pj_bool_t allow_concur)
 {
 #if defined(PJ_SYMBIAN) && PJ_SYMBIAN!=0
@@ -648,4 +661,11 @@
     }
 
+
+    // Concurrency
+    rc = pj_ioqueue_set_default_concurrency(ioque, allow_concur);
+    if (rc != PJ_SUCCESS) {
+        app_perror("...ERROR in pj_ioqueue_set_default_concurrency()", rc);
+	return -11;
+    }
 
     // Allocate buffers for send and receive.
@@ -888,11 +908,13 @@
 
 
-int tcp_ioqueue_test()
+static int tcp_ioqueue_test_impl(pj_bool_t allow_concur)
 {
     int status;
+
+    PJ_LOG(3,(THIS_FILE, "..testing with concurency=%d", allow_concur));
 
     PJ_LOG(3, (THIS_FILE, "..%s compliance test 0 (success scenario)",
 	       pj_ioqueue_name()));
-    if ((status=compliance_test_0()) != 0) {
+    if ((status=compliance_test_0(allow_concur)) != 0) {
 	PJ_LOG(1, (THIS_FILE, "....FAILED (status=%d)\n", status));
 	return status;
@@ -900,5 +922,5 @@
     PJ_LOG(3, (THIS_FILE, "..%s compliance test 1 (failed scenario)",
                pj_ioqueue_name()));
-    if ((status=compliance_test_1()) != 0) {
+    if ((status=compliance_test_1(allow_concur)) != 0) {
 	PJ_LOG(1, (THIS_FILE, "....FAILED (status=%d)\n", status));
 	return status;
@@ -907,8 +929,23 @@
     PJ_LOG(3, (THIS_FILE, "..%s compliance test 2 (repeated accept)",
                pj_ioqueue_name()));
-    if ((status=compliance_test_2()) != 0) {
+    if ((status=compliance_test_2(allow_concur)) != 0) {
 	PJ_LOG(1, (THIS_FILE, "....FAILED (status=%d)\n", status));
 	return status;
     }
+
+    return 0;
+}
+
+int tcp_ioqueue_test()
+{
+    int rc;
+
+    rc = tcp_ioqueue_test_impl(PJ_TRUE);
+    if (rc != 0)
+	return rc;
+
+    rc = tcp_ioqueue_test_impl(PJ_FALSE);
+    if (rc != 0)
+	return rc;
 
     return 0;
Index: /pjproject/trunk/pjlib/src/pjlib-test/ioq_udp.c
===================================================================
--- /pjproject/trunk/pjlib/src/pjlib-test/ioq_udp.c	(revision 1788)
+++ /pjproject/trunk/pjlib/src/pjlib-test/ioq_udp.c	(revision 1789)
@@ -126,5 +126,5 @@
  * data between two sockets.
  */ 
-static int compliance_test(void)
+static int compliance_test(pj_bool_t allow_concur)
 {
     pj_sock_t ssock=-1, csock=-1;
@@ -177,4 +177,11 @@
     if (rc != PJ_SUCCESS) {
 	status=-20; goto on_error;
+    }
+
+    // Set concurrency
+    TRACE_("set concurrency...");
+    rc = pj_ioqueue_set_default_concurrency(ioque, allow_concur);
+    if (rc != PJ_SUCCESS) {
+	status=-21; goto on_error;
     }
 
@@ -352,5 +359,5 @@
  * closed.
  */ 
-static int unregister_test(void)
+static int unregister_test(pj_bool_t allow_concur)
 {
     enum { RPORT = 50000, SPORT = 50001 };
@@ -382,4 +389,11 @@
     }
 
+    // Set concurrency
+    TRACE_("set concurrency...");
+    status = pj_ioqueue_set_default_concurrency(ioqueue, allow_concur);
+    if (status != PJ_SUCCESS) {
+	return -112;
+    }
+
     /* Create sender socket */
     status = app_socket(pj_AF_INET(), pj_SOCK_DGRAM(), 0, SPORT, &ssock);
@@ -513,5 +527,5 @@
  * of sockets to the ioqueue.
  */
-static int many_handles_test(void)
+static int many_handles_test(pj_bool_t allow_concur)
 {
     enum { MAX = PJ_IOQUEUE_MAX_HANDLES };
@@ -538,4 +552,10 @@
 	app_perror("...error in pj_ioqueue_create", rc);
 	return -10;
+    }
+
+    // Set concurrency
+    rc = pj_ioqueue_set_default_concurrency(ioqueue, allow_concur);
+    if (rc != PJ_SUCCESS) {
+	return -11;
     }
 
@@ -601,5 +621,6 @@
  * Benchmarking IOQueue
  */
-static int bench_test(int bufsize, int inactive_sock_count)
+static int bench_test(pj_bool_t allow_concur, int bufsize, 
+		      int inactive_sock_count)
 {
     pj_sock_t ssock=-1, csock=-1;
@@ -649,4 +670,11 @@
     if (rc != PJ_SUCCESS) {
 	app_perror("...error: pj_ioqueue_create()", rc);
+	goto on_error;
+    }
+
+    // Set concurrency
+    rc = pj_ioqueue_set_default_concurrency(ioque, allow_concur);
+    if (rc != PJ_SUCCESS) {
+	app_perror("...error: pj_ioqueue_set_default_concurrency()", rc);
 	goto on_error;
     }
@@ -840,13 +868,15 @@
 }
 
-int udp_ioqueue_test()
+static int udp_ioqueue_test_imp(pj_bool_t allow_concur)
 {
     int status;
     int bufsize, sock_count;
 
+    PJ_LOG(3,(THIS_FILE, "..testing with concurency=%d", allow_concur));
+
     //goto pass1;
 
     PJ_LOG(3, (THIS_FILE, "...compliance test (%s)", pj_ioqueue_name()));
-    if ((status=compliance_test()) != 0) {
+    if ((status=compliance_test(allow_concur)) != 0) {
 	return status;
     }
@@ -855,10 +885,10 @@
 
     PJ_LOG(3, (THIS_FILE, "...unregister test (%s)", pj_ioqueue_name()));
-    if ((status=unregister_test()) != 0) {
+    if ((status=unregister_test(allow_concur)) != 0) {
 	return status;
     }
     PJ_LOG(3, (THIS_FILE, "....unregister test ok"));
 
-    if ((status=many_handles_test()) != 0) {
+    if ((status=many_handles_test(allow_concur)) != 0) {
 	return status;
     }
@@ -880,5 +910,5 @@
 
     for (bufsize=BUF_MIN_SIZE; bufsize <= BUF_MAX_SIZE; bufsize *= 2) {
-	if ((status=bench_test(bufsize, SOCK_INACTIVE_MIN)) != 0)
+	if ((status=bench_test(allow_concur, bufsize, SOCK_INACTIVE_MIN)) != 0)
 	    return status;
     }
@@ -890,7 +920,22 @@
     {
 	//PJ_LOG(3,(THIS_FILE, "...testing with %d fds", sock_count));
-	if ((status=bench_test(bufsize, sock_count-2)) != 0)
+	if ((status=bench_test(allow_concur, bufsize, sock_count-2)) != 0)
 	    return status;
     }
+    return 0;
+}
+
+int udp_ioqueue_test()
+{
+    int rc;
+
+    rc = udp_ioqueue_test_imp(PJ_TRUE);
+    if (rc != 0)
+	return rc;
+
+    rc = udp_ioqueue_test_imp(PJ_FALSE);
+    if (rc != 0)
+	return rc;
+
     return 0;
 }
Index: /pjproject/trunk/pjlib/src/pjlib-test/ioq_unreg.c
===================================================================
--- /pjproject/trunk/pjlib/src/pjlib-test/ioq_unreg.c	(revision 1788)
+++ /pjproject/trunk/pjlib/src/pjlib-test/ioq_unreg.c	(revision 1789)
@@ -287,5 +287,5 @@
 }
 
-int udp_ioqueue_unreg_test(void)
+static int udp_ioqueue_unreg_test_imp(pj_bool_t allow_concur)
 {
     enum { LOOP = 10 };
@@ -294,5 +294,7 @@
     pj_ioqueue_t *ioqueue;
     pj_pool_t *test_pool;
-			      
+	
+    PJ_LOG(3,(THIS_FILE, "..testing with concurency=%d", allow_concur));
+
     test_method = UNREGISTER_IN_APP;
 
@@ -305,4 +307,9 @@
     }
 
+    rc = pj_ioqueue_set_default_concurrency(ioqueue, allow_concur);
+    if (rc != PJ_SUCCESS) {
+	app_perror("Error in pj_ioqueue_set_default_concurrency()", rc);
+	return -12;
+    }
 
     PJ_LOG(3, (THIS_FILE, "...ioqueue unregister stress test 0/3 (%s)", 
@@ -352,5 +359,18 @@
 }
 
-
+int udp_ioqueue_unreg_test(void)
+{
+    int rc;
+
+    rc = udp_ioqueue_unreg_test_imp(PJ_TRUE);
+    if (rc != 0)
+	return rc;
+
+    rc = udp_ioqueue_unreg_test_imp(PJ_FALSE);
+    if (rc != 0)
+	return rc;
+
+    return 0;
+}
 
 #else
