Make the spa_config_lock reader fast path lockless

Readers took scl_lock on every spa_config_enter()/exit(), which cost
~29% of CPU time in 4KB record write workloads.  Fold the writer gate
into the reference count: SCL_COUNT_WRITER in the top bit of scl_count
lets a reader take a reference and test for a writer in one atomic
operation, and lets the last reader detect a waiting writer without a
second load.  Writers still serialize on scl_lock, and priority readers
still take it, since they must tell a waiting writer from a holding one.

Reviewed-by: Brian Behlendorf <[email protected]>
Signed-off-by: Alexander Motin <[email protected]>
Closes #19131

include/os/freebsd/spl/sys/atomic.h

@@ -54,6 +54,10 @@ extern void atomic_dec_64(volatile uint64_t *target);
 extern uint64_t atomic_swap_64(volatile uint64_t *a, uint64_t value);
 extern uint64_t atomic_load_64(volatile uint64_t *a);
 extern uint64_t atomic_add_64_nv(volatile uint64_t *target, int64_t delta);
+extern void atomic_or_64(volatile uint64_t *target, uint64_t bits);
+extern void atomic_and_64(volatile uint64_t *target, uint64_t bits);
+extern uint64_t atomic_or_64_nv(volatile uint64_t *target, uint64_t bits);
+extern uint64_t atomic_and_64_nv(volatile uint64_t *target, uint64_t bits);
 extern uint64_t atomic_cas_64(volatile uint64_t *target, uint64_t cmp,
     uint64_t newval);
 #endif
@@ -98,6 +102,38 @@ atomic_dec_32_nv(volatile uint32_t *target)
 	return (atomic_add_32_nv(target, -1));
 }
 
+static __inline void
+atomic_or_32(volatile uint32_t *target, uint32_t bits)
+{
+	atomic_set_32(target, bits);
+}
+
+static __inline void
+atomic_and_32(volatile uint32_t *target, uint32_t bits)
+{
+	atomic_clear_32(target, ~bits);
+}
+
+static __inline uint32_t
+atomic_or_32_nv(volatile uint32_t *target, uint32_t bits)
+{
+	uint32_t old = *target;
+
+	while (!atomic_fcmpset_32(target, &old, old | bits))
+		continue;
+	return (old | bits);
+}
+
+static __inline uint32_t
+atomic_and_32_nv(volatile uint32_t *target, uint32_t bits)
+{
+	uint32_t old = *target;
+
+	while (!atomic_fcmpset_32(target, &old, old & bits))
+		continue;
+	return (old & bits);
+}
+
 #ifndef __sparc64__
 static inline uint32_t
 atomic_cas_32(volatile uint32_t *target, uint32_t cmp, uint32_t newval)
@@ -131,6 +167,38 @@ atomic_add_64_nv(volatile uint64_t *target, int64_t delta)
 	return (atomic_fetchadd_64(target, delta) + delta);
 }
 
+static __inline void
+atomic_or_64(volatile uint64_t *target, uint64_t bits)
+{
+	atomic_set_64(target, bits);
+}
+
+static __inline void
+atomic_and_64(volatile uint64_t *target, uint64_t bits)
+{
+	atomic_clear_64(target, ~bits);
+}
+
+static __inline uint64_t
+atomic_or_64_nv(volatile uint64_t *target, uint64_t bits)
+{
+	uint64_t old = *target;
+
+	while (!atomic_fcmpset_64(target, &old, old | bits))
+		continue;
+	return (old | bits);
+}
+
+static __inline uint64_t
+atomic_and_64_nv(volatile uint64_t *target, uint64_t bits)
+{
+	uint64_t old = *target;
+
+	while (!atomic_fcmpset_64(target, &old, old & bits))
+		continue;
+	return (old & bits);
+}
+
 #ifndef __sparc64__
 static inline uint64_t
 atomic_cas_64(volatile uint64_t *target, uint64_t cmp, uint64_t newval)

include/os/linux/spl/sys/atomic.h

@@ -47,6 +47,8 @@
 #define	atomic_dec_32_nv(v)	atomic_dec_return((atomic_t *)(v))
 #define	atomic_add_32_nv(v, i)	atomic_add_return((i), (atomic_t *)(v))
 #define	atomic_sub_32_nv(v, i)	atomic_sub_return((i), (atomic_t *)(v))
+#define	atomic_or_32(v, i)	atomic_or((i), (atomic_t *)(v))
+#define	atomic_and_32(v, i)	atomic_and((i), (atomic_t *)(v))
 #define	atomic_cas_32(v, x, y)	atomic_cmpxchg((atomic_t *)(v), x, y)
 #define	atomic_swap_32(v, x)	atomic_xchg((atomic_t *)(v), x)
 #define	atomic_load_32(v)	atomic_read((atomic_t *)(v))
@@ -59,11 +61,37 @@
 #define	atomic_dec_64_nv(v)	atomic64_dec_return((atomic64_t *)(v))
 #define	atomic_add_64_nv(v, i)	atomic64_add_return((i), (atomic64_t *)(v))
 #define	atomic_sub_64_nv(v, i)	atomic64_sub_return((i), (atomic64_t *)(v))
+#define	atomic_or_64(v, i)	atomic64_or((i), (atomic64_t *)(v))
+#define	atomic_and_64(v, i)	atomic64_and((i), (atomic64_t *)(v))
 #define	atomic_cas_64(v, x, y)	atomic64_cmpxchg((atomic64_t *)(v), x, y)
 #define	atomic_swap_64(v, x)	atomic64_xchg((atomic64_t *)(v), x)
 #define	atomic_load_64(v)	atomic64_read((atomic64_t *)(v))
 #define	atomic_store_64(v, x)	atomic64_set((atomic64_t *)(v), x)
 
+static __inline__ uint32_t
+atomic_or_32_nv(volatile uint32_t *target, uint32_t bits)
+{
+	return (atomic_fetch_or(bits, (atomic_t *)target) | bits);
+}
+
+static __inline__ uint32_t
+atomic_and_32_nv(volatile uint32_t *target, uint32_t bits)
+{
+	return (atomic_fetch_and(bits, (atomic_t *)target) & bits);
+}
+
+static __inline__ uint64_t
+atomic_or_64_nv(volatile uint64_t *target, uint64_t bits)
+{
+	return (atomic64_fetch_or(bits, (atomic64_t *)target) | bits);
+}
+
+static __inline__ uint64_t
+atomic_and_64_nv(volatile uint64_t *target, uint64_t bits)
+{
+	return (atomic64_fetch_and(bits, (atomic64_t *)target) & bits);
+}
+
 #ifdef _LP64
 static __inline__ void *
 atomic_cas_ptr(volatile void *target,  void *cmp, void *newval)

include/sys/spa_impl.h

@@ -142,11 +142,15 @@ struct spa_aux_vdev {
 	uint_t		sav_npending;		/* # pending devices */
 };
 
+/*
+ * scl_count holds the reference count plus SCL_COUNT_WRITER, so that a reader
+ * can take a reference and check for a writer in one atomic operation.
+ */
 typedef struct spa_config_lock {
 	kmutex_t	scl_lock;
 	kthread_t	*scl_writer;
 	int		scl_write_wanted;
-	int		scl_count;
+	uint32_t	scl_count;
 	kcondvar_t	scl_cv;
 } ____cacheline_aligned spa_config_lock_t;
 

module/os/freebsd/spl/spl_atomic.c

@@ -91,6 +91,46 @@ atomic_add_64_nv(volatile uint64_t *target, int64_t delta)
 	return (newval);
 }
 
+void
+atomic_or_64(volatile uint64_t *target, uint64_t bits)
+{
+
+	mtx_lock(&atomic_mtx);
+	*target |= bits;
+	mtx_unlock(&atomic_mtx);
+}
+
+void
+atomic_and_64(volatile uint64_t *target, uint64_t bits)
+{
+
+	mtx_lock(&atomic_mtx);
+	*target &= bits;
+	mtx_unlock(&atomic_mtx);
+}
+
+uint64_t
+atomic_or_64_nv(volatile uint64_t *target, uint64_t bits)
+{
+	uint64_t newval;
+
+	mtx_lock(&atomic_mtx);
+	newval = (*target |= bits);
+	mtx_unlock(&atomic_mtx);
+	return (newval);
+}
+
+uint64_t
+atomic_and_64_nv(volatile uint64_t *target, uint64_t bits)
+{
+	uint64_t newval;
+
+	mtx_lock(&atomic_mtx);
+	newval = (*target &= bits);
+	mtx_unlock(&atomic_mtx);
+	return (newval);
+}
+
 uint64_t
 atomic_cas_64(volatile uint64_t *target, uint64_t cmp, uint64_t newval)
 {

module/zfs/spa_misc.c

@@ -468,6 +468,29 @@ spa_config_lock_destroy(spa_t *spa)
 	}
 }
 
+/* A writer wants or holds the lock.  Mutated only under scl_lock. */
+#define	SCL_COUNT_WRITER	0x80000000U
+#define	SCL_COUNT_REFS(c)	((c) & ~SCL_COUNT_WRITER)
+
+/*
+ * Drop one reader reference.  The result tells us both that it was the last
+ * one and that a writer is waiting for it.
+ */
+static void
+spa_config_exit_read(spa_config_lock_t *scl)
+{
+	uint32_t count;
+
+	ASSERT3U(SCL_COUNT_REFS(scl->scl_count), >, 0);
+
+	count = atomic_dec_32_nv(&scl->scl_count);
+	if (count != SCL_COUNT_WRITER)
+		return;
+	mutex_enter(&scl->scl_lock);
+	cv_broadcast(&scl->scl_cv);
+	mutex_exit(&scl->scl_lock);
+}
+
 int
 spa_config_tryenter(spa_t *spa, int locks, const void *tag, krw_t rw)
 {
@@ -475,27 +498,31 @@ spa_config_tryenter(spa_t *spa, int locks, const void *tag, krw_t rw)
 		spa_config_lock_t *scl = &spa->spa_config_lock[i];
 		if (!(locks & (1 << i)))
 			continue;
-		mutex_enter(&scl->scl_lock);
 		if (rw == RW_READER) {
-			if (scl->scl_writer || scl->scl_write_wanted) {
-				mutex_exit(&scl->scl_lock);
+			if (atomic_inc_32_nv(&scl->scl_count) &
+			    SCL_COUNT_WRITER) {
+				spa_config_exit_read(scl);
 				spa_config_exit(spa, locks & ((1 << i) - 1),
 				    tag);
 				return (0);
 			}
 		} else {
+			mutex_enter(&scl->scl_lock);
 			ASSERT(scl->scl_writer != curthread);
-			if (scl->scl_count != 0) {
+			/* Take the reference together with the bit or fail. */
+			if (atomic_cas_32(&scl->scl_count, 0,
+			    SCL_COUNT_WRITER | 1) != 0) {
 				mutex_exit(&scl->scl_lock);
 				spa_config_exit(spa, locks & ((1 << i) - 1),
 				    tag);
 				return (0);
 			}
+			scl->scl_write_wanted++;
 			scl->scl_writer = curthread;
+			mutex_exit(&scl->scl_lock);
 		}
-		scl->scl_count++;
-		mutex_exit(&scl->scl_lock);
 	}
+	membar_consumer();
 	return (1);
 }
 
@@ -514,6 +541,19 @@ spa_config_enter_impl(spa_t *spa, int locks, const void *tag, krw_t rw,
 			wlocks_held |= (1 << i);
 		if (!(locks & (1 << i)))
 			continue;
+
+		/*
+		 * Priority readers have to tell a waiting writer from a
+		 * holding one, which one word can not express, so they take
+		 * the lock.  They are rare.
+		 */
+		if (rw == RW_READER && !priority_flag) {
+			if ((atomic_inc_32_nv(&scl->scl_count) &
+			    SCL_COUNT_WRITER) == 0)
+				continue;
+			spa_config_exit_read(scl);
+		}
+
 		mutex_enter(&scl->scl_lock);
 		if (rw == RW_READER) {
 			while (scl->scl_writer ||
@@ -522,16 +562,18 @@ spa_config_enter_impl(spa_t *spa, int locks, const void *tag, krw_t rw,
 			}
 		} else {
 			ASSERT(scl->scl_writer != curthread);
-			while (scl->scl_count != 0) {
-				scl->scl_write_wanted++;
+			scl->scl_write_wanted++;
+			atomic_or_32(&scl->scl_count, SCL_COUNT_WRITER);
+			while (SCL_COUNT_REFS(scl->scl_count) != 0)
 				cv_wait(&scl->scl_cv, &scl->scl_lock);
-				scl->scl_write_wanted--;
-			}
 			scl->scl_writer = curthread;
 		}
-		scl->scl_count++;
+		atomic_inc_32(&scl->scl_count);
 		mutex_exit(&scl->scl_lock);
 	}
+
+	/* Pair with the membar_producer() in spa_config_exit(). */
+	membar_consumer();
 	ASSERT3U(wlocks_held, <=, locks);
 }
 
@@ -564,14 +606,25 @@ spa_config_exit(spa_t *spa, int locks, const void *tag)
 		spa_config_lock_t *scl = &spa->spa_config_lock[i];
 		if (!(locks & (1 << i)))
 			continue;
-		mutex_enter(&scl->scl_lock);
-		ASSERT(scl->scl_count > 0);
-		if (--scl->scl_count == 0) {
-			ASSERT(scl->scl_writer == NULL ||
-			    scl->scl_writer == curthread);
-			scl->scl_writer = NULL;	/* OK in either case */
-			cv_broadcast(&scl->scl_cv);
+		if (scl->scl_writer != curthread) {
+			spa_config_exit_read(scl);
+			continue;
 		}
+
+		/*
+		 * A reader mid-backoff may hold a reference, so the count is
+		 * not ours alone.  Clearing SCL_COUNT_WRITER opens the reader
+		 * fast path, so it goes last and after a barrier.
+		 */
+		mutex_enter(&scl->scl_lock);
+		ASSERT3U(SCL_COUNT_REFS(scl->scl_count), >, 0);
+		scl->scl_writer = NULL;
+		atomic_dec_32(&scl->scl_count);
+		if (--scl->scl_write_wanted == 0) {
+			membar_producer();
+			atomic_and_32(&scl->scl_count, ~SCL_COUNT_WRITER);
+		}
+		cv_broadcast(&scl->scl_cv);
 		mutex_exit(&scl->scl_lock);
 	}
 }
@@ -585,7 +638,7 @@ spa_config_held(spa_t *spa, int locks, krw_t rw)
 		spa_config_lock_t *scl = &spa->spa_config_lock[i];
 		if (!(locks & (1 << i)))
 			continue;
-		if ((rw == RW_READER && scl->scl_count != 0) ||
+		if ((rw == RW_READER && SCL_COUNT_REFS(scl->scl_count) != 0) ||
 		    (rw == RW_WRITER && scl->scl_writer == curthread))
 			locks_held |= 1 << i;
 	}