added true async in zfs filesystem by adding a thin async zpl proxy

config/kernel-ki_complete.m4

@@ -0,0 +1,26 @@
+dnl # SPDX-License-Identifier: CDDL-1.0
+dnl #
+dnl # 5.19 API change,
+dnl # kiocb->ki_complete() reduced from 3 args to 2:
+dnl #   old: void (*ki_complete)(struct kiocb *, long, long)
+dnl #   new: void (*ki_complete)(struct kiocb *, long)
+dnl #
+AC_DEFUN([ZFS_AC_KERNEL_SRC_KIOCB_KI_COMPLETE], [
+	ZFS_LINUX_TEST_SRC([kiocb_ki_complete_2args], [
+		#include <linux/fs.h>
+	],[
+		struct kiocb *kiocb = NULL;
+		kiocb->ki_complete(kiocb, 0);
+	])
+])
+
+AC_DEFUN([ZFS_AC_KERNEL_KIOCB_KI_COMPLETE], [
+	AC_MSG_CHECKING([whether kiocb->ki_complete() wants 2 args])
+	ZFS_LINUX_TEST_RESULT([kiocb_ki_complete_2args], [
+		AC_MSG_RESULT(yes)
+		AC_DEFINE(HAVE_2ARGS_KI_COMPLETE, 1,
+		    [kiocb->ki_complete() wants 2 args])
+	],[
+		AC_MSG_RESULT(no)
+	])
+])

config/kernel.m4

@@ -91,6 +91,7 @@ AC_DEFUN([ZFS_AC_KERNEL_TEST_SRC], [
 	ZFS_AC_KERNEL_SRC_VFS_REMAP_FILE_RANGE
 	ZFS_AC_KERNEL_SRC_VFS_CLONE_FILE_RANGE
 	ZFS_AC_KERNEL_SRC_VFS_DEDUPE_FILE_RANGE
+	ZFS_AC_KERNEL_SRC_KIOCB_KI_COMPLETE
 	ZFS_AC_KERNEL_SRC_KMAP_ATOMIC_ARGS
 	ZFS_AC_KERNEL_SRC_KMAP_LOCAL_PAGE
 	ZFS_AC_KERNEL_SRC_FOLLOW_DOWN
@@ -207,6 +208,7 @@ AC_DEFUN([ZFS_AC_KERNEL_TEST_RESULT], [
 	ZFS_AC_KERNEL_VFS_REMAP_FILE_RANGE
 	ZFS_AC_KERNEL_VFS_CLONE_FILE_RANGE
 	ZFS_AC_KERNEL_VFS_DEDUPE_FILE_RANGE
+	ZFS_AC_KERNEL_KIOCB_KI_COMPLETE
 	ZFS_AC_KERNEL_KMAP_ATOMIC_ARGS
 	ZFS_AC_KERNEL_KMAP_LOCAL_PAGE
 	ZFS_AC_KERNEL_FOLLOW_DOWN

include/os/linux/zfs/sys/zfs_vfsops_os.h

@@ -103,6 +103,9 @@ struct zfsvfs {
 	boolean_t	z_use_hold;	/* held via dmu_objset_hold */
 	rrmlock_t	z_teardown_lock;
 	krwlock_t	z_teardown_inactive_lock;
+	kmutex_t	z_async_dio_lock; /* protects z_async_dio_inflight */
+	kcondvar_t	z_async_dio_cv;	/* signals drain to teardown */
+	uint64_t	z_async_dio_inflight; /* bytes submitted and not done */
 	list_t		z_all_znodes;	/* all znodes in the fs */
 	unsigned long	z_rollback_time; /* last online rollback time */
 	uint64_t	z_snap_atime;	/* last snapshot access time */

include/os/linux/zfs/sys/zfs_vnops_os.h

@@ -40,6 +40,8 @@
 extern "C" {
 #endif
 
+struct spa;
+
 extern int zfs_open(struct inode *ip, int mode, int flag, cred_t *cr);
 extern int zfs_close(struct inode *ip, int flag, cred_t *cr);
 extern int zfs_write_simple(znode_t *zp, const void *data, size_t len,
@@ -94,6 +96,11 @@ extern int zfs_rename_idmap(znode_t *sdzp, char *snm, znode_t *tdzp,
     zidmap_t *idmap);
 extern int zfs_symlink_idmap(znode_t *dzp, char *name, vattr_t *vap,
     char *link, znode_t **zpp, cred_t *cr, int flags, zidmap_t *idmap);
+/* async Direct I/O reads */
+extern void zpl_async_read_init(void);
+extern void zpl_async_read_fini(void);
+extern void zpl_async_read_pool_destroy(struct spa *);
+extern int zfs_dio_pin_pages(zfs_uio_t *, zfs_uio_rw_t);
 
 #ifdef	__cplusplus
 }

include/sys/spa_impl.h

@@ -505,6 +505,15 @@ struct spa {
 	spa_condense_stat_t	spa_condense_stats[SPA_CONDENSE_TYPES];
 	kmutex_t		spa_condense_stats_lock;
 
+#if defined(__linux__)
+	/*
+	 * Per-pool async Direct I/O read worker pool (zpl_async_read_pool_t *).
+	 * One pool per SPA, shared by all datasets of the pool, created lazily
+	 * on the first async read and destroyed in spa_remove().
+	 */
+	void			*spa_zpl_async_read_pool;
+#endif
+
 #ifdef ZFS_DEBUG
 	/* see spa_condense_debug_task() */
 	taskqid_t	spa_condense_debug_tqid;

include/sys/zfs_vnops.h

@@ -33,6 +33,7 @@ extern int zfs_bclone_enabled;
 
 extern int zfs_fsync(znode_t *, int, cred_t *);
 extern int zfs_read(znode_t *, zfs_uio_t *, int, cred_t *);
+extern int zfs_read_impl(znode_t *, zfs_uio_t *, int, cred_t *);
 extern int zfs_write(znode_t *, zfs_uio_t *, int, cred_t *);
 extern int zfs_holey(znode_t *, ulong_t, loff_t *);
 extern int zfs_access(znode_t *, int, int, cred_t *);

module/os/linux/zfs/zfs_ioctl_os.c

@@ -66,6 +66,7 @@
 #include <sys/zfs_ioctl_impl.h>
 
 #include <sys/zfs_sysfs.h>
+#include <sys/zfs_vnops_os.h>
 #include <linux/miscdevice.h>
 #include <linux/slab.h>
 
@@ -302,6 +303,7 @@ openzfs_init_os(void)
 		return (-error);
 	}
 
+	zpl_async_read_init();
 	zfs_sysfs_init();
 
 	printk(KERN_NOTICE "ZFS: Loaded module v%s-%s%s, "
@@ -327,6 +329,7 @@ static void
 openzfs_fini_os(void)
 {
 	zfs_sysfs_fini();
+	zpl_async_read_fini();
 	zfs_kmod_fini();
 
 	printk(KERN_NOTICE "ZFS: Unloaded module v%s-%s%s\n",

module/os/linux/zfs/zfs_uio.c

@@ -56,6 +56,20 @@
 #include <linux/pagemap.h>
 #include <linux/mman.h>
 
+#ifdef ZFS_DEBUG
+/*
+ * Count of read copies served through the pinned-page branch of
+ * zfs_uiomove_iter() (pages pinned by the async Direct I/O submission but
+ * the data copied through the ARC).  Debug builds only; writable so tests
+ * can reset it to verify the branch is exercised.
+ */
+static unsigned long zfs_async_read_pinned_copies = 0;
+module_param(zfs_async_read_pinned_copies, ulong, 0644);
+MODULE_PARM_DESC(zfs_async_read_pinned_copies,
+	"Number of read copies into pre-pinned pages (async DIO ARC fallback); "
+	"debug builds only, writable to reset for testing");
+#endif
+
 /*
  * Move "n" bytes at byte address "p"; "rw" indicates the direction
  * of the move, and the I/O parameters are provided in "uio", which is
@@ -237,10 +251,51 @@ zfs_uiomove_iter(void *p, size_t n, zfs_uio_rw_t rw, zfs_uio_t *uio,
 	size_t oldcnt = cnt;
 	int error = 0;
 
-	if (rw == UIO_READ)
-		cnt = copy_to_iter(p, cnt, uio->uio_iter);
-	else
-		cnt = copy_from_iter(p, cnt, uio->uio_iter);
+	/*
+	 * If the uio has pages pinned by the async Direct I/O submission but
+	 * the request is being served through the ARC (UIO_DIRECT cleared by
+	 * a read-time eligibility decline, the page-unaligned tail, or the
+	 * checksum retry), copy into the pinned pages directly rather than
+	 * through the user virtual addresses.  The taskq thread has no user
+	 * mm, and the pinned pages are guaranteed resident, so the copy can
+	 * never fault.  The pinned pages are the user's buffer pages, so the
+	 * data lands in the same place. EOF tail fo a Ordinary synchronous
+	 * Direct I/O read reaches this branch too which is also ok.
+	 */
+	if (rw == UIO_READ && uio->uio_dio.pages != NULL &&
+	    !(uio->uio_extflg & UIO_DIRECT)) {
+		size_t copied = 0;
+
+#ifdef ZFS_DEBUG
+		zfs_async_read_pinned_copies++;
+#endif
+		while (copied < cnt) {
+			size_t rel = uio->uio_loffset + copied -
+			    zfs_uio_soffset(uio);
+			size_t idx = rel >> PAGESHIFT;
+			size_t pgoff = rel & (PAGESIZE - 1);
+			size_t chunk = MIN(cnt - copied, PAGESIZE - pgoff);
+			void *paddr;
+
+			ASSERT3U(idx, <, uio->uio_dio.npages);
+			paddr = zfs_kmap_local(uio->uio_dio.pages[idx]);
+			memcpy((char *)paddr + pgoff, (char *)p + copied,
+			    chunk);
+			zfs_kunmap_local(paddr);
+			copied += chunk;
+		}
+		cnt = copied;
+	} else {
+		if (rw == UIO_READ)
+			cnt = copy_to_iter(p, cnt, uio->uio_iter);
+		else
+			cnt = copy_from_iter(p, cnt, uio->uio_iter);
+
+		if (revert)
+			iov_iter_revert(uio->uio_iter, cnt);
+		else if (cnt != oldcnt)
+			error = EFAULT;
+	}
 
 	/*
 	 * When operating on a full pipe no bytes are processed.
@@ -257,11 +312,6 @@ zfs_uiomove_iter(void *p, size_t n, zfs_uio_rw_t rw, zfs_uio_t *uio,
 	 * copies are allowed for both copy and move but EFAULT should
 	 * be returned for zfs_uiomove().
 	 */
-	if (revert)
-		iov_iter_revert(uio->uio_iter, cnt);
-	else if (cnt != oldcnt)
-		error = EFAULT;
-
 	uio->uio_resid -= cnt;
 	uio->uio_loffset += cnt;
 
@@ -666,6 +716,13 @@ zfs_uio_get_dio_pages_alloc(zfs_uio_t *uio, zfs_uio_rw_t rw)
 		}
 
 		vmem_free(uio->uio_dio.pages, size);
+		/*
+		 * Reset the Direct I/O state so uio->uio_dio.pages non-NULL
+		 * reliably means "pages pinned".
+		 */
+		uio->uio_dio.pages = NULL;
+		uio->uio_dio.npages = 0;
+		uio->uio_dio.pinned = B_FALSE;
 		return (error);
 	} else {
 		ASSERT3S(uio->uio_dio.npages, ==, npages);

module/os/linux/zfs/zfs_vfsops.c

@@ -687,6 +687,9 @@ zfsvfs_create_impl(zfsvfs_t **zfvp, zfsvfs_t *zfsvfs, objset_t *os)
 	list_create(&zfsvfs->z_all_znodes, sizeof (znode_t),
 	    offsetof(znode_t, z_link_node));
 	ZFS_TEARDOWN_INIT(zfsvfs);
+	mutex_init(&zfsvfs->z_async_dio_lock, NULL, MUTEX_DEFAULT, NULL);
+	cv_init(&zfsvfs->z_async_dio_cv, NULL, CV_DEFAULT, NULL);
+	zfsvfs->z_async_dio_inflight = 0;
 	rw_init(&zfsvfs->z_teardown_inactive_lock, NULL, RW_DEFAULT, NULL);
 	rw_init(&zfsvfs->z_fuid_lock, NULL, RW_DEFAULT, NULL);
 
@@ -833,6 +836,9 @@ zfsvfs_free(zfsvfs_t *zfsvfs)
 	mutex_destroy(&zfsvfs->z_lock);
 	list_destroy(&zfsvfs->z_all_znodes);
 	ZFS_TEARDOWN_DESTROY(zfsvfs);
+	ASSERT0(zfsvfs->z_async_dio_inflight);
+	mutex_destroy(&zfsvfs->z_async_dio_lock);
+	cv_destroy(&zfsvfs->z_async_dio_cv);
 	rw_destroy(&zfsvfs->z_teardown_inactive_lock);
 	rw_destroy(&zfsvfs->z_fuid_lock);
 	for (i = 0; i != size; i++) {
@@ -1236,6 +1242,11 @@ zfsvfs_teardown(zfsvfs_t *zfsvfs, boolean_t unmounting)
 
 	ZFS_TEARDOWN_ENTER_WRITE(zfsvfs, FTAG);
 
+	mutex_enter(&zfsvfs->z_async_dio_lock);
+	while (zfsvfs->z_async_dio_inflight != 0)
+		cv_wait(&zfsvfs->z_async_dio_cv, &zfsvfs->z_async_dio_lock);
+	mutex_exit(&zfsvfs->z_async_dio_lock);
+
 	if (!unmounting) {
 		/*
 		 * We purge the parent filesystem's super block as the

module/os/linux/zfs/zfs_vnops_os.c

@@ -4322,6 +4322,21 @@ zfs_getpage(struct inode *ip, struct page *pp)
 	return (error);
 }
 
+/*
+ * Pin the user pages for a Direct I/O request without evaluating any of the
+ * eligibility conditions.  Used by the async read path (zpl_file.c) to pin
+ * at submission time.
+ */
+int
+zfs_dio_pin_pages(zfs_uio_t *uio, zfs_uio_rw_t rw)
+{
+	int error = zfs_uio_get_dio_pages_alloc(uio, rw);
+
+	if (error == 0)
+		uio->uio_extflg &= ~UIO_DIRECT;
+	return (error);
+}
+
 /*
  * Check ZFS specific permissions to memory map a section of a file.
  *