MFC Alexander Motin's GEOM direct dispatch work:

r256603:
Introduce new function devstat_end_transaction_bio_bt(), adding new argument
to specify present time.  Use this function to move binuptime() out of lock,
substantially reducing lock congestion when slow timecounter is used.

r256606:
Move g_io_deliver() out of the lock, as required for direct dispatch.
Move g_destroy_bio() out too to reduce lock scope even more.

r256607:
Fix passing uninitialized bio_resid argument to g_trace().

r256610:
Add unmapped I/O support to GEOM RAID.

r256830:
Restore BIO_UNMAPPED and BIO_TRANSIENT_MAPPING in biodonne() when unmapping
temporary mapped buffer.  That fixes double unmap if biodone() called twice
for the same BIO (but with different done methods).

r256880:
Merge GEOM direct dispatch changes from the projects/camlock branch.

When safety requirements are met, it allows to avoid passing I/O requests
to GEOM g_up/g_down thread, executing them directly in the caller context.
That allows to avoid CPU bottlenecks in g_up/g_down threads, plus avoid
several context switches per I/O.

r259247:
Fix bug introduced at r256607.  We have to recalculate bp_resid here since
sizes of original and completed requests may differ due to end of media.

Testing of the stable/10 merge was done by Netflix, but all of the credit
goes to Alexander and iX Systems.

Submitted by:   mav
Sponsored by:   iX Systems

2 files changed, 62 insertions, 25 deletions

diff --git a/sys/geom/concat/g_concat.c b/sys/geom/concat/g_concat.c
index 9143166..2efc1b5 100644
--- a/sys/geom/concat/g_concat.c
+++ b/sys/geom/concat/g_concat.c
@@ -239,6 +239,27 @@ g_concat_kernel_dump(struct bio *bp)
 }
 
 static void
+g_concat_done(struct bio *bp)
+{
+	struct g_concat_softc *sc;
+	struct bio *pbp;
+
+	pbp = bp->bio_parent;
+	sc = pbp->bio_to->geom->softc;
+	mtx_lock(&sc->sc_lock);
+	if (pbp->bio_error == 0)
+		pbp->bio_error = bp->bio_error;
+	pbp->bio_completed += bp->bio_completed;
+	pbp->bio_inbed++;
+	if (pbp->bio_children == pbp->bio_inbed) {
+		mtx_unlock(&sc->sc_lock);
+		g_io_deliver(pbp, pbp->bio_error);
+	} else
+		mtx_unlock(&sc->sc_lock);
+	g_destroy_bio(bp);
+}
+
+static void
 g_concat_flush(struct g_concat_softc *sc, struct bio *bp)
 {
 	struct bio_queue_head queue;
@@ -250,23 +271,19 @@ g_concat_flush(struct g_concat_softc *sc, struct bio *bp)
 	for (no = 0; no < sc->sc_ndisks; no++) {
 		cbp = g_clone_bio(bp);
 		if (cbp == NULL) {
-			for (cbp = bioq_first(&queue); cbp != NULL;
-			    cbp = bioq_first(&queue)) {
-				bioq_remove(&queue, cbp);
+			while ((cbp = bioq_takefirst(&queue)) != NULL)
 				g_destroy_bio(cbp);
-			}
 			if (bp->bio_error == 0)
 				bp->bio_error = ENOMEM;
 			g_io_deliver(bp, bp->bio_error);
 			return;
 		}
 		bioq_insert_tail(&queue, cbp);
-		cbp->bio_done = g_std_done;
+		cbp->bio_done = g_concat_done;
 		cbp->bio_caller1 = sc->sc_disks[no].d_consumer;
 		cbp->bio_to = sc->sc_disks[no].d_consumer->provider;
 	}
-	for (cbp = bioq_first(&queue); cbp != NULL; cbp = bioq_first(&queue)) {
-		bioq_remove(&queue, cbp);
+	while ((cbp = bioq_takefirst(&queue)) != NULL) {
 		G_CONCAT_LOGREQ(cbp, "Sending request.");
 		cp = cbp->bio_caller1;
 		cbp->bio_caller1 = NULL;
@@ -320,7 +337,10 @@ g_concat_start(struct bio *bp)
 
 	offset = bp->bio_offset;
 	length = bp->bio_length;
-	addr = bp->bio_data;
+	if ((bp->bio_flags & BIO_UNMAPPED) != 0)
+		addr = NULL;
+	else
+		addr = bp->bio_data;
 	end = offset + length;
 
 	bioq_init(&queue);
@@ -338,11 +358,8 @@ g_concat_start(struct bio *bp)
 
 		cbp = g_clone_bio(bp);
 		if (cbp == NULL) {
-			for (cbp = bioq_first(&queue); cbp != NULL;
-			    cbp = bioq_first(&queue)) {
-				bioq_remove(&queue, cbp);
+			while ((cbp = bioq_takefirst(&queue)) != NULL)
 				g_destroy_bio(cbp);
-			}
 			if (bp->bio_error == 0)
 				bp->bio_error = ENOMEM;
 			g_io_deliver(bp, bp->bio_error);
@@ -352,11 +369,21 @@ g_concat_start(struct bio *bp)
 		/*
 		 * Fill in the component buf structure.
 		 */
-		cbp->bio_done = g_std_done;
+		if (len == bp->bio_length)
+			cbp->bio_done = g_std_done;
+		else
+			cbp->bio_done = g_concat_done;
 		cbp->bio_offset = off;
-		cbp->bio_data = addr;
-		addr += len;
 		cbp->bio_length = len;
+		if ((bp->bio_flags & BIO_UNMAPPED) != 0) {
+			cbp->bio_ma_offset += (uintptr_t)addr;
+			cbp->bio_ma += cbp->bio_ma_offset / PAGE_SIZE;
+			cbp->bio_ma_offset %= PAGE_SIZE;
+			cbp->bio_ma_n = round_page(cbp->bio_ma_offset +
+			    cbp->bio_length) / PAGE_SIZE;
+		} else
+			cbp->bio_data = addr;
+		addr += len;
 		cbp->bio_to = disk->d_consumer->provider;
 		cbp->bio_caller1 = disk;
 
@@ -366,8 +393,7 @@ g_concat_start(struct bio *bp)
 	KASSERT(length == 0,
 	    ("Length is still greater than 0 (class=%s, name=%s).",
 	    bp->bio_to->geom->class->name, bp->bio_to->geom->name));
-	for (cbp = bioq_first(&queue); cbp != NULL; cbp = bioq_first(&queue)) {
-		bioq_remove(&queue, cbp);
+	while ((cbp = bioq_takefirst(&queue)) != NULL) {
 		G_CONCAT_LOGREQ(cbp, "Sending request.");
 		disk = cbp->bio_caller1;
 		cbp->bio_caller1 = NULL;
@@ -379,7 +405,7 @@ static void
 g_concat_check_and_run(struct g_concat_softc *sc)
 {
 	struct g_concat_disk *disk;
-	struct g_provider *pp;
+	struct g_provider *dp, *pp;
 	u_int no, sectorsize = 0;
 	off_t start;
 
@@ -388,20 +414,27 @@ g_concat_check_and_run(struct g_concat_softc *sc)
 		return;
 
 	pp = g_new_providerf(sc->sc_geom, "concat/%s", sc->sc_name);
+	pp->flags |= G_PF_DIRECT_SEND | G_PF_DIRECT_RECEIVE |
+	    G_PF_ACCEPT_UNMAPPED;
 	start = 0;
 	for (no = 0; no < sc->sc_ndisks; no++) {
 		disk = &sc->sc_disks[no];
+		dp = disk->d_consumer->provider;
 		disk->d_start = start;
-		disk->d_end = disk->d_start +
-		    disk->d_consumer->provider->mediasize;
+		disk->d_end = disk->d_start + dp->mediasize;
 		if (sc->sc_type == G_CONCAT_TYPE_AUTOMATIC)
-			disk->d_end -= disk->d_consumer->provider->sectorsize;
+			disk->d_end -= dp->sectorsize;
 		start = disk->d_end;
 		if (no == 0)
-			sectorsize = disk->d_consumer->provider->sectorsize;
-		else {
-			sectorsize = lcm(sectorsize,
-			    disk->d_consumer->provider->sectorsize);
+			sectorsize = dp->sectorsize;
+		else
+			sectorsize = lcm(sectorsize, dp->sectorsize);
+
+		/* A provider underneath us doesn't support unmapped */
+		if ((dp->flags & G_PF_ACCEPT_UNMAPPED) == 0) {
+			G_CONCAT_DEBUG(1, "Cancelling unmapped "
+			    "because of %s.", dp->name);
+			pp->flags &= ~G_PF_ACCEPT_UNMAPPED;
 		}
 	}
 	pp->sectorsize = sectorsize;
@@ -468,6 +501,7 @@ g_concat_add_disk(struct g_concat_softc *sc, struct g_provider *pp, u_int no)
 	fcp = LIST_FIRST(&gp->consumer);
 
 	cp = g_new_consumer(gp);
+	cp->flags |= G_CF_DIRECT_SEND | G_CF_DIRECT_RECEIVE;
 	error = g_attach(cp, pp);
 	if (error != 0) {
 		g_destroy_consumer(cp);
@@ -557,6 +591,7 @@ g_concat_create(struct g_class *mp, const struct g_concat_metadata *md,
 	for (no = 0; no < sc->sc_ndisks; no++)
 		sc->sc_disks[no].d_consumer = NULL;
 	sc->sc_type = type;
+	mtx_init(&sc->sc_lock, "gconcat lock", NULL, MTX_DEF);
 
 	gp->softc = sc;
 	sc->sc_geom = gp;
@@ -605,6 +640,7 @@ g_concat_destroy(struct g_concat_softc *sc, boolean_t force)
 	KASSERT(sc->sc_provider == NULL, ("Provider still exists? (device=%s)",
 	    gp->name));
 	free(sc->sc_disks, M_CONCAT);
+	mtx_destroy(&sc->sc_lock);
 	free(sc, M_CONCAT);
 
 	G_CONCAT_DEBUG(0, "Device %s destroyed.", gp->name);
diff --git a/sys/geom/concat/g_concat.h b/sys/geom/concat/g_concat.h
index 1c1e6f5..c2ea366 100644
--- a/sys/geom/concat/g_concat.h
+++ b/sys/geom/concat/g_concat.h
@@ -83,6 +83,7 @@ struct g_concat_softc {
 
 	struct g_concat_disk *sc_disks;
 	uint16_t	 sc_ndisks;
+	struct mtx	 sc_lock;
 };
 #define	sc_name	sc_geom->name
 #endif	/* _KERNEL */