Bridge speculative and prescient prefetchers

There are cases when we need to read large objects sequentially.
Prescient prefetcher may be limited by size, trying to prefetch
it all at once.  Speculative prefetcher can do it nicer for ARC,
but needs a ramp-up period at the beginning.

To cover this gap, introduce dmu_prefetch_stream(), priming the
speculative prefetcher to either go full speed on first demand
request, or even start some prefetch immediately, but continue
following the demand later.

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

include/sys/dmu.h

@@ -963,6 +963,10 @@ void dmu_prefetch_by_dnode(dnode_t *dn, int64_t level, uint64_t offset,
 void dmu_prefetch_dnode(objset_t *os, uint64_t object, enum zio_priority pri);
 int dmu_prefetch_wait(objset_t *os, uint64_t object, uint64_t offset,
     uint64_t size);
+void dmu_prefetch_stream(objset_t *os, uint64_t object, uint64_t offset,
+    uint64_t len, boolean_t start_now);
+void dmu_prefetch_stream_by_dnode(dnode_t *dn, uint64_t offset,
+    uint64_t len, boolean_t start_now);
 
 typedef struct dmu_object_info {
 	/* All sizes are in bytes unless otherwise indicated. */

include/sys/dmu_zfetch.h

@@ -79,6 +79,7 @@ void		zfetch_fini(void);
 
 void		dmu_zfetch_init(zfetch_t *, struct dnode *);
 void		dmu_zfetch_fini(zfetch_t *);
+boolean_t	dmu_zfetch_prime(zfetch_t *, uint64_t, uint64_t);
 zstream_t	*dmu_zfetch_prepare(zfetch_t *, uint64_t, uint64_t, boolean_t,
     boolean_t);
 void		dmu_zfetch_run(zfetch_t *, zstream_t *, boolean_t, boolean_t,

module/zfs/ddt_log.c

@@ -604,8 +604,7 @@ ddt_log_load_one(ddt_t *ddt, uint_t n)
 	}
 
 	if (hdr.dlh_length > 0) {
-		dmu_prefetch_by_dnode(dn, 0, 0, hdr.dlh_length,
-		    ZIO_PRIORITY_SYNC_READ);
+		dmu_prefetch_stream_by_dnode(dn, 0, hdr.dlh_length, B_FALSE);
 
 		for (uint64_t offset = 0; offset < hdr.dlh_length;
 		    offset += dn->dn_datablksz) {

module/zfs/dmu.c

@@ -779,6 +779,54 @@ dmu_prefetch_by_dnode(dnode_t *dn, int64_t level, uint64_t offset,
 	rw_exit(&dn->dn_struct_rwlock);
 }
 
+/*
+ * Prime a prefetch for sequential accesses from offset for at least len bytes.
+ */
+void
+dmu_prefetch_stream(objset_t *os, uint64_t object, uint64_t offset,
+    uint64_t len, boolean_t start_now)
+{
+	dnode_t *dn;
+
+	if (dnode_hold(os, object, FTAG, &dn) != 0)
+		return;
+	dmu_prefetch_stream_by_dnode(dn, offset, len, start_now);
+	dnode_rele(dn, FTAG);
+}
+
+void
+dmu_prefetch_stream_by_dnode(dnode_t *dn, uint64_t offset, uint64_t len,
+    boolean_t start_now)
+{
+	rw_enter(&dn->dn_struct_rwlock, RW_READER);
+	if (dn->dn_datablkshift != 0) {
+		uint64_t start = dbuf_whichblock(dn, 0, offset);
+		if (len == 0) {
+			if (dmu_zfetch_prime(&dn->dn_zfetch, start, start) &&
+			    start_now) {
+				dmu_zfetch(&dn->dn_zfetch, start, 0, B_TRUE,
+				    B_TRUE, B_TRUE, B_FALSE);
+			}
+		} else {
+			uint64_t end = dbuf_whichblock(dn, 0, offset + len - 1);
+			if (start == end) {
+				if (start_now) {
+					dbuf_prefetch(dn, 0, start,
+					    ZIO_PRIORITY_ASYNC_READ, 0);
+				}
+			} else if (
+			    dmu_zfetch_prime(&dn->dn_zfetch, start, end + 1) &&
+			    start_now) {
+				dmu_zfetch(&dn->dn_zfetch, start, 0, B_TRUE,
+				    B_TRUE, B_TRUE, B_FALSE);
+			}
+		}
+	} else if (offset < dn->dn_datablksz && start_now) {
+		dbuf_prefetch(dn, 0, 0, ZIO_PRIORITY_ASYNC_READ, 0);
+	}
+	rw_exit(&dn->dn_struct_rwlock);
+}
+
 typedef struct {
 	kmutex_t	dpa_lock;
 	kcondvar_t	dpa_cv;
@@ -2943,6 +2991,8 @@ EXPORT_SYMBOL(dmu_buf_rele_array);
 EXPORT_SYMBOL(dmu_prefetch);
 EXPORT_SYMBOL(dmu_prefetch_by_dnode);
 EXPORT_SYMBOL(dmu_prefetch_dnode);
+EXPORT_SYMBOL(dmu_prefetch_stream);
+EXPORT_SYMBOL(dmu_prefetch_stream_by_dnode);
 EXPORT_SYMBOL(dmu_free_range);
 EXPORT_SYMBOL(dmu_free_long_range);
 EXPORT_SYMBOL(dmu_free_long_object);

module/zfs/dmu_zfetch.c

@@ -454,6 +454,61 @@ dmu_zfetch_future(zstream_t *zs, uint64_t blkid, uint64_t nblks)
 	return (0);
 }
 
+/*
+ * Prime a zfetch stream at blkid, so that the first demand access triggered
+ * enough prefetch without ramp-up to sequentially read up to end_blkid.
+ */
+boolean_t
+dmu_zfetch_prime(zfetch_t *zf, uint64_t blkid, uint64_t end_blkid)
+{
+	zstream_t *zs;
+	dnode_t *dn = zf->zf_dnode;
+	spa_t *spa = dn->dn_objset->os_spa;
+
+	ASSERT(RW_LOCK_HELD(&dn->dn_struct_rwlock));
+	if (zfs_prefetch_disable ||
+	    dn->dn_objset->os_prefetch == ZFS_PREFETCH_NONE)
+		return (B_FALSE);
+
+	if (!spa_indirect_vdevs_loaded(spa))
+		return (B_FALSE);
+
+	uint64_t maxblkid = dn->dn_maxblkid;
+	unsigned int dbs = dn->dn_datablkshift;
+
+	if (blkid >= maxblkid)
+		return (B_FALSE);
+	if (end_blkid > maxblkid + 1)
+		end_blkid = maxblkid + 1;
+
+	mutex_enter(&zf->zf_lock);
+
+	/* Skip if a nearby stream already covers this range. */
+	uint_t max_near = zfetch_max_reorder >> dbs;
+	for (zs = list_head(&zf->zf_stream); zs != NULL;
+	    zs = list_next(&zf->zf_stream, zs)) {
+		uint64_t diff = (blkid >= zs->zs_blkid) ?
+		    (blkid - zs->zs_blkid) : (zs->zs_blkid - blkid);
+		if (diff <= max_near) {
+			mutex_exit(&zf->zf_lock);
+			return (B_FALSE);
+		}
+	}
+
+	dmu_zfetch_stream_create(zf, blkid);
+	zs = list_head(&zf->zf_stream);
+	ASSERT(zs != NULL);
+	ASSERT3U(zs->zs_blkid, ==, blkid);
+
+	/* dmu_zfetch_prepare() will double the distances, so take a half. */
+	unsigned int nbytes = ((end_blkid - blkid) << dbs) / 2;
+	zs->zs_pf_dist = MIN(nbytes, zfetch_min_distance);
+	zs->zs_ipf_dist = MIN(nbytes, zfetch_max_idistance);
+
+	mutex_exit(&zf->zf_lock);
+	return (B_TRUE);
+}
+
 /*
  * This is the predictive prefetch entry point.  dmu_zfetch_prepare()
  * associates dnode access specified with blkid and nblks arguments with
@@ -493,20 +548,21 @@ dmu_zfetch_prepare(zfetch_t *zf, uint64_t blkid, uint64_t nblks,
 
 	/*
 	 * As a fast path for small (single-block) files, ignore access
-	 * to the first block.
+	 * to the first block, unless some streams exist, since a prime
+	 * may be waiting.
 	 */
-	if (!have_lock && blkid == 0)
+	if (!have_lock && blkid == 0 && zf->zf_numstreams == 0)
 		return (NULL);
 
 	if (!have_lock)
 		rw_enter(&zf->zf_dnode->dn_struct_rwlock, RW_READER);
 
 	/*
-	 * A fast path for small files for which no prefetch will
-	 * happen.
+	 * A fast path for small files for which no prefetch will happen,
+	 * unless streams exist, since a prime may be waiting.
 	 */
 	uint64_t maxblkid = zf->zf_dnode->dn_maxblkid;
-	if (maxblkid < 2) {
+	if (maxblkid < 2 && (maxblkid == 0 || zf->zf_numstreams == 0)) {
 		if (!have_lock)
 			rw_exit(&zf->zf_dnode->dn_struct_rwlock);
 		return (NULL);
@@ -589,7 +645,7 @@ future:
 	zs->zs_atime = gethrestime_sec();
 
 	/* Exit if we already prefetched for this position before. */
-	if (nblks == 0)
+	if (nblks == 0 && zs->zs_ipf_end > end_blkid)
 		goto out;
 
 	/* If the file is ending, remove the stream. */
@@ -643,6 +699,7 @@ out:
 	 * Do the same for indirects, starting where we will stop reading
 	 * data blocks (and the indirects that point to them).
 	 */
+	nbytes = MAX(nbytes, (1 << dbs));
 	if (unlikely(zs->zs_ipf_dist < nbytes))
 		zs->zs_ipf_dist = nbytes;
 	else

module/zfs/spa_log_spacemap.c

@@ -1160,7 +1160,7 @@ spa_ld_log_sm_data(spa_t *spa)
 	while (sls != NULL) {
 		/* Prefetch log spacemaps up to 16 TXGs or MBs ahead. */
 		if (psls != NULL && pn < 16 &&
-		    (pn < 2 || ps < 2 * dmu_prefetch_max)) {
+		    (pn < 2 || ps < dmu_prefetch_max)) {
 			error = space_map_open(&psls->sls_sm,
 			    spa_meta_objset(spa), psls->sls_sm_obj, 0,
 			    UINT64_MAX, SPA_MINBLOCKSHIFT);
@@ -1171,9 +1171,9 @@ spa_ld_log_sm_data(spa_t *spa)
 				    (u_longlong_t)sls->sls_sm_obj, error);
 				goto out;
 			}
-			dmu_prefetch(spa_meta_objset(spa), psls->sls_sm_obj,
-			    0, 0, space_map_length(psls->sls_sm),
-			    ZIO_PRIORITY_ASYNC_READ);
+			dmu_prefetch_stream(spa_meta_objset(spa),
+			    psls->sls_sm_obj, 0,
+			    space_map_length(psls->sls_sm), B_TRUE);
 			pn++;
 			ps += space_map_length(psls->sls_sm);
 			psls = AVL_NEXT(&spa->spa_sm_logs_by_txg, psls);

module/zfs/space_map.c

@@ -92,8 +92,7 @@ space_map_iterate(space_map_t *sm, uint64_t end, sm_cb_t callback, void *arg)
 	ASSERT3U(end, <=, space_map_length(sm));
 	ASSERT0(P2PHASE(end, sizeof (uint64_t)));
 
-	dmu_prefetch(sm->sm_os, space_map_object(sm), 0, 0, end,
-	    ZIO_PRIORITY_SYNC_READ);
+	dmu_prefetch_stream(sm->sm_os, space_map_object(sm), 0, end, B_FALSE);
 
 	int error = 0;
 	uint64_t txg = 0, sync_pass = 0;