Avoid extra taskq_dispatch() calls by DMU.

DMU sync code calls taskq_dispatch() for each sublist of os_dirty_dnodes
and os_synced_dnodes.  Since the number of sublists by default is equal
to number of CPUs, it will dispatch equal, potentially large, number of
tasks, waking up many CPUs to handle them, even if only one or few of
sublists actually have any work to do.

This change adds very quick lockless check for empty sublists to avoid
this.  Locking is not needed there, since the multilists are stable at
that time.

I also removed existing locking from multilist_is_empty(), since it any
way provides no any guaranties, unless multilist is stable, in which
case locking is not needed.  Even if the multilist is not stable, it
is not critical for the list implementation, since list_is_empty() only
compares one variable with constant, and the worst that can happen is
it read some stale value, that may happen any way.

Signed-off-by:  Alexander Motin <mav@FreeBSD.org>

include/sys/multilist.h

@@ -80,6 +80,7 @@ int  multilist_is_empty(multilist_t *);
 
 unsigned int multilist_get_num_sublists(multilist_t *);
 unsigned int multilist_get_random_index(multilist_t *);
+int multilist_sublist_is_empty(multilist_t *, unsigned int);
 
 multilist_sublist_t *multilist_sublist_lock(multilist_t *, unsigned int);
 multilist_sublist_t *multilist_sublist_lock_obj(multilist_t *, void *);

module/zfs/dmu_objset.c

@@ -1789,6 +1789,8 @@ dmu_objset_sync(objset_t *os, zio_t *pio, dmu_tx_t *tx)
 
 	for (int i = 0;
 	    i < multilist_get_num_sublists(os->os_dirty_dnodes[txgoff]); i++) {
+		if (multilist_sublist_is_empty(os->os_dirty_dnodes[txgoff], i))
+			continue;
 		sync_dnodes_arg_t *sda = kmem_alloc(sizeof (*sda), KM_SLEEP);
 		sda->sda_list = os->os_dirty_dnodes[txgoff];
 		sda->sda_sublist_idx = i;
@@ -2127,6 +2129,8 @@ dmu_objset_do_userquota_updates(objset_t *os, dmu_tx_t *tx)
 
 	for (int i = 0;
 	    i < multilist_get_num_sublists(os->os_synced_dnodes); i++) {
+		if (multilist_sublist_is_empty(os->os_synced_dnodes, i))
+			continue;
 		userquota_updates_arg_t *uua =
 		    kmem_alloc(sizeof (*uua), KM_SLEEP);
 		uua->uua_os = os;

module/zfs/multilist.c

@@ -226,38 +226,20 @@ multilist_remove(multilist_t *ml, void *obj)
  * Check to see if this multilist object is empty.
  *
  * This will return TRUE if it finds all of the sublists of this
- * multilist to be empty, and FALSE otherwise. Each sublist lock will be
- * automatically acquired as necessary.
+ * multilist to be empty, and FALSE otherwise.
  *
- * If concurrent insertions and removals are occurring, the semantics
- * of this function become a little fuzzy. Instead of locking all
- * sublists for the entire call time of the function, each sublist is
- * only locked as it is individually checked for emptiness. Thus, it's
- * possible for this function to return TRUE with non-empty sublists at
- * the time the function returns. This would be due to another thread
- * inserting into a given sublist, after that specific sublist was check
- * and deemed empty, but before all sublists have been checked.
+ * Result of this function can be meaningful only if either the multilist is
+ * know to be static at the time or all of its sublists are locked by caller.
+ * For this reason there is no locking in this function.
  */
 int
 multilist_is_empty(multilist_t *ml)
 {
 	for (int i = 0; i < ml->ml_num_sublists; i++) {
 		multilist_sublist_t *mls = &ml->ml_sublists[i];
-		/* See comment in multilist_insert(). */
-		boolean_t need_lock = !MUTEX_HELD(&mls->mls_lock);
-
-		if (need_lock)
-			mutex_enter(&mls->mls_lock);
-
-		if (!list_is_empty(&mls->mls_list)) {
-			if (need_lock)
-				mutex_exit(&mls->mls_lock);
 
+		if (!list_is_empty(&mls->mls_list))
 			return (FALSE);
-		}
-
-		if (need_lock)
-			mutex_exit(&mls->mls_lock);
 	}
 
 	return (TRUE);
@@ -277,6 +259,22 @@ multilist_get_random_index(multilist_t *ml)
 	return (spa_get_random(ml->ml_num_sublists));
 }
 
+/*
+ * Check to see if the specified sublist is empty.
+ *
+ * Result of this function can be meaningful only if either the sublist is
+ * know to be static at the time or locked by caller.
+ */
+int
+multilist_sublist_is_empty(multilist_t *ml, unsigned int sublist_idx)
+{
+	multilist_sublist_t *mls;
+
+	ASSERT3U(sublist_idx, <, ml->ml_num_sublists);
+	mls = &ml->ml_sublists[sublist_idx];
+	return (list_is_empty(&mls->mls_list));
+}
+
 /* Lock and return the sublist specified at the given index */
 multilist_sublist_t *
 multilist_sublist_lock(multilist_t *ml, unsigned int sublist_idx)