Remove even more contention between threads

Remove unneeded qp*_ref/unref and qpcache_ref/unref calls by skipping
the extra ref/unref on the database in qp*node_release().  This makes
the extra ref/unref in detachnode() unnecessary as the reference on the
database will always only go down when releasing the database nodes.

lib/dns/qpcache.c

@@ -309,6 +309,8 @@ struct qpcache {
 
 	isc_mem_t *hmctx; /* Memory context for the heaps */
 
+	struct rcu_head rcu_head;
+
 	size_t buckets_count;
 	qpcache_bucket_t buckets[]; /* attribute((counted_by(buckets_count))) */
 };
@@ -686,9 +688,8 @@ delete_node(qpcache_t *qpdb, qpcnode_t *node) {
  * and external references are acquired at the same time, and from
  * qpcnode_release() when we only need to increase the internal references.
  */
-static void
-qpcnode_erefs_increment(qpcache_t *qpdb, qpcnode_t *node,
-			isc_rwlocktype_t nlocktype,
+static bool
+qpcnode_erefs_increment(qpcnode_t *node, isc_rwlocktype_t nlocktype,
 			isc_rwlocktype_t tlocktype DNS__DB_FLARG) {
 	uint_fast32_t refs = isc_refcount_increment0(&node->erefs);
 
@@ -698,7 +699,7 @@ qpcnode_erefs_increment(qpcache_t *qpdb, qpcnode_t *node,
 #endif
 
 	if (refs > 0) {
-		return;
+		return false;
 	}
 
 	/*
@@ -712,15 +713,18 @@ qpcnode_erefs_increment(qpcache_t *qpdb, qpcnode_t *node,
 	INSIST(nlocktype != isc_rwlocktype_none ||
 	       tlocktype != isc_rwlocktype_none);
 
-	qpcache_ref(qpdb);
+	return true;
 }
 
 static void
 qpcnode_acquire(qpcache_t *qpdb, qpcnode_t *node, isc_rwlocktype_t nlocktype,
 		isc_rwlocktype_t tlocktype DNS__DB_FLARG) {
 	qpcnode_ref(node);
-	qpcnode_erefs_increment(qpdb, node, nlocktype,
-				tlocktype DNS__DB_FLARG_PASS);
+	if (qpcnode_erefs_increment(node, nlocktype,
+				    tlocktype DNS__DB_FLARG_PASS))
+	{
+		qpcache_ref(qpdb);
+	}
 }
 
 static void
@@ -732,7 +736,7 @@ cleanup_deadnodes(void *arg);
  * as well, and return true. Otherwise return false.
  */
 static bool
-qpcnode_erefs_decrement(qpcache_t *qpdb, qpcnode_t *node DNS__DB_FLARG) {
+qpcnode_erefs_decrement(qpcnode_t *node DNS__DB_FLARG) {
 	uint_fast32_t refs = isc_refcount_decrement(&node->erefs);
 
 #if DNS_DB_NODETRACE
@@ -743,7 +747,6 @@ qpcnode_erefs_decrement(qpcache_t *qpdb, qpcnode_t *node DNS__DB_FLARG) {
 		return false;
 	}
 
-	qpcache_unref(qpdb);
 	return true;
 }
 
@@ -769,13 +772,13 @@ qpcnode_release(qpcache_t *qpdb, qpcnode_t *node, isc_rwlocktype_t *nlocktypep,
 	bool locked = *tlocktypep != isc_rwlocktype_none;
 	bool write_locked = false;
 
-	if (!qpcnode_erefs_decrement(qpdb, node DNS__DB_FLARG_PASS)) {
-		goto unref;
+	if (!qpcnode_erefs_decrement(node DNS__DB_FLARG_PASS)) {
+		goto unref_node;
 	}
 
 	/* Handle easy and typical case first. */
 	if (!node->dirty && node->data != NULL) {
-		goto unref;
+		goto unref_cache;
 	}
 
 	if (*nlocktypep == isc_rwlocktype_read) {
@@ -789,11 +792,11 @@ qpcnode_release(qpcache_t *qpdb, qpcnode_t *node, isc_rwlocktype_t *nlocktypep,
 		 * decrement erefs again, and see if it's still zero.
 		 */
 		isc_rwlock_t *nlock = &qpdb->buckets[node->locknum].lock;
-		qpcnode_erefs_increment(qpdb, node, *nlocktypep,
-					*tlocktypep DNS__DB_FLARG_PASS);
+		(void)qpcnode_erefs_increment(node, *nlocktypep,
+					      *tlocktypep DNS__DB_FLARG_PASS);
 		NODE_FORCEUPGRADE(nlock, nlocktypep);
-		if (!qpcnode_erefs_decrement(qpdb, node DNS__DB_FLARG_PASS)) {
-			goto unref;
+		if (!qpcnode_erefs_decrement(node DNS__DB_FLARG_PASS)) {
+			goto unref_node;
 		}
 	}
 
@@ -801,6 +804,10 @@ qpcnode_release(qpcache_t *qpdb, qpcnode_t *node, isc_rwlocktype_t *nlocktypep,
 		clean_cache_node(qpdb, node);
 	}
 
+	if (node->data != NULL) {
+		goto unref_cache;
+	}
+
 	/*
 	 * Attempt to switch to a write lock on the tree.  If this fails,
 	 * we will add this node to a linked list of nodes in this locking
@@ -834,10 +841,6 @@ qpcnode_release(qpcache_t *qpdb, qpcnode_t *node, isc_rwlocktype_t *nlocktypep,
 		write_locked = true;
 	}
 
-	if (node->data != NULL) {
-		goto restore_locks;
-	}
-
 	if (write_locked) {
 		/*
 		 * We can now delete the node.
@@ -860,7 +863,6 @@ qpcnode_release(qpcache_t *qpdb, qpcnode_t *node, isc_rwlocktype_t *nlocktypep,
 		}
 	}
 
-restore_locks:
 	/*
 	 * Unlock the tree lock if it wasn't held previously.
 	 */
@@ -868,7 +870,10 @@ restore_locks:
 		TREE_UNLOCK(&qpdb->tree_lock, tlocktypep);
 	}
 
-unref:
+unref_cache:
+	/* .erefs is 0 */
+	qpcache_unref(qpdb);
+unref_node:
 	qpcnode_unref(node);
 }
 
@@ -2513,39 +2518,13 @@ set_index(void *what, unsigned int idx) {
 }
 
 static void
-qpcache__destroy(qpcache_t *qpdb) {
-	unsigned int i;
-	char buf[DNS_NAME_FORMATSIZE];
-	dns_qp_t **treep = NULL;
-
-	for (;;) {
-		/*
-		 * pick the next tree to (start to) destroy
-		 */
-		treep = &qpdb->tree;
-		if (*treep == NULL) {
-			treep = &qpdb->nsec;
-			if (*treep == NULL) {
-				break;
-			}
-		}
-
-		dns_qp_destroy(treep);
-		INSIST(*treep == NULL);
-	}
-
-	if (dns_name_dynamic(&qpdb->common.origin)) {
-		dns_name_format(&qpdb->common.origin, buf, sizeof(buf));
-	} else {
-		strlcpy(buf, "<UNKNOWN>", sizeof(buf));
-	}
-	isc_log_write(DNS_LOGCATEGORY_DATABASE, DNS_LOGMODULE_CACHE,
-		      ISC_LOG_DEBUG(1), "done %s(%s)", __func__, buf);
+qpcache__destroy_rcu(struct rcu_head *rcu_head) {
+	qpcache_t *qpdb = caa_container_of(rcu_head, qpcache_t, rcu_head);
 
 	if (dns_name_dynamic(&qpdb->common.origin)) {
 		dns_name_free(&qpdb->common.origin, qpdb->common.mctx);
 	}
-	for (i = 0; i < qpdb->buckets_count; i++) {
+	for (size_t i = 0; i < qpdb->buckets_count; i++) {
 		NODE_DESTROYLOCK(&qpdb->buckets[i].lock);
 
 		INSIST(ISC_LIST_EMPTY(qpdb->buckets[i].lru));
@@ -2556,13 +2535,6 @@ qpcache__destroy(qpcache_t *qpdb) {
 		isc_heap_destroy(&qpdb->buckets[i].heap);
 	}
 
-	if (qpdb->rrsetstats != NULL) {
-		dns_stats_detach(&qpdb->rrsetstats);
-	}
-	if (qpdb->cachestats != NULL) {
-		isc_stats_detach(&qpdb->cachestats);
-	}
-
 	TREE_DESTROYLOCK(&qpdb->tree_lock);
 	isc_refcount_destroy(&qpdb->references);
 	isc_refcount_destroy(&qpdb->common.references);
@@ -2577,6 +2549,36 @@ qpcache__destroy(qpcache_t *qpdb) {
 						     sizeof(qpdb->buckets[0]));
 }
 
+static void
+qpcache__destroy(qpcache_t *qpdb) {
+	char buf[DNS_NAME_FORMATSIZE];
+
+	if (qpdb->tree != NULL) {
+		dns_qp_destroy(&qpdb->tree);
+	}
+
+	if (qpdb->nsec != NULL) {
+		dns_qp_destroy(&qpdb->nsec);
+	}
+
+	if (qpdb->rrsetstats != NULL) {
+		dns_stats_detach(&qpdb->rrsetstats);
+	}
+	if (qpdb->cachestats != NULL) {
+		isc_stats_detach(&qpdb->cachestats);
+	}
+
+	if (dns_name_dynamic(&qpdb->common.origin)) {
+		dns_name_format(&qpdb->common.origin, buf, sizeof(buf));
+	} else {
+		strlcpy(buf, "<UNKNOWN>", sizeof(buf));
+	}
+	isc_log_write(DNS_LOGCATEGORY_DATABASE, DNS_LOGMODULE_CACHE,
+		      ISC_LOG_DEBUG(1), "done %s(%s)", __func__, buf);
+
+	call_rcu(&qpdb->rcu_head, qpcache__destroy_rcu);
+}
+
 static void
 qpcache_destroy(dns_db_t *arg) {
 	qpcache_t *qpdb = (qpcache_t *)arg;
@@ -2723,18 +2725,16 @@ qpcache_detachnode(dns_db_t *db, dns_dbnode_t **nodep DNS__DB_FLARG) {
 	nlock = &qpdb->buckets[node->locknum].lock;
 
 	/*
-	 * We can't destroy qpcache while holding a nodelock, so
-	 * we need to reference it before acquiring the lock
-	 * and release it afterward.
+	 * qpcache__destroy() uses call_rcu() API to destroy the node locks,
+	 * so it is safe to call it in the middle of NODE_LOCK.
 	 */
-	qpcache_ref(qpdb);
 
+	rcu_read_lock();
 	NODE_RDLOCK(nlock, &nlocktype);
 	qpcnode_release(qpdb, node, &nlocktype, &tlocktype,
 			true DNS__DB_FLARG_PASS);
 	NODE_UNLOCK(nlock, &nlocktype);
-
-	qpcache_detach(&qpdb);
+	rcu_read_unlock();
 }
 
 static isc_result_t

lib/dns/qpzone.c

@@ -741,8 +741,8 @@ dns__qpzone_create(isc_mem_t *mctx, const dns_name_t *origin, dns_dbtype_t type,
  * and external references are acquired at the same time, and from
  * qpznode_release() when we only need to increase the internal references.
  */
-static void
-qpznode_erefs_increment(qpzonedb_t *qpdb, qpznode_t *node DNS__DB_FLARG) {
+static bool
+qpznode_erefs_increment(qpznode_t *node DNS__DB_FLARG) {
 	uint_fast32_t refs = isc_refcount_increment0(&node->erefs);
 #if DNS_DB_NODETRACE
 	fprintf(stderr, "incr:node:%s:%s:%u:%p->erefs = %" PRIuFAST32 "\n",
@@ -750,16 +750,18 @@ qpznode_erefs_increment(qpzonedb_t *qpdb, qpznode_t *node DNS__DB_FLARG) {
 #endif
 
 	if (refs > 0) {
-		return;
+		return false;
 	}
 
-	qpzonedb_ref(qpdb);
+	return true;
 }
 
 static void
 qpznode_acquire(qpzonedb_t *qpdb, qpznode_t *node DNS__DB_FLARG) {
 	qpznode_ref(node);
-	qpznode_erefs_increment(qpdb, node DNS__DB_FLARG_PASS);
+	if (qpznode_erefs_increment(node DNS__DB_FLARG_PASS)) {
+		qpzonedb_ref(qpdb);
+	}
 }
 
 static void
@@ -885,7 +887,7 @@ clean_zone_node(qpznode_t *node, uint32_t least_serial) {
  * as well, and return true. Otherwise return false.
  */
 static bool
-qpznode_erefs_decrement(qpzonedb_t *qpdb, qpznode_t *node DNS__DB_FLARG) {
+qpznode_erefs_decrement(qpznode_t *node DNS__DB_FLARG) {
 	uint_fast32_t refs = isc_refcount_decrement(&node->erefs);
 
 #if DNS_DB_NODETRACE
@@ -896,8 +898,6 @@ qpznode_erefs_decrement(qpzonedb_t *qpdb, qpznode_t *node DNS__DB_FLARG) {
 		return false;
 	}
 
-	qpzonedb_unref(qpdb);
-
 	return true;
 }
 
@@ -918,15 +918,15 @@ qpznode_release(qpzonedb_t *qpdb, qpznode_t *node, uint32_t least_serial,
 		isc_rwlocktype_t *nlocktypep DNS__DB_FLARG) {
 	REQUIRE(*nlocktypep != isc_rwlocktype_none);
 
-	if (!qpznode_erefs_decrement(qpdb, node DNS__DB_FLARG_PASS)) {
-		goto unref;
+	if (!qpznode_erefs_decrement(node DNS__DB_FLARG_PASS)) {
+		goto unref_node;
 	}
 
 	/* Handle easy and typical case first. */
 	if (!node->dirty && (node->data != NULL || node == qpdb->origin ||
 			     node == qpdb->nsec3_origin))
 	{
-		goto unref;
+		goto unref_cache;
 	}
 
 	if (*nlocktypep == isc_rwlocktype_read) {
@@ -940,10 +940,10 @@ qpznode_release(qpzonedb_t *qpdb, qpznode_t *node, uint32_t least_serial,
 		 * decrement erefs again, and see if it's still zero.
 		 */
 		isc_rwlock_t *nlock = &qpdb->buckets[node->locknum].lock;
-		qpznode_erefs_increment(qpdb, node DNS__DB_FLARG_PASS);
+		(void)qpznode_erefs_increment(node DNS__DB_FLARG_PASS);
 		NODE_FORCEUPGRADE(nlock, nlocktypep);
-		if (!qpznode_erefs_decrement(qpdb, node DNS__DB_FLARG_PASS)) {
-			goto unref;
+		if (!qpznode_erefs_decrement(node DNS__DB_FLARG_PASS)) {
+			goto unref_node;
 		}
 	}
 
@@ -960,7 +960,10 @@ qpznode_release(qpzonedb_t *qpdb, qpznode_t *node, uint32_t least_serial,
 		clean_zone_node(node, least_serial);
 	}
 
-unref:
+unref_cache:
+	/* .erefs is 0 */
+	qpzonedb_unref(qpdb);
+unref_node:
 	qpznode_unref(node);
 }