--- sys/geom/geom.h	24 Mar 2011 19:23:42 -0000	1.109
+++ sys/geom/geom.h	12 May 2011 19:46:52 -0000
@@ -95,6 +95,9 @@ struct g_class {
 	g_init_t		*init;
 	g_fini_t		*fini;
 	g_ctl_destroy_geom_t	*destroy_geom;
+	u_int			flags;
+#define	G_DIRECT_UP	0x01
+#define	G_DIRECT_DOWN	0x02
 	/*
 	 * Default values for geom methods
 	 */
--- sys/geom/geom_dev.c	24 Mar 2011 08:37:48 -0000	1.108
+++ sys/geom/geom_dev.c	13 May 2011 19:24:15 -0000
@@ -78,6 +78,7 @@ static struct g_class g_dev_class	= {
 	.version = G_VERSION,
 	.taste = g_dev_taste,
 	.orphan = g_dev_orphan,
+	.flags = G_DIRECT_UP | G_DIRECT_DOWN
 };
 
 void
--- sys/geom/nop/g_nop.c	31 Mar 2011 06:30:59 -0000	1.23
+++ sys/geom/nop/g_nop.c	13 May 2011 14:19:37 -0000
@@ -58,7 +58,8 @@ struct g_class g_nop_class = {
 	.name = G_NOP_CLASS_NAME,
 	.version = G_VERSION,
 	.ctlreq = g_nop_config,
-	.destroy_geom = g_nop_destroy_geom
+	.destroy_geom = g_nop_destroy_geom,
+	.flags = G_DIRECT_UP | G_DIRECT_DOWN
 };
 
 
--- sys/geom/zero/g_zero.c	10 May 2010 19:08:53 -0000	1.5
+++ sys/geom/zero/g_zero.c	12 May 2011 19:42:24 -0000
@@ -111,7 +111,8 @@ static struct g_class g_zero_class = {
 	.name = G_ZERO_CLASS_NAME,
 	.version = G_VERSION,
 	.init = g_zero_init,
-	.destroy_geom = g_zero_destroy_geom
+	.destroy_geom = g_zero_destroy_geom,
+	.flags = G_DIRECT_UP | G_DIRECT_DOWN
 };
 
 DECLARE_GEOM_CLASS(g_zero_class, g_zero);
--- sys/geom/geom_io.c	2 Sep 2010 19:40:28 -0000	1.86
+++ sys/geom/geom_io.c	13 May 2011 19:23:25 -0000
@@ -332,6 +332,59 @@ g_io_check(struct bio *bp)
 }
 
 /*
+ * Return value != 0 if the request was delivered internally.
+ */
+static int
+g_io_sanitize(struct bio *bp)
+{
+	off_t excess;
+	int error;
+
+	error = g_io_check(bp);
+	if (error != 0) {
+		CTR3(KTR_GEOM, "g_down g_io_check on bp %p provider "
+		    "%s returned %d", bp, bp->bio_to->name, error);
+		g_io_deliver(bp, error);
+		return (1);
+	}
+	CTR2(KTR_GEOM, "g_down processing bp %p provider %s", bp,
+	    bp->bio_to->name);
+
+	switch (bp->bio_cmd) {
+	case BIO_READ:
+	case BIO_WRITE:
+	case BIO_DELETE:
+		/* Truncate requests to the end of providers media. */
+		/*
+		 * XXX: What if we truncate because of offset being
+		 * bad, not length?
+		 */
+		excess = bp->bio_offset + bp->bio_length;
+		if (excess > bp->bio_to->mediasize) {
+			excess -= bp->bio_to->mediasize;
+			bp->bio_length -= excess;
+			if (excess > 0) {
+				CTR3(KTR_GEOM, "g_down truncated bio "
+				    "%p provider %s by %d", bp,
+				    bp->bio_to->name, excess);
+			}
+		}
+		/* Deliver zero length transfers right here. */
+		if (bp->bio_length == 0) {
+			g_io_deliver(bp, 0);
+			CTR2(KTR_GEOM, "g_down terminated 0-length "
+			    "bp %p provider %s", bp, bp->bio_to->name);
+			return (1);
+		}
+		break;
+	default:
+		break;
+	}
+
+	return (0);
+}
+
+/*
  * bio classification support.
  *
  * g_register_classifier() and g_unregister_classifier()
@@ -392,7 +445,7 @@ void
 g_io_request(struct bio *bp, struct g_consumer *cp)
 {
 	struct g_provider *pp;
-	int first;
+	int first, dostats;
 
 	KASSERT(cp != NULL, ("NULL cp in g_io_request"));
 	KASSERT(bp != NULL, ("NULL bp in g_io_request"));
@@ -440,15 +493,14 @@ g_io_request(struct bio *bp, struct g_co
 	bp->bio_error = 0;
 	bp->bio_completed = 0;
 
-	KASSERT(!(bp->bio_flags & BIO_ONQUEUE),
-	    ("Bio already on queue bp=%p", bp));
-	bp->bio_flags |= BIO_ONQUEUE;
-
 	if (g_collectstats)
 		binuptime(&bp->bio_t0);
 	else
 		getbinuptime(&bp->bio_t0);
 
+	dostats = (g_collectstats != 0 ||
+	    (!TAILQ_EMPTY(&g_classifier_tailq) && !bp->bio_classifier1));
+
 	/*
 	 * The statistics collection is lockless, as such, but we
 	 * can not update one instance of the statistics from more
@@ -456,26 +508,40 @@ g_io_request(struct bio *bp, struct g_co
 	 *
 	 * We also use the lock to protect the list of classifiers.
 	 */
-	g_bioq_lock(&g_bio_run_down);
-
-	if (!TAILQ_EMPTY(&g_classifier_tailq) && !bp->bio_classifier1)
-		g_run_classifiers(bp);
+	if (dostats) {
+		g_bioq_lock(&g_bio_run_down);
 
-	if (g_collectstats & 1)
-		devstat_start_transaction(pp->stat, &bp->bio_t0);
-	if (g_collectstats & 2)
-		devstat_start_transaction(cp->stat, &bp->bio_t0);
-
-	pp->nstart++;
-	cp->nstart++;
-	first = TAILQ_EMPTY(&g_bio_run_down.bio_queue);
-	TAILQ_INSERT_TAIL(&g_bio_run_down.bio_queue, bp, bio_queue);
-	g_bio_run_down.bio_queue_length++;
-	g_bioq_unlock(&g_bio_run_down);
+		if (!TAILQ_EMPTY(&g_classifier_tailq) && !bp->bio_classifier1)
+			g_run_classifiers(bp);
 
-	/* Pass it on down. */
-	if (first)
-		wakeup(&g_wait_down);
+		if (g_collectstats & 1)
+			devstat_start_transaction(pp->stat, &bp->bio_t0);
+		if (g_collectstats & 2)
+			devstat_start_transaction(cp->stat, &bp->bio_t0);
+	}
+
+	atomic_add_int(&pp->nstart, 1);
+	atomic_add_int(&cp->nstart, 1);
+
+	if (pace == 0 && (pp->geom->class->flags & G_DIRECT_DOWN) != 0) {
+		if (dostats)
+			g_bioq_unlock(&g_bio_run_down);
+		if (!g_io_sanitize(bp))
+			bp->bio_to->geom->start(bp);
+	} else {
+		KASSERT(!(bp->bio_flags & BIO_ONQUEUE),
+		    ("Bio already on queue bp=%p", bp));
+		bp->bio_flags |= BIO_ONQUEUE;
+		if (!dostats)
+			g_bioq_lock(&g_bio_run_down);
+		first = TAILQ_EMPTY(&g_bio_run_down.bio_queue);
+		TAILQ_INSERT_TAIL(&g_bio_run_down.bio_queue, bp, bio_queue);
+		g_bio_run_down.bio_queue_length++;
+		g_bioq_unlock(&g_bio_run_down);
+		/* Pass it on down. */
+		if (first)
+			wakeup(&g_wait_down);
+	}
 }
 
 void
@@ -483,7 +549,7 @@ g_io_deliver(struct bio *bp, int error)
 {
 	struct g_consumer *cp;
 	struct g_provider *pp;
-	int first;
+	int first, dostats;
 
 	KASSERT(bp != NULL, ("NULL bp in g_io_deliver"));
 	pp = bp->bio_to;
@@ -534,16 +600,38 @@ g_io_deliver(struct bio *bp, int error)
 	 * can not update one instance of the statistics from more
 	 * than one thread at a time, so grab the lock first.
 	 */
-	g_bioq_lock(&g_bio_run_up);
-	if (g_collectstats & 1)
-		devstat_end_transaction_bio(pp->stat, bp);
-	if (g_collectstats & 2)
-		devstat_end_transaction_bio(cp->stat, bp);
-
-	cp->nend++;
-	pp->nend++;
-	if (error != ENOMEM) {
+	dostats = (g_collectstats != 0);
+	if (dostats) {
+		g_bioq_lock(&g_bio_run_up);
+
+		if (g_collectstats & 1)
+			devstat_end_transaction_bio(pp->stat, bp);
+		if (g_collectstats & 2)
+			devstat_end_transaction_bio(cp->stat, bp);
+	}
+
+	atomic_add_int(&cp->nend, 1);
+	atomic_add_int(&pp->nend, 1);
+
+	if (error == ENOMEM) {
+		if (!dostats)
+			g_bioq_lock(&g_bio_run_up);
+		pace++;
+		g_bioq_unlock(&g_bio_run_up);
+		if (bootverbose)
+			printf("ENOMEM %p on %p(%s)\n", bp, pp, pp->name);
+		bp->bio_children = 0;
+		bp->bio_inbed = 0;
+		g_io_request(bp, cp);
+	} else if ((cp->geom->class->flags & G_DIRECT_UP) != 0) {
+		if (dostats)
+			g_bioq_unlock(&g_bio_run_up);
+		bp->bio_error = error;
+		biodone(bp);
+	} else {
 		bp->bio_error = error;
+		if (!dostats)
+			g_bioq_lock(&g_bio_run_up);
 		first = TAILQ_EMPTY(&g_bio_run_up.bio_queue);
 		TAILQ_INSERT_TAIL(&g_bio_run_up.bio_queue, bp, bio_queue);
 		bp->bio_flags |= BIO_ONQUEUE;
@@ -551,25 +639,13 @@ g_io_deliver(struct bio *bp, int error)
 		g_bioq_unlock(&g_bio_run_up);
 		if (first)
 			wakeup(&g_wait_up);
-		return;
 	}
-	g_bioq_unlock(&g_bio_run_up);
-
-	if (bootverbose)
-		printf("ENOMEM %p on %p(%s)\n", bp, pp, pp->name);
-	bp->bio_children = 0;
-	bp->bio_inbed = 0;
-	g_io_request(bp, cp);
-	pace++;
-	return;
 }
 
 void
 g_io_schedule_down(struct thread *tp __unused)
 {
 	struct bio *bp;
-	off_t excess;
-	int error;
 
 	for(;;) {
 		g_bioq_lock(&g_bio_run_down);
@@ -581,50 +657,16 @@ g_io_schedule_down(struct thread *tp __u
 			continue;
 		}
 		CTR0(KTR_GEOM, "g_down has work to do");
-		g_bioq_unlock(&g_bio_run_down);
 		if (pace > 0) {
 			CTR1(KTR_GEOM, "g_down pacing self (pace %d)", pace);
-			pause("g_down", hz/10);
 			pace--;
+			g_bioq_unlock(&g_bio_run_down);
+			pause("g_down", hz/10);
+		} else {
+			g_bioq_unlock(&g_bio_run_down);
 		}
-		error = g_io_check(bp);
-		if (error) {
-			CTR3(KTR_GEOM, "g_down g_io_check on bp %p provider "
-			    "%s returned %d", bp, bp->bio_to->name, error);
-			g_io_deliver(bp, error);
+		if (g_io_sanitize(bp))
 			continue;
-		}
-		CTR2(KTR_GEOM, "g_down processing bp %p provider %s", bp,
-		    bp->bio_to->name);
-		switch (bp->bio_cmd) {
-		case BIO_READ:
-		case BIO_WRITE:
-		case BIO_DELETE:
-			/* Truncate requests to the end of providers media. */
-			/*
-			 * XXX: What if we truncate because of offset being
-			 * bad, not length?
-			 */
-			excess = bp->bio_offset + bp->bio_length;
-			if (excess > bp->bio_to->mediasize) {
-				excess -= bp->bio_to->mediasize;
-				bp->bio_length -= excess;
-				if (excess > 0)
-					CTR3(KTR_GEOM, "g_down truncated bio "
-					    "%p provider %s by %d", bp,
-					    bp->bio_to->name, excess);
-			}
-			/* Deliver zero length transfers right here. */
-			if (bp->bio_length == 0) {
-				g_io_deliver(bp, 0);
-				CTR2(KTR_GEOM, "g_down terminated 0-length "
-				    "bp %p provider %s", bp, bp->bio_to->name);
-				continue;
-			}
-			break;
-		default:
-			break;
-		}
 		THREAD_NO_SLEEPING();
 		CTR4(KTR_GEOM, "g_down starting bp %p provider %s off %ld "
 		    "len %ld", bp, bp->bio_to->name, bp->bio_offset,