xref: /freebsd/sys/contrib/openzfs/module/zfs/zfs_vnops.c (revision 2f10ffc003be396f3fc23cd2888023896560252b)

// SPDX-License-Identifier: CDDL-1.0
/*
 * This file and its contents are supplied under the terms of the
 * Common Development and Distribution License ("CDDL"), version 1.0.
 * You may only use this file in accordance with the terms of version
 * 1.0 of the CDDL.
 *
 * A full copy of the text of the CDDL should have accompanied this
 * source.  A copy of the CDDL is also available via the Internet at
 * https://opensource.org/license/CDDL-1.0.
 */

/*
 * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
 * Copyright (c) 2012, 2018 by Delphix. All rights reserved.
 * Copyright (c) 2015 by Chunwei Chen. All rights reserved.
 * Copyright 2017 Nexenta Systems, Inc.
 * Copyright (c) 2021, 2022 by Pawel Jakub Dawidek
 * Copyright (c) 2025, Rob Norris <robn@despairlabs.com>
 * Copyright (c) 2025, Klara, Inc.
 */

/* Portions Copyright 2007 Jeremy Teo */
/* Portions Copyright 2010 Robert Milkowski */

#include <sys/types.h>
#include <sys/param.h>
#include <sys/time.h>
#include <sys/sysmacros.h>
#include <sys/vfs.h>
#include <sys/file.h>
#include <sys/stat.h>
#include <sys/kmem.h>
#include <sys/cmn_err.h>
#include <sys/errno.h>
#include <sys/zfs_dir.h>
#include <sys/zfs_acl.h>
#include <sys/zfs_ioctl.h>
#include <sys/fs/zfs.h>
#include <sys/dmu.h>
#include <sys/dmu_objset.h>
#include <sys/dsl_crypt.h>
#include <sys/dsl_dataset.h>
#include <sys/spa.h>
#include <sys/zio_checksum.h>
#include <sys/txg.h>
#include <sys/blkptr.h>
#include <sys/brt.h>
#include <sys/dbuf.h>
#include <sys/policy.h>
#include <sys/zfeature.h>
#include <sys/zfs_vnops.h>
#include <sys/zfs_quota.h>
#include <sys/zfs_vfsops.h>
#include <sys/zfs_znode.h>

/*
 * Enables access to the block cloning feature. If this setting is 0, then even
 * if feature@block_cloning is enabled, using functions and system calls that
 * attempt to clone blocks will act as though the feature is disabled.
 */
int zfs_bclone_enabled = 1;

/*
 * Restricts block cloning between datasets with different properties
 * (checksum, compression, copies, dedup, or special_small_blocks).
 */
int zfs_bclone_strict_properties = 1;

/*
 * When set to 1 the FICLONE and FICLONERANGE ioctls will wait for any dirty
 * data to be written to disk before proceeding. This ensures that the clone
 * operation reliably succeeds, even if a file is modified and then immediately
 * cloned. Note that for small files this may be slower than simply copying
 * the file. When set to 0 the clone operation will immediately fail if it
 * encounters any dirty blocks. By default waiting is enabled.
 */
int zfs_bclone_wait_dirty = 1;

/*
 * Enable Direct I/O. If this setting is 0, then all I/O requests will be
 * directed through the ARC acting as though the dataset property direct was
 * set to disabled.
 */
static int zfs_dio_enabled = 1;

/*
 * Strictly enforce alignment for Direct I/O requests, returning EINVAL
 * if not page-aligned instead of silently falling back to uncached I/O.
 */
static int zfs_dio_strict = 0;


/*
 * Maximum bytes to read per chunk in zfs_read().
 */
#ifdef _ILP32
static uint64_t zfs_vnops_read_chunk_size = 1024 * 1024;
#else
static uint64_t zfs_vnops_read_chunk_size = DMU_MAX_ACCESS / 2;
#endif

int
zfs_fsync(znode_t *zp, int syncflag, cred_t *cr)
{
	int error = 0;
	zfsvfs_t *zfsvfs = ZTOZSB(zp);

	if (zfsvfs->z_os->os_sync != ZFS_SYNC_DISABLED) {
		if ((error = zfs_enter_verify_zp(zfsvfs, zp, FTAG)) != 0)
			return (error);
		error = zil_commit(zfsvfs->z_log, zp->z_id);
		zfs_exit(zfsvfs, FTAG);
	}
	return (error);
}


#if defined(SEEK_HOLE) && defined(SEEK_DATA)
/*
 * Lseek support for finding holes (cmd == SEEK_HOLE) and
 * data (cmd == SEEK_DATA). "off" is an in/out parameter.
 */
static int
zfs_holey_common(znode_t *zp, ulong_t cmd, loff_t *off)
{
	zfs_locked_range_t *lr;
	uint64_t noff = (uint64_t)*off; /* new offset */
	uint64_t file_sz;
	int error;
	boolean_t hole;

	file_sz = zp->z_size;
	if (noff >= file_sz)  {
		return (SET_ERROR(ENXIO));
	}

	if (cmd == F_SEEK_HOLE)
		hole = B_TRUE;
	else
		hole = B_FALSE;

	/* Flush any mmap()'d data to disk */
	if (zn_has_cached_data(zp, 0, file_sz - 1))
		zn_flush_cached_data(zp, B_TRUE);

	lr = zfs_rangelock_enter(&zp->z_rangelock, 0, UINT64_MAX, RL_READER);
	error = dmu_offset_next(ZTOZSB(zp)->z_os, zp->z_id, hole, &noff);
	zfs_rangelock_exit(lr);

	if (error == ESRCH)
		return (SET_ERROR(ENXIO));

	/* File was dirty, so fall back to using generic logic */
	if (error == EBUSY) {
		if (hole)
			*off = file_sz;

		return (0);
	}

	/*
	 * We could find a hole that begins after the logical end-of-file,
	 * because dmu_offset_next() only works on whole blocks.  If the
	 * EOF falls mid-block, then indicate that the "virtual hole"
	 * at the end of the file begins at the logical EOF, rather than
	 * at the end of the last block.
	 */
	if (noff > file_sz) {
		ASSERT(hole);
		noff = file_sz;
	}

	if (noff < *off)
		return (error);
	*off = noff;
	return (error);
}

int
zfs_holey(znode_t *zp, ulong_t cmd, loff_t *off)
{
	zfsvfs_t *zfsvfs = ZTOZSB(zp);
	int error;

	if ((error = zfs_enter_verify_zp(zfsvfs, zp, FTAG)) != 0)
		return (error);

	error = zfs_holey_common(zp, cmd, off);

	zfs_exit(zfsvfs, FTAG);
	return (error);
}
#endif /* SEEK_HOLE && SEEK_DATA */

int
zfs_access(znode_t *zp, int mode, int flag, cred_t *cr)
{
	zfsvfs_t *zfsvfs = ZTOZSB(zp);
	int error;

	if ((error = zfs_enter_verify_zp(zfsvfs, zp, FTAG)) != 0)
		return (error);

	if (flag & V_ACE_MASK)
		error = zfs_zaccess(zp, mode, flag, B_FALSE, cr);
	else
		error = zfs_zaccess_rwx(zp, mode, flag, cr);

	zfs_exit(zfsvfs, FTAG);
	return (error);
}

/*
 * Determine if Direct I/O has been requested (either via the O_DIRECT flag or
 * the "direct" dataset property). When inherited by the property only apply
 * the O_DIRECT flag to correctly aligned IO requests. The rational for this
 * is it allows the property to be safely set on a dataset without forcing
 * all of the applications to be aware of the alignment restrictions. When
 * O_DIRECT is explicitly requested by an application return EINVAL if the
 * request is unaligned.  In all cases, if the range for this request has
 * been mmap'ed then we will perform buffered I/O to keep the mapped region
 * synhronized with the ARC.
 *
 * It is possible that a file's pages could be mmap'ed after it is checked
 * here. If so, that is handled coorarding in zfs_write(). See comments in the
 * following area for how this is handled:
 * zfs_write() -> update_pages()
 */
static int
zfs_setup_direct(struct znode *zp, zfs_uio_t *uio, zfs_uio_rw_t rw,
    int *ioflagp)
{
	zfsvfs_t *zfsvfs = ZTOZSB(zp);
	objset_t *os = zfsvfs->z_os;
	int ioflag = *ioflagp;
	int error = 0;

	if (os->os_direct == ZFS_DIRECT_ALWAYS) {
		/* Force either direct or uncached I/O. */
		ioflag |= O_DIRECT;
	}

	if ((ioflag & O_DIRECT) == 0)
		goto out;

	if (!zfs_dio_enabled || os->os_direct == ZFS_DIRECT_DISABLED) {
		/*
		 * Direct I/O is disabled.  The I/O request will be directed
		 * through the ARC as uncached I/O.
		 */
		goto out;
	}

	if (!zfs_uio_page_aligned(uio) ||
	    !zfs_uio_aligned(uio, PAGE_SIZE)) {
		/*
		 * Misaligned requests can be executed through the ARC as
		 * uncached I/O.  But if O_DIRECT was set by user and we
		 * were set to be strict, then it is a failure.
		 */
		if ((*ioflagp & O_DIRECT) && zfs_dio_strict)
			error = SET_ERROR(EINVAL);
		goto out;
	}

	if (zn_has_cached_data(zp, zfs_uio_offset(uio),
	    zfs_uio_offset(uio) + zfs_uio_resid(uio) - 1)) {
		/*
		 * The region is mmap'ed.  The I/O request will be directed
		 * through the ARC as uncached I/O.
		 */
		goto out;
	}

	/*
	 * The zpl caller declined Direct I/O for this request (a file handle
	 * that already hit a benign DIO read verify failure from a recycled
	 * O_DIRECT buffer).  Fall through to uncached buffered I/O.  Placed
	 * after the ZFS_DIRECT_ALWAYS handling above so direct=always is
	 * declined too.  The verify itself is untouched, so mirror and raidz
	 * self-heal for genuine corruption still runs on the buffered path.
	 */
	if (uio->uio_extflg & UIO_DIO_DENY)
		goto out;

	/*
	 * For short writes the page mapping of Direct I/O makes no sense.
	 * Direct them through the ARC as uncached I/O.
	 */
	if (rw == UIO_WRITE && zfs_uio_resid(uio) < zp->z_blksz)
		goto out;

	error = zfs_uio_get_dio_pages_alloc(uio, rw);
	if (error)
		goto out;
	ASSERT(uio->uio_extflg & UIO_DIRECT);

out:
	*ioflagp = ioflag;
	return (error);
}

/*
 * Read bytes from specified file into supplied buffer.
 *
 *	IN:	zp	- inode of file to be read from.
 *		uio	- structure supplying read location, range info,
 *			  and return buffer.
 *		ioflag	- O_SYNC flags; used to provide FRSYNC semantics.
 *			  O_DIRECT flag; used to bypass page cache.
 *		cr	- credentials of caller.
 *
 *	OUT:	uio	- updated offset and range, buffer filled.
 *
 *	RETURN:	0 on success, error code on failure.
 *
 * Side Effects:
 *	inode - atime updated if byte count > 0
 */
int
zfs_read(struct znode *zp, zfs_uio_t *uio, int ioflag, cred_t *cr)
{
	(void) cr;
	int error = 0;
	boolean_t frsync = B_FALSE;
	boolean_t dio_checksum_failure = B_FALSE;

	zfsvfs_t *zfsvfs = ZTOZSB(zp);
	if ((error = zfs_enter_verify_zp(zfsvfs, zp, FTAG)) != 0)
		return (error);

	if (zp->z_pflags & ZFS_AV_QUARANTINED) {
		zfs_exit(zfsvfs, FTAG);
		return (SET_ERROR(EACCES));
	}

	/* We don't copy out anything useful for directories. */
	if (Z_ISDIR(ZTOTYPE(zp))) {
		zfs_exit(zfsvfs, FTAG);
		return (SET_ERROR(EISDIR));
	}

	/*
	 * Validate file offset
	 */
	if (zfs_uio_offset(uio) < (offset_t)0) {
		zfs_exit(zfsvfs, FTAG);
		return (SET_ERROR(EINVAL));
	}

	/*
	 * Fasttrack empty reads
	 */
	if (zfs_uio_resid(uio) == 0) {
		zfs_exit(zfsvfs, FTAG);
		return (0);
	}

#ifdef FRSYNC
	/*
	 * If we're in FRSYNC mode, sync out this znode before reading it.
	 * Only do this for non-snapshots.
	 *
	 * Some platforms do not support FRSYNC and instead map it
	 * to O_SYNC, which results in unnecessary calls to zil_commit. We
	 * only honor FRSYNC requests on platforms which support it.
	 */
	frsync = !!(ioflag & FRSYNC);
#endif
	if (zfsvfs->z_log &&
	    (frsync || zfsvfs->z_os->os_sync == ZFS_SYNC_ALWAYS)) {
		error = zil_commit(zfsvfs->z_log, zp->z_id);
		if (error != 0) {
			zfs_exit(zfsvfs, FTAG);
			return (error);
		}
	}

	/*
	 * Lock the range against changes.
	 */
	zfs_locked_range_t *lr = zfs_rangelock_enter(&zp->z_rangelock,
	    zfs_uio_offset(uio), zfs_uio_resid(uio), RL_READER);

	/*
	 * If we are reading past end-of-file we can skip
	 * to the end; but we might still need to set atime.
	 */
	if (zfs_uio_offset(uio) >= zp->z_size) {
		error = 0;
		goto out;
	}
	ASSERT(zfs_uio_offset(uio) < zp->z_size);

	/*
	 * Setting up Direct I/O if requested.
	 */
	error = zfs_setup_direct(zp, uio, UIO_READ, &ioflag);
	if (error) {
		goto out;
	}

#if defined(__linux__)
	ssize_t start_offset = zfs_uio_offset(uio);
#endif
	uint_t blksz = zp->z_blksz;
	ssize_t chunk_size;
	ssize_t n = MIN(zfs_uio_resid(uio), zp->z_size - zfs_uio_offset(uio));
	ssize_t start_resid = n;
	ssize_t dio_remaining_resid = 0;

	dmu_flags_t dflags = DMU_READ_PREFETCH;
	if ((ioflag & O_DIRECT) || (uio->uio_extflg & UIO_UNCACHED))
		dflags |= DMU_UNCACHEDIO;
	if (uio->uio_extflg & UIO_DIRECT) {
		/*
		 * All pages for an O_DIRECT request ahve already been mapped
		 * so there's no compelling reason to handle this uio in
		 * smaller chunks.
		 */
		chunk_size = DMU_MAX_ACCESS;

		/*
		 * In the event that the O_DIRECT request is reading the entire
		 * file, it is possible file's length is not page sized
		 * aligned. However, lower layers expect that the Direct I/O
		 * request is page-aligned. In this case, as much of the file
		 * that can be read using Direct I/O happens and the remaining
		 * amount will be read through the ARC.
		 *
		 * This is still consistent with the semantics of Direct I/O in
		 * ZFS as at a minimum the I/O request must be page-aligned.
		 */
		dio_remaining_resid = n - P2ALIGN_TYPED(n, PAGE_SIZE, ssize_t);
		if (dio_remaining_resid != 0)
			n -= dio_remaining_resid;
		dflags |= DMU_DIRECTIO;
	} else {
		chunk_size = MIN(MAX(zfs_vnops_read_chunk_size, blksz),
		    DMU_MAX_ACCESS / 2);
	}

	while (n > 0) {
		ssize_t nbytes = MIN(n, chunk_size -
		    P2PHASE(zfs_uio_offset(uio), blksz));
#ifdef UIO_NOCOPY
		if (zfs_uio_segflg(uio) == UIO_NOCOPY)
			error = mappedread_sf(zp, nbytes, uio);
		else
#endif
		if (zn_has_cached_data(zp, zfs_uio_offset(uio),
		    zfs_uio_offset(uio) + nbytes - 1)) {
			error = mappedread(zp, nbytes, uio);
		} else {
			error = dmu_read_uio_dbuf(sa_get_db(zp->z_sa_hdl),
			    uio, nbytes, dflags);
		}

		if (error) {
			/* convert checksum errors into IO errors */
			if (error == ECKSUM) {
				/*
				 * If a Direct I/O read returned a checksum
				 * verify error, then it must be treated as
				 * suspicious. The contents of the buffer could
				 * have beeen manipulated while the I/O was in
				 * flight. In this case, the remainder of I/O
				 * request will just be reissued through the
				 * ARC.
				 */
				if (uio->uio_extflg & UIO_DIRECT) {
					dio_checksum_failure = B_TRUE;
					uio->uio_extflg &= ~UIO_DIRECT;
					n += dio_remaining_resid;
					dio_remaining_resid = 0;
					continue;
				} else {
					error = SET_ERROR(EIO);
				}
			}

#if defined(__linux__)
			/*
			 * if we actually read some bytes, bubbling EFAULT
			 * up to become EAGAIN isn't what we want here...
			 *
			 * ...on Linux, at least. On FBSD, doing this breaks.
			 */
			if (error == EFAULT &&
			    (zfs_uio_offset(uio) - start_offset) != 0)
				error = 0;
#endif
			break;
		}

		n -= nbytes;
	}

	if (error == 0 && (uio->uio_extflg & UIO_DIRECT) &&
	    dio_remaining_resid != 0) {
		/*
		 * Temporarily remove the UIO_DIRECT flag from the UIO so the
		 * remainder of the file can be read using the ARC.
		 */
		uio->uio_extflg &= ~UIO_DIRECT;
		dflags &= ~DMU_DIRECTIO;

		if (zn_has_cached_data(zp, zfs_uio_offset(uio),
		    zfs_uio_offset(uio) + dio_remaining_resid - 1)) {
			error = mappedread(zp, dio_remaining_resid, uio);
		} else {
			error = dmu_read_uio_dbuf(sa_get_db(zp->z_sa_hdl), uio,
			    dio_remaining_resid, dflags);
		}
		uio->uio_extflg |= UIO_DIRECT;
		dflags |= DMU_DIRECTIO;

		if (error != 0)
			n += dio_remaining_resid;
	} else if (error && (uio->uio_extflg & UIO_DIRECT)) {
		n += dio_remaining_resid;
	}
	int64_t nread = start_resid - n;

	dataset_kstats_update_read_kstats(&zfsvfs->z_kstat, nread);
out:
	zfs_rangelock_exit(lr);

	if (dio_checksum_failure == B_TRUE) {
		uio->uio_extflg |= UIO_DIRECT;
		/*
		 * The DIO read verify failed but the buffered re-read that
		 * followed succeeded, so the on-disk data is good and the
		 * caller mutated its own O_DIRECT buffer in flight.  Report
		 * this outward so the platform layer can decline Direct I/O
		 * for this file handle from here on.  A real on-disk error
		 * would have returned nonzero and is not flagged.
		 */
		if (error == 0)
			uio->uio_extflg |= UIO_DIO_CKSUM_RETRIED;
	}

	/*
	 * Cleanup for Direct I/O if requested.
	 */
	if (uio->uio_extflg & UIO_DIRECT)
		zfs_uio_free_dio_pages(uio, UIO_READ);

	ZFS_ACCESSTIME_STAMP(zfsvfs, zp);
	zfs_exit(zfsvfs, FTAG);
	return (error);
}

static void
zfs_clear_setid_bits_if_necessary(zfsvfs_t *zfsvfs, znode_t *zp, cred_t *cr,
    uint64_t *clear_setid_bits_txgp, dmu_tx_t *tx)
{
	zilog_t *zilog = zfsvfs->z_log;
	const uint64_t uid = KUID_TO_SUID(ZTOUID(zp));

	ASSERT(clear_setid_bits_txgp != NULL);
	ASSERT(tx != NULL);

	/*
	 * Clear Set-UID/Set-GID bits on successful write if not
	 * privileged and at least one of the execute bits is set.
	 *
	 * It would be nice to do this after all writes have
	 * been done, but that would still expose the ISUID/ISGID
	 * to another app after the partial write is committed.
	 *
	 * Note: we don't call zfs_fuid_map_id() here because
	 * user 0 is not an ephemeral uid.
	 */
	mutex_enter(&zp->z_acl_lock);
	if ((zp->z_mode & (S_IXUSR | (S_IXUSR >> 3) | (S_IXUSR >> 6))) != 0 &&
	    (zp->z_mode & (S_ISUID | S_ISGID)) != 0 &&
	    secpolicy_vnode_setid_retain(zp, cr,
	    ((zp->z_mode & S_ISUID) != 0 && uid == 0)) != 0) {
		uint64_t newmode;

		zp->z_mode &= ~(S_ISUID | S_ISGID);
		newmode = zp->z_mode;
		(void) sa_update(zp->z_sa_hdl, SA_ZPL_MODE(zfsvfs),
		    (void *)&newmode, sizeof (uint64_t), tx);

		mutex_exit(&zp->z_acl_lock);

		/*
		 * Make sure SUID/SGID bits will be removed when we replay the
		 * log. If the setid bits are keep coming back, don't log more
		 * than one TX_SETATTR per transaction group.
		 */
		if (*clear_setid_bits_txgp != dmu_tx_get_txg(tx)) {
			vattr_t va = {0};

			va.va_mask = ATTR_MODE;
			va.va_nodeid = zp->z_id;
			va.va_mode = newmode;
			zfs_log_setattr(zilog, tx, TX_SETATTR, zp, &va,
			    ATTR_MODE, NULL);
			*clear_setid_bits_txgp = dmu_tx_get_txg(tx);
		}
	} else {
		mutex_exit(&zp->z_acl_lock);
	}
}

/*
 * Write the bytes to a file.
 *
 *	IN:	zp	- znode of file to be written to.
 *		uio	- structure supplying write location, range info,
 *			  and data buffer.
 *		ioflag	- O_APPEND flag set if in append mode.
 *			  O_DIRECT flag; used to bypass page cache.
 *		cr	- credentials of caller.
 *
 *	OUT:	uio	- updated offset and range.
 *
 *	RETURN:	0 if success
 *		error code if failure
 *
 * Timestamps:
 *	ip - ctime|mtime updated if byte count > 0
 */
int
zfs_write(znode_t *zp, zfs_uio_t *uio, int ioflag, cred_t *cr)
{
	int error = 0, error1;
	ssize_t start_resid = zfs_uio_resid(uio);
	uint64_t clear_setid_bits_txg = 0;
	boolean_t o_direct_defer = B_FALSE;

	/*
	 * Fasttrack empty write
	 */
	ssize_t n = start_resid;
	if (n == 0)
		return (0);

	zfsvfs_t *zfsvfs = ZTOZSB(zp);
	if ((error = zfs_enter_verify_zp(zfsvfs, zp, FTAG)) != 0)
		return (error);

	sa_bulk_attr_t bulk[5];
	int count = 0;
	uint64_t mtime[2], ctime[2];
	SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_MTIME(zfsvfs), NULL, &mtime, 16);
	SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_CTIME(zfsvfs), NULL, &ctime, 16);
	SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_SIZE(zfsvfs), NULL,
	    &zp->z_size, 8);
	SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_FLAGS(zfsvfs), NULL,
	    &zp->z_pflags, 8);
	if (zp->z_is_sa)
		SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_SEQ(zfsvfs), NULL,
		    &zp->z_seq, 8);

	/*
	 * Callers might not be able to detect properly that we are read-only,
	 * so check it explicitly here.
	 */
	if (zfs_is_readonly(zfsvfs)) {
		zfs_exit(zfsvfs, FTAG);
		return (SET_ERROR(EROFS));
	}

	/*
	 * If immutable or not appending then return EPERM.
	 * Intentionally allow ZFS_READONLY through here.
	 * See zfs_zaccess_common()
	 */
	if ((zp->z_pflags & ZFS_IMMUTABLE) ||
	    ((zp->z_pflags & ZFS_APPENDONLY) && !(ioflag & O_APPEND) &&
	    (zfs_uio_offset(uio) < zp->z_size))) {
		zfs_exit(zfsvfs, FTAG);
		return (SET_ERROR(EPERM));
	}

	/*
	 * Validate file offset
	 */
	offset_t woff = ioflag & O_APPEND ? zp->z_size : zfs_uio_offset(uio);
	if (woff < 0) {
		zfs_exit(zfsvfs, FTAG);
		return (SET_ERROR(EINVAL));
	}

	/*
	 * Setting up Direct I/O if requested.
	 */
	error = zfs_setup_direct(zp, uio, UIO_WRITE, &ioflag);
	if (error) {
		zfs_exit(zfsvfs, FTAG);
		return (SET_ERROR(error));
	}

	/*
	 * Pre-fault the pages to ensure slow (eg NFS) pages
	 * don't hold up txg.
	 */
	ssize_t pfbytes = MIN(n, DMU_MAX_ACCESS >> 1);
	if (zfs_uio_prefaultpages(pfbytes, uio)) {
		zfs_exit(zfsvfs, FTAG);
		return (SET_ERROR(EFAULT));
	}

	/*
	 * If in append mode, set the io offset pointer to eof.
	 */
	zfs_locked_range_t *lr;
	if (ioflag & O_APPEND) {
		/*
		 * Obtain an appending range lock to guarantee file append
		 * semantics.  We reset the write offset once we have the lock.
		 */
		lr = zfs_rangelock_enter(&zp->z_rangelock, 0, n, RL_APPEND);
		woff = lr->lr_offset;
		if (lr->lr_length == UINT64_MAX) {
			/*
			 * We overlocked the file because this write will cause
			 * the file block size to increase.
			 * Note that zp_size cannot change with this lock held.
			 */
			woff = zp->z_size;
		}
		zfs_uio_setoffset(uio, woff);
		/*
		 * We need to update the starting offset as well because it is
		 * set previously in the ZPL (Linux) and VNOPS (FreeBSD)
		 * layers.
		 */
		zfs_uio_setsoffset(uio, woff);
	} else {
		/*
		 * Note that if the file block size will change as a result of
		 * this write, then this range lock will lock the entire file
		 * so that we can re-write the block safely.
		 */
		lr = zfs_rangelock_enter(&zp->z_rangelock, woff, n, RL_WRITER);
	}

	if (zn_rlimit_fsize_uio(zp, uio)) {
		zfs_rangelock_exit(lr);
		zfs_exit(zfsvfs, FTAG);
		return (SET_ERROR(EFBIG));
	}

	const rlim64_t limit = MAXOFFSET_T;

	if (woff >= limit) {
		zfs_rangelock_exit(lr);
		zfs_exit(zfsvfs, FTAG);
		return (SET_ERROR(EFBIG));
	}

	if (n > limit - woff)
		n = limit - woff;

	uint64_t end_size = MAX(zp->z_size, woff + n);
	zilog_t *zilog = zfsvfs->z_log;
	boolean_t commit = (ioflag & (O_SYNC | O_DSYNC)) ||
	    (zfsvfs->z_os->os_sync == ZFS_SYNC_ALWAYS);

	const uint64_t uid = KUID_TO_SUID(ZTOUID(zp));
	const uint64_t gid = KGID_TO_SGID(ZTOGID(zp));
	const uint64_t projid = zp->z_projid;

	/*
	 * In the event we are increasing the file block size
	 * (lr_length == UINT64_MAX), we will direct the write to the ARC.
	 * Because zfs_grow_blocksize() will read from the ARC in order to
	 * grow the dbuf, we avoid doing Direct I/O here as that would cause
	 * data written to disk to be overwritten by data in the ARC during
	 * the sync phase. Besides writing data twice to disk, we also
	 * want to avoid consistency concerns between data in the the ARC and
	 * on disk while growing the file's blocksize.
	 *
	 * We will only temporarily remove Direct I/O and put it back after
	 * we have grown the blocksize. We do this in the event a request
	 * is larger than max_blksz, so further requests to
	 * dmu_write_uio_dbuf() will still issue the requests using Direct
	 * IO.
	 *
	 * As an example:
	 * The first block to file is being written as a 4k request with
	 * a recorsize of 1K. The first 1K issued in the loop below will go
	 * through the ARC; however, the following 3 1K requests will
	 * use Direct I/O.
	 */
	if (uio->uio_extflg & UIO_DIRECT && lr->lr_length == UINT64_MAX) {
		uio->uio_extflg &= ~UIO_DIRECT;
		o_direct_defer = B_TRUE;
	}

	/*
	 * Write the file in reasonable size chunks.  Each chunk is written
	 * in a separate transaction; this keeps the intent log records small
	 * and allows us to do more fine-grained space accounting.
	 */
	while (n > 0) {
		woff = zfs_uio_offset(uio);

		if (zfs_id_overblockquota(zfsvfs, DMU_USERUSED_OBJECT, uid) ||
		    zfs_id_overblockquota(zfsvfs, DMU_GROUPUSED_OBJECT, gid) ||
		    (projid != ZFS_DEFAULT_PROJID &&
		    zfs_id_overblockquota(zfsvfs, DMU_PROJECTUSED_OBJECT,
		    projid))) {
			error = SET_ERROR(EDQUOT);
			break;
		}

		uint64_t blksz;
		if (lr->lr_length == UINT64_MAX && zp->z_size <= zp->z_blksz) {
			if (zp->z_blksz > zfsvfs->z_max_blksz &&
			    !ISP2(zp->z_blksz)) {
				/*
				 * File's blocksize is already larger than the
				 * "recordsize" property.  Only let it grow to
				 * the next power of 2.
				 */
				blksz = 1 << highbit64(zp->z_blksz);
			} else {
				blksz = zfsvfs->z_max_blksz;
			}
			blksz = MIN(blksz, P2ROUNDUP(end_size,
			    SPA_MINBLOCKSIZE));
			blksz = MAX(blksz, zp->z_blksz);
		} else {
			blksz = zp->z_blksz;
		}

		arc_buf_t *abuf = NULL;
		ssize_t nbytes = n;
		if (n >= blksz && woff >= zp->z_size &&
		    P2PHASE(woff, blksz) == 0 &&
		    !(uio->uio_extflg & UIO_DIRECT) &&
		    (blksz >= SPA_OLD_MAXBLOCKSIZE || n < 4 * blksz)) {
			/*
			 * This write covers a full block.  "Borrow" a buffer
			 * from the dmu so that we can fill it before we enter
			 * a transaction.  This avoids the possibility of
			 * holding up the transaction if the data copy hangs
			 * up on a pagefault (e.g., from an NFS server mapping).
			 */
			abuf = dmu_request_arcbuf(sa_get_db(zp->z_sa_hdl),
			    blksz);
			ASSERT(abuf != NULL);
			ASSERT(arc_buf_size(abuf) == blksz);
			if ((error = zfs_uiocopy(abuf->b_data, blksz,
			    UIO_WRITE, uio, &nbytes))) {
				dmu_return_arcbuf(abuf);
				break;
			}
			ASSERT3S(nbytes, ==, blksz);
		} else {
			nbytes = MIN(n, (DMU_MAX_ACCESS >> 1) -
			    P2PHASE(woff, blksz));
			if (pfbytes < nbytes) {
				if (zfs_uio_prefaultpages(nbytes, uio)) {
					error = SET_ERROR(EFAULT);
					break;
				}
				pfbytes = nbytes;
			}
		}

		/*
		 * Start a transaction.
		 */
		dmu_tx_t *tx = dmu_tx_create(zfsvfs->z_os);
		dmu_tx_hold_sa(tx, zp->z_sa_hdl, ZFS_SEQ_MAY_GROW(zp));
		dmu_buf_impl_t *db = (dmu_buf_impl_t *)sa_get_db(zp->z_sa_hdl);
		DB_DNODE_ENTER(db);
		dmu_tx_hold_write_by_dnode(tx, DB_DNODE(db), woff, nbytes);
		DB_DNODE_EXIT(db);
		zfs_sa_upgrade_txholds(tx, zp);
		error = dmu_tx_assign(tx, DMU_TX_WAIT);
		if (error) {
			dmu_tx_abort(tx);
			if (abuf != NULL)
				dmu_return_arcbuf(abuf);
			break;
		}

		/*
		 * NB: We must call zfs_clear_setid_bits_if_necessary before
		 * committing the transaction!
		 */

		/*
		 * If rangelock_enter() over-locked we grow the blocksize
		 * and then reduce the lock range.  This will only happen
		 * on the first iteration since rangelock_reduce() will
		 * shrink down lr_length to the appropriate size.
		 */
		if (lr->lr_length == UINT64_MAX) {
			zfs_grow_blocksize(zp, blksz, tx);
			zfs_rangelock_reduce(lr, woff, n);
		}

		dmu_flags_t dflags = DMU_READ_PREFETCH;
		if ((ioflag & O_DIRECT) || (uio->uio_extflg & UIO_UNCACHED))
			dflags |= DMU_UNCACHEDIO;
		if (uio->uio_extflg & UIO_DIRECT)
			dflags |= DMU_DIRECTIO;

		ssize_t tx_bytes;
		if (abuf == NULL) {
			tx_bytes = zfs_uio_resid(uio);
			zfs_uio_fault_disable(uio, B_TRUE);
			error = dmu_write_uio_dbuf(sa_get_db(zp->z_sa_hdl),
			    uio, nbytes, tx, dflags);
			zfs_uio_fault_disable(uio, B_FALSE);
#ifdef __linux__
			if (error == EFAULT) {
				zfs_clear_setid_bits_if_necessary(zfsvfs, zp,
				    cr, &clear_setid_bits_txg, tx);
				dmu_tx_commit(tx);
				/*
				 * Account for partial writes before
				 * continuing the loop.
				 * Update needs to occur before the next
				 * zfs_uio_prefaultpages, or prefaultpages may
				 * error, and we may break the loop early.
				 */
				n -= tx_bytes - zfs_uio_resid(uio);
				/*
				 * The prefaulted pages still faulted, so force
				 * a re-prefault on the next pass.  If they can
				 * no longer be faulted in (e.g. the calling
				 * process is exiting), zfs_uio_prefaultpages()
				 * fails and the loop breaks with EFAULT instead
				 * of spinning on them forever.
				 */
				pfbytes = 0;
				continue;
			}
#endif
			/*
			 * On FreeBSD, EFAULT should be propagated back to the
			 * VFS, which will handle faulting and will retry.
			 */
			if (error != 0 && error != EFAULT) {
				zfs_clear_setid_bits_if_necessary(zfsvfs, zp,
				    cr, &clear_setid_bits_txg, tx);
				dmu_tx_commit(tx);
				break;
			}
			tx_bytes -= zfs_uio_resid(uio);
		} else {
			/*
			 * Thus, we're writing a full block at a block-aligned
			 * offset and extending the file past EOF.
			 *
			 * dmu_assign_arcbuf_by_dbuf() will directly assign the
			 * arc buffer to a dbuf.
			 */
			error = dmu_assign_arcbuf_by_dbuf(
			    sa_get_db(zp->z_sa_hdl), woff, abuf, tx, dflags);
			if (error != 0) {
				/*
				 * XXX This might not be necessary if
				 * dmu_assign_arcbuf_by_dbuf is guaranteed
				 * to be atomic.
				 */
				zfs_clear_setid_bits_if_necessary(zfsvfs, zp,
				    cr, &clear_setid_bits_txg, tx);
				dmu_return_arcbuf(abuf);
				dmu_tx_commit(tx);
				break;
			}
			ASSERT3S(nbytes, <=, zfs_uio_resid(uio));
			zfs_uioskip(uio, nbytes);
			tx_bytes = nbytes;
		}
		/*
		 * There is a window where a file's pages can be mmap'ed after
		 * zfs_setup_direct() is called. This is due to the fact that
		 * the rangelock in this function is acquired after calling
		 * zfs_setup_direct(). This is done so that
		 * zfs_uio_prefaultpages() does not attempt to fault in pages
		 * on Linux for Direct I/O requests. This is not necessary as
		 * the pages are pinned in memory and can not be faulted out.
		 * Ideally, the rangelock would be held before calling
		 * zfs_setup_direct() and zfs_uio_prefaultpages(); however,
		 * this can lead to a deadlock as zfs_getpage() also acquires
		 * the rangelock as a RL_WRITER and prefaulting the pages can
		 * lead to zfs_getpage() being called.
		 *
		 * In the case of the pages being mapped after
		 * zfs_setup_direct() is called, the call to update_pages()
		 * will still be made to make sure there is consistency between
		 * the ARC and the Linux page cache. This is an ufortunate
		 * situation as the data will be read back into the ARC after
		 * the Direct I/O write has completed, but this is the penality
		 * for writing to a mmap'ed region of a file using Direct I/O.
		 */
		if (tx_bytes &&
		    zn_has_cached_data(zp, woff, woff + tx_bytes - 1)) {
			update_pages(zp, woff, tx_bytes, zfsvfs->z_os);
		}

		/*
		 * If we made no progress, we're done.  If we made even
		 * partial progress, update the znode and ZIL accordingly.
		 */
		if (tx_bytes == 0) {
			(void) sa_update(zp->z_sa_hdl, SA_ZPL_SIZE(zfsvfs),
			    (void *)&zp->z_size, sizeof (uint64_t), tx);
			dmu_tx_commit(tx);
			ASSERT(error != 0);
			break;
		}

		zfs_clear_setid_bits_if_necessary(zfsvfs, zp, cr,
		    &clear_setid_bits_txg, tx);

		zfs_tstamp_update_setup(zp, CONTENT_MODIFIED, mtime, ctime);
		if (zp->z_is_sa)
			zp->z_has_seq = B_TRUE;

		/*
		 * Update the file size (zp_size) if it has changed;
		 * account for possible concurrent updates.
		 */
		while ((end_size = zp->z_size) < zfs_uio_offset(uio)) {
			(void) atomic_cas_64(&zp->z_size, end_size,
			    zfs_uio_offset(uio));
			ASSERT(error == 0 || error == EFAULT);
		}
		/*
		 * If we are replaying and eof is non zero then force
		 * the file size to the specified eof. Note, there's no
		 * concurrency during replay.
		 */
		if (zfsvfs->z_replay && zfsvfs->z_replay_eof != 0)
			zp->z_size = zfsvfs->z_replay_eof;

		ASSERT3S(count, <=, ARRAY_SIZE(bulk));
		error1 = sa_bulk_update(zp->z_sa_hdl, bulk, count, tx);
		if (error1 != 0)
			/* Avoid clobbering EFAULT. */
			error = error1;

		/*
		 * NB: During replay, the TX_SETATTR record logged by
		 * zfs_clear_setid_bits_if_necessary must precede any of
		 * the TX_WRITE records logged here.
		 */
		zfs_log_write(zilog, tx, TX_WRITE, zp, woff, tx_bytes, commit,
		    uio->uio_extflg & UIO_DIRECT ? B_TRUE : B_FALSE, NULL,
		    NULL);

		dmu_tx_commit(tx);

		/*
		 * Direct I/O was deferred in order to grow the first block.
		 * At this point it can be re-enabled for subsequent writes.
		 */
		if (o_direct_defer) {
			ASSERT(ioflag & O_DIRECT);
			uio->uio_extflg |= UIO_DIRECT;
			o_direct_defer = B_FALSE;
		}

		if (error != 0)
			break;
		ASSERT3S(tx_bytes, ==, nbytes);
		n -= nbytes;
		pfbytes -= nbytes;
	}

	if (o_direct_defer) {
		ASSERT(ioflag & O_DIRECT);
		uio->uio_extflg |= UIO_DIRECT;
		o_direct_defer = B_FALSE;
	}

	zfs_znode_update_vfs(zp);
	zfs_rangelock_exit(lr);

	/*
	 * Cleanup for Direct I/O if requested.
	 */
	if (uio->uio_extflg & UIO_DIRECT)
		zfs_uio_free_dio_pages(uio, UIO_WRITE);

	/*
	 * If we're in replay mode, or we made no progress, or the
	 * uio data is inaccessible return an error.  Otherwise, it's
	 * at least a partial write, so it's successful.
	 */
	if (zfsvfs->z_replay || zfs_uio_resid(uio) == start_resid ||
	    error == EFAULT) {
		zfs_exit(zfsvfs, FTAG);
		return (error);
	}

	if (commit) {
		error = zil_commit(zilog, zp->z_id);
		if (error != 0) {
			zfs_exit(zfsvfs, FTAG);
			return (error);
		}
	}

	int64_t nwritten = start_resid - zfs_uio_resid(uio);
	dataset_kstats_update_write_kstats(&zfsvfs->z_kstat, nwritten);

	zfs_exit(zfsvfs, FTAG);
	return (0);
}

/*
 * Check if a block should be skipped during rewrite.
 * Returns B_TRUE if block should be skipped.
 */
static boolean_t
zfs_rewrite_skip(dmu_buf_t *db, objset_t *os, uint64_t flags)
{
	/*
	 * This may be slightly stale and racy, but should be OK for
	 * the advisory use.
	 */
	blkptr_t *bp = dmu_buf_get_blkptr(db);
	if (bp == NULL)
		return (B_TRUE);

	if (flags & ZFS_REWRITE_SKIP_SNAPSHOT) {
		if (dmu_objset_block_is_shared(os, bp))
			return (B_TRUE);
	}

	if (flags & ZFS_REWRITE_SKIP_BRT) {
		if (brt_maybe_exists(os->os_spa, bp))
			return (B_TRUE);
	}

	return (B_FALSE);
}

/*
 * Rewrite a range of file as-is without modification.
 *
 *	IN:	zp	- znode of file to be rewritten.
 *		off	- Offset of the range to rewrite.
 *		len	- Length of the range to rewrite.
 *		flags	- Random rewrite parameters.
 *		arg	- flags-specific argument.
 *
 *	RETURN:	0 if success
 *		error code if failure
 */
int
zfs_rewrite(znode_t *zp, uint64_t off, uint64_t len, uint64_t flags,
    uint64_t arg)
{
	int error;

#define	ZFS_REWRITE_VALID_FLAGS \
	(ZFS_REWRITE_PHYSICAL | ZFS_REWRITE_SKIP_SNAPSHOT | \
	ZFS_REWRITE_SKIP_BRT)

	if ((flags & ~ZFS_REWRITE_VALID_FLAGS) != 0 || arg != 0)
		return (SET_ERROR(EINVAL));

	zfsvfs_t *zfsvfs = ZTOZSB(zp);
	if ((error = zfs_enter_verify_zp(zfsvfs, zp, FTAG)) != 0)
		return (error);

	/* Check if physical rewrite is allowed */
	spa_t *spa = zfsvfs->z_os->os_spa;
	if ((flags & ZFS_REWRITE_PHYSICAL) &&
	    !spa_feature_is_enabled(spa, SPA_FEATURE_PHYSICAL_REWRITE)) {
		zfs_exit(zfsvfs, FTAG);
		return (SET_ERROR(ENOTSUP));
	}

	if (zfs_is_readonly(zfsvfs)) {
		zfs_exit(zfsvfs, FTAG);
		return (SET_ERROR(EROFS));
	}

	if (off >= zp->z_size) {
		zfs_exit(zfsvfs, FTAG);
		return (0);
	}
	if (len == 0 || len > zp->z_size - off)
		len = zp->z_size - off;

	/* Flush any mmap()'d data to disk */
	if (zn_has_cached_data(zp, off, off + len - 1))
		zn_flush_cached_data(zp, B_TRUE);

	zfs_locked_range_t *lr;
	lr = zfs_rangelock_enter(&zp->z_rangelock, off, len, RL_WRITER);

	const uint64_t uid = KUID_TO_SUID(ZTOUID(zp));
	const uint64_t gid = KGID_TO_SGID(ZTOGID(zp));
	const uint64_t projid = zp->z_projid;

	dmu_buf_impl_t *db = (dmu_buf_impl_t *)sa_get_db(zp->z_sa_hdl);
	DB_DNODE_ENTER(db);
	dnode_t *dn = DB_DNODE(db);

	uint64_t n, noff = off, nr = 0, nw = 0;
	while (len > 0) {
		/*
		 * Rewrite only actual data, skipping any holes.  This might
		 * be inaccurate for dirty files, but we don't really care.
		 */
		if (noff == off) {
			/* Find next data in the file. */
			error = dnode_next_offset(dn, 0, &noff, 1, 1, 0);
			if (error || noff >= off + len) {
				if (error == ESRCH)	/* No more data. */
					error = 0;
				break;
			}
			ASSERT3U(noff, >=, off);
			len -= noff - off;
			off = noff;

			/* Find where the data end. */
			error = dnode_next_offset(dn, DNODE_FIND_HOLE, &noff,
			    1, 1, 0);
			if (error != 0)
				noff = off + len;
		}
		ASSERT3U(noff, >, off);

		if (zfs_id_overblockquota(zfsvfs, DMU_USERUSED_OBJECT, uid) ||
		    zfs_id_overblockquota(zfsvfs, DMU_GROUPUSED_OBJECT, gid) ||
		    (projid != ZFS_DEFAULT_PROJID &&
		    zfs_id_overblockquota(zfsvfs, DMU_PROJECTUSED_OBJECT,
		    projid))) {
			error = SET_ERROR(EDQUOT);
			break;
		}

		n = MIN(MIN(len, noff - off),
		    DMU_MAX_ACCESS / 2 - P2PHASE(off, zp->z_blksz));

		dmu_tx_t *tx = dmu_tx_create(zfsvfs->z_os);
		dmu_tx_hold_write_by_dnode(tx, dn, off, n);
		error = dmu_tx_assign(tx, DMU_TX_WAIT);
		if (error) {
			dmu_tx_abort(tx);
			break;
		}

		/* Mark all dbufs within range as dirty to trigger rewrite. */
		dmu_buf_t **dbp;
		int numbufs;
		error = dmu_buf_hold_array_by_dnode(dn, off, n, TRUE, FTAG,
		    &numbufs, &dbp, DMU_READ_PREFETCH | DMU_UNCACHEDIO);
		if (error) {
			dmu_tx_commit(tx);
			break;
		}
		for (int i = 0; i < numbufs; i++) {
			nr += dbp[i]->db_size;
			if (dmu_buf_is_dirty(dbp[i], tx))
				continue;

			if (zfs_rewrite_skip(dbp[i], zfsvfs->z_os, flags))
				continue;

			nw += dbp[i]->db_size;
			if (flags & ZFS_REWRITE_PHYSICAL)
				dmu_buf_will_rewrite(dbp[i], tx);
			else
				dmu_buf_will_dirty(dbp[i], tx);
		}
		dmu_buf_rele_array(dbp, numbufs, FTAG);

		dmu_tx_commit(tx);

		len -= n;
		off += n;

		if (issig()) {
			error = SET_ERROR(EINTR);
			break;
		}
	}

	DB_DNODE_EXIT(db);

	dataset_kstats_update_read_kstats(&zfsvfs->z_kstat, nr);
	dataset_kstats_update_write_kstats(&zfsvfs->z_kstat, nw);

	zfs_rangelock_exit(lr);
	zfs_exit(zfsvfs, FTAG);
	return (error);
}

int
zfs_getsecattr(znode_t *zp, vsecattr_t *vsecp, int flag, cred_t *cr)
{
	zfsvfs_t *zfsvfs = ZTOZSB(zp);
	int error;
	boolean_t skipaclchk = (flag & ATTR_NOACLCHECK) ? B_TRUE : B_FALSE;

	if ((error = zfs_enter_verify_zp(zfsvfs, zp, FTAG)) != 0)
		return (error);
	error = zfs_getacl(zp, vsecp, skipaclchk, cr);
	zfs_exit(zfsvfs, FTAG);

	return (error);
}

int
zfs_setsecattr(znode_t *zp, vsecattr_t *vsecp, int flag, cred_t *cr)
{
	zfsvfs_t *zfsvfs = ZTOZSB(zp);
	int error;
	boolean_t skipaclchk = (flag & ATTR_NOACLCHECK) ? B_TRUE : B_FALSE;
	zilog_t	*zilog;

	if ((error = zfs_enter_verify_zp(zfsvfs, zp, FTAG)) != 0)
		return (error);
	zilog = zfsvfs->z_log;
	error = zfs_setacl(zp, vsecp, skipaclchk, cr);

	if (error == 0 && zfsvfs->z_os->os_sync == ZFS_SYNC_ALWAYS)
		error = zil_commit(zilog, 0);

	zfs_exit(zfsvfs, FTAG);
	return (error);
}

/*
 * Get the optimal alignment to ensure direct IO can be performed without
 * incurring any RMW penalty on write. If direct IO is not enabled for this
 * file, returns an error.
 */
int
zfs_get_direct_alignment(znode_t *zp, uint64_t *alignp)
{
	zfsvfs_t *zfsvfs = ZTOZSB(zp);

	if (!zfs_dio_enabled || zfsvfs->z_os->os_direct == ZFS_DIRECT_DISABLED)
		return (SET_ERROR(EOPNOTSUPP));

	/*
	 * If the file has multiple blocks, then its block size is fixed
	 * forever, and so is the ideal alignment.
	 *
	 * If however it only has a single block, then we want to return the
	 * max block size it could possibly grown to (ie, the dataset
	 * recordsize). We do this so that a program querying alignment
	 * immediately after the file is created gets a value that won't change
	 * once the file has grown into the second block and beyond.
	 *
	 * Because we don't have a count of blocks easily available here, we
	 * check if the apparent file size is smaller than its current block
	 * size (meaning, the file hasn't yet grown into the current block
	 * size) and then, check if the block size is smaller than the dataset
	 * maximum (meaning, if the file grew past the current block size, the
	 * block size could would be increased).
	 */
	if (zp->z_size <= zp->z_blksz && zp->z_blksz < zfsvfs->z_max_blksz)
		*alignp = MAX(zfsvfs->z_max_blksz, PAGE_SIZE);
	else
		*alignp = MAX(zp->z_blksz, PAGE_SIZE);

	return (0);
}

#ifdef ZFS_DEBUG
static int zil_fault_io = 0;
#endif

static void zfs_get_done(zgd_t *zgd, int error);

/*
 * Get data to generate a TX_WRITE intent log record.
 */
int
zfs_get_data(void *arg, uint64_t gen, lr_write_t *lr, char *buf,
    struct lwb *lwb, zio_t *zio)
{
	zfsvfs_t *zfsvfs = arg;
	objset_t *os = zfsvfs->z_os;
	znode_t *zp;
	uint64_t object = lr->lr_foid;
	uint64_t offset = lr->lr_offset;
	uint64_t size = lr->lr_length;
	zgd_t *zgd;
	int error = 0;
	uint64_t zp_gen;

	ASSERT3P(lwb, !=, NULL);
	ASSERT3U(size, !=, 0);

	/*
	 * Nothing to do if the file has been removed
	 */
	if (zfs_zget(zfsvfs, object, &zp) != 0)
		return (SET_ERROR(ENOENT));
	if (zp->z_unlinked) {
		/*
		 * Release the vnode asynchronously as we currently have the
		 * txg stopped from syncing.
		 */
		zfs_zrele_async(zp);
		return (SET_ERROR(ENOENT));
	}
	/* check if generation number matches */
	if (sa_lookup(zp->z_sa_hdl, SA_ZPL_GEN(zfsvfs), &zp_gen,
	    sizeof (zp_gen)) != 0) {
		zfs_zrele_async(zp);
		return (SET_ERROR(EIO));
	}
	if (zp_gen != gen) {
		zfs_zrele_async(zp);
		return (SET_ERROR(ENOENT));
	}

	zgd = kmem_zalloc(sizeof (zgd_t), KM_SLEEP);
	zgd->zgd_lwb = lwb;
	zgd->zgd_private = zp;

	/*
	 * Write records come in two flavors: immediate and indirect.
	 * For small writes it's cheaper to store the data with the
	 * log record (immediate); for large writes it's cheaper to
	 * sync the data and get a pointer to it (indirect) so that
	 * we don't have to write the data twice.
	 */
	if (buf != NULL) { /* immediate write */
		zgd->zgd_lr = zfs_rangelock_enter(&zp->z_rangelock, offset,
		    size, RL_READER);
		/* test for truncation needs to be done while range locked */
		if (offset >= zp->z_size) {
			error = SET_ERROR(ENOENT);
		} else {
			error = dmu_read(os, object, offset, size, buf,
			    DMU_READ_NO_PREFETCH | DMU_KEEP_CACHING);
		}
		ASSERT(error == 0 || error == ENOENT);
	} else { /* indirect write */
		ASSERT3P(zio, !=, NULL);
		/*
		 * Have to lock the whole block to ensure when it's
		 * written out and its checksum is being calculated
		 * that no one can change the data. We need to re-check
		 * blocksize after we get the lock in case it's changed!
		 */
		for (;;) {
			uint64_t blkoff;
			size = zp->z_blksz;
			blkoff = ISP2(size) ? P2PHASE(offset, size) : offset;
			offset -= blkoff;
			zgd->zgd_lr = zfs_rangelock_enter(&zp->z_rangelock,
			    offset, size, RL_READER);
			if (zp->z_blksz == size)
				break;
			offset += blkoff;
			zfs_rangelock_exit(zgd->zgd_lr);
		}
		/* test for truncation needs to be done while range locked */
		if (lr->lr_offset >= zp->z_size)
			error = SET_ERROR(ENOENT);
#ifdef ZFS_DEBUG
		if (zil_fault_io) {
			error = SET_ERROR(EIO);
			zil_fault_io = 0;
		}
#endif

		dmu_buf_t *dbp;
		if (error == 0)
			error = dmu_buf_hold_noread(os, object, offset, zgd,
			    &dbp);

		if (error == 0) {
			zgd->zgd_db = dbp;
			dmu_buf_impl_t *db = (dmu_buf_impl_t *)dbp;
			boolean_t direct_write = B_FALSE;
			mutex_enter(&db->db_mtx);
			dbuf_dirty_record_t *dr =
			    dbuf_find_dirty_eq(db, lr->lr_common.lrc_txg);
			if (dr != NULL && dr->dt.dl.dr_diowrite)
				direct_write = B_TRUE;
			mutex_exit(&db->db_mtx);

			/*
			 * All Direct I/O writes will have already completed and
			 * the block pointer can be immediately stored in the
			 * log record.
			 */
			if (direct_write) {
				/*
				 * A Direct I/O write always covers an entire
				 * block.
				 */
				ASSERT3U(dbp->db_size, ==, zp->z_blksz);
				lr->lr_blkptr = dr->dt.dl.dr_overridden_by;
				zfs_get_done(zgd, 0);
				return (0);
			}

			blkptr_t *bp = &lr->lr_blkptr;
			zgd->zgd_bp = bp;

			ASSERT3U(dbp->db_offset, ==, offset);
			ASSERT3U(dbp->db_size, ==, size);

			error = dmu_sync(zio, lr->lr_common.lrc_txg,
			    zfs_get_done, zgd);
			ASSERT(error || lr->lr_length <= size);

			/*
			 * On success, we need to wait for the write I/O
			 * initiated by dmu_sync() to complete before we can
			 * release this dbuf.  We will finish everything up
			 * in the zfs_get_done() callback.
			 */
			if (error == 0)
				return (0);

			if (error == EALREADY) {
				lr->lr_common.lrc_txtype = TX_WRITE2;
				/*
				 * TX_WRITE2 relies on the data previously
				 * written by the TX_WRITE that caused
				 * EALREADY.  We zero out the BP because
				 * it is the old, currently-on-disk BP.
				 */
				zgd->zgd_bp = NULL;
				BP_ZERO(bp);
				error = 0;
			}
		}
	}

	zfs_get_done(zgd, error);

	return (error);
}

static void
zfs_get_done(zgd_t *zgd, int error)
{
	(void) error;
	znode_t *zp = zgd->zgd_private;

	if (zgd->zgd_db)
		dmu_buf_rele(zgd->zgd_db, zgd);

	zfs_rangelock_exit(zgd->zgd_lr);

	/*
	 * Release the vnode asynchronously as we currently have the
	 * txg stopped from syncing.
	 */
	zfs_zrele_async(zp);

	kmem_free(zgd, sizeof (zgd_t));
}

static int
zfs_enter_two(zfsvfs_t *zfsvfs1, zfsvfs_t *zfsvfs2, const char *tag)
{
	int error;

	/* Swap. Not sure if the order of zfs_enter()s is important. */
	if (zfsvfs1 > zfsvfs2) {
		zfsvfs_t *tmpzfsvfs;

		tmpzfsvfs = zfsvfs2;
		zfsvfs2 = zfsvfs1;
		zfsvfs1 = tmpzfsvfs;
	}

	error = zfs_enter(zfsvfs1, tag);
	if (error != 0)
		return (error);
	if (zfsvfs1 != zfsvfs2) {
		error = zfs_enter(zfsvfs2, tag);
		if (error != 0) {
			zfs_exit(zfsvfs1, tag);
			return (error);
		}
	}

	return (0);
}

static void
zfs_exit_two(zfsvfs_t *zfsvfs1, zfsvfs_t *zfsvfs2, const char *tag)
{

	zfs_exit(zfsvfs1, tag);
	if (zfsvfs1 != zfsvfs2)
		zfs_exit(zfsvfs2, tag);
}

/*
 * Checks shared by zfs_clone_range() and zfs_dedupe_range() that do not
 * depend on the offsets or length of the request.  Both datasets must
 * already be entered with zfs_enter_two().
 */
static int
zfs_clone_range_precheck(znode_t *inzp, znode_t *outzp)
{
	zfsvfs_t	*inzfsvfs = ZTOZSB(inzp);
	zfsvfs_t	*outzfsvfs = ZTOZSB(outzp);
	objset_t	*inos = inzfsvfs->z_os;
	objset_t	*outos = outzfsvfs->z_os;
	int		error;

	/*
	 * Both source and destination have to belong to the same storage pool.
	 */
	if (dmu_objset_spa(inos) != dmu_objset_spa(outos))
		return (SET_ERROR(EXDEV));

	/*
	 * outos and inos belongs to the same storage pool.
	 * see a few lines above, only one check.
	 */
	if (!spa_feature_is_enabled(dmu_objset_spa(outos),
	    SPA_FEATURE_BLOCK_CLONING))
		return (SET_ERROR(EOPNOTSUPP));

	ASSERT(!outzfsvfs->z_replay);

	/*
	 * Block cloning from an unencrypted dataset into an encrypted
	 * dataset and vice versa is not supported.
	 */
	if (inos->os_encrypted != outos->os_encrypted)
		return (SET_ERROR(EXDEV));

	/*
	 * Cloning across encrypted datasets is possible only if they
	 * share the same master key.
	 */
	if (inos != outos && inos->os_encrypted &&
	    !dmu_objset_crypto_key_equal(inos, outos))
		return (SET_ERROR(EXDEV));

	/*
	 * Cloning between datasets with different properties is possible,
	 * but it may cause confusions when copying data between them and
	 * expecting new properties to apply.
	 */
	if (zfs_bclone_strict_properties && inos != outos &&
	    !inzfsvfs->z_issnap &&
	    (inos->os_checksum != outos->os_checksum ||
	    inos->os_compress != outos->os_compress ||
	    inos->os_copies != outos->os_copies ||
	    inos->os_dedup_checksum != outos->os_dedup_checksum))
		return (SET_ERROR(EXDEV));

	error = zfs_verify_zp(inzp);
	if (error == 0)
		error = zfs_verify_zp(outzp);
	if (error != 0)
		return (error);

	/*
	 * We don't copy source file's flags that's why we don't allow to clone
	 * files that are in quarantine.
	 */
	if (inzp->z_pflags & ZFS_AV_QUARANTINED)
		return (SET_ERROR(EACCES));

	return (0);
}

/*
 * Clone a range of blocks from inzp into outzp.  The caller must have entered
 * both datasets, resolved the request against the source EOF, and be holding a
 * RL_READER range lock on inzp and a RL_WRITER range lock (outlr) on outzp.
 * The number of bytes cloned is returned via donep; the caller is responsible
 * for the trailing accounting (access time, zil_commit) and for dropping the
 * range locks.
 *
 * When dedup is set the caller is implementing FIDEDUPERANGE, where the file
 * content is unchanged: in that case we must not update the destination's
 * mtime/ctime or strip its setid bits, matching the Linux convention that
 * REMAP_FILE_DEDUP skips file_modified().
 */
static int
zfs_clone_range_locked(znode_t *inzp, uint64_t inoff, znode_t *outzp,
    uint64_t outoff, uint64_t len, cred_t *cr, zfs_locked_range_t *outlr,
    boolean_t dedup, uint64_t *donep)
{
	zfsvfs_t	*inzfsvfs = ZTOZSB(inzp);
	zfsvfs_t	*outzfsvfs = ZTOZSB(outzp);
	objset_t	*inos = inzfsvfs->z_os;
	objset_t	*outos = outzfsvfs->z_os;
	dmu_buf_impl_t	*db;
	dmu_tx_t	*tx;
	zilog_t		*zilog = outzfsvfs->z_log;
	uint64_t	done = 0;
	uint64_t	outsize, size;
	int		error = 0;
	int		count = 0;
	sa_bulk_attr_t	bulk[5];
	uint64_t	mtime[2], ctime[2];
	uint64_t	uid, gid, projid;
	blkptr_t	*bps;
	size_t		maxblocks, nbps;
	uint_t		inblksz;
	uint64_t	clear_setid_bits_txg = 0;
	uint64_t	last_synced_txg = 0;

	*donep = 0;

	inblksz = inzp->z_blksz;

	/*
	 * Cloning between datasets with different special_small_blocks would
	 * bypass storage tier migration that would occur with a regular copy.
	 */
	if (zfs_bclone_strict_properties && inos != outos &&
	    !inzfsvfs->z_issnap && spa_has_special(dmu_objset_spa(inos))) {
		uint64_t in_smallblk = inos->os_zpl_special_smallblock;
		uint64_t out_smallblk = outos->os_zpl_special_smallblock;
		if (in_smallblk != out_smallblk) {
			uint64_t min_smallblk = MIN(in_smallblk, out_smallblk);
			uint64_t max_smallblk = MAX(in_smallblk, out_smallblk);
			if (min_smallblk < inblksz &&
			    (inos->os_compress != ZIO_COMPRESS_OFF ||
			    max_smallblk >= inblksz)) {
				return (SET_ERROR(EXDEV));
			}
		}
	}

	/*
	 * We cannot clone into a file with different block size if we can't
	 * grow it (block size is already bigger, has more than one block, or
	 * not locked for growth).  There are other possible reasons for the
	 * grow to fail, but we cover what we can before opening transaction
	 * and the rest detect after we try to do it.
	 */
	if (inblksz < outzp->z_blksz)
		return (SET_ERROR(EINVAL));
	if (inblksz != outzp->z_blksz && (outzp->z_size > outzp->z_blksz ||
	    outlr->lr_length != UINT64_MAX))
		return (SET_ERROR(EINVAL));

	/*
	 * Block size must be power-of-2 if destination offset != 0.
	 * There can be no multiple blocks of non-power-of-2 size.
	 */
	if (outoff != 0 && !ISP2(inblksz))
		return (SET_ERROR(EINVAL));

	/*
	 * Offsets and len must be at block boundries.
	 */
	if ((inoff % inblksz) != 0 || (outoff % inblksz) != 0)
		return (SET_ERROR(EINVAL));
	/*
	 * Length must be multipe of blksz, except for the end of the file.
	 */
	if ((len % inblksz) != 0 &&
	    (len < inzp->z_size - inoff || len < outzp->z_size - outoff))
		return (SET_ERROR(EINVAL));

	/*
	 * If we are copying only one block and it is smaller than recordsize
	 * property, do not allow destination to grow beyond one block if it
	 * is not there yet.  Otherwise the destination will get stuck with
	 * that block size forever, that can be as small as 512 bytes, no
	 * matter how big the destination grow later.
	 */
	if (len <= inblksz && inblksz < outzfsvfs->z_max_blksz &&
	    outzp->z_size <= inblksz && outoff + len > inblksz)
		return (SET_ERROR(EINVAL));

	error = zn_rlimit_fsize(outoff + len);
	if (error != 0)
		return (error);

	if (inoff >= MAXOFFSET_T || outoff >= MAXOFFSET_T)
		return (SET_ERROR(EFBIG));

	/*
	 * A dedupe leaves the destination's content and metadata alone: it can
	 * only replace blocks with identical ones.  The times must not move,
	 * and size, flags and seq cannot change either - a dedupe never extends
	 * the file, never clears setid bits and never bumps z_seq - so there is
	 * nothing to write back at all.
	 */
	if (!dedup) {
		SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_MTIME(outzfsvfs), NULL,
		    &mtime, 16);
		SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_CTIME(outzfsvfs), NULL,
		    &ctime, 16);
		SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_SIZE(outzfsvfs), NULL,
		    &outzp->z_size, 8);
		SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_FLAGS(outzfsvfs), NULL,
		    &outzp->z_pflags, 8);
		if (outzp->z_is_sa)
			SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_SEQ(outzfsvfs),
			    NULL, &outzp->z_seq, 8);
	}

	maxblocks = zil_max_log_data(zilog, sizeof (lr_clone_range_t)) /
	    sizeof (bps[0]);

	uid = KUID_TO_SUID(ZTOUID(outzp));
	gid = KGID_TO_SGID(ZTOGID(outzp));
	projid = outzp->z_projid;

	bps = vmem_alloc(sizeof (bps[0]) * maxblocks, KM_SLEEP);

	/*
	 * Clone the file in reasonable size chunks.  Each chunk is cloned
	 * in a separate transaction; this keeps the intent log records small
	 * and allows us to do more fine-grained space accounting.
	 */
	while (len > 0) {
		size = MIN(inblksz * maxblocks, len);

		if (zfs_id_overblockquota(outzfsvfs, DMU_USERUSED_OBJECT,
		    uid) ||
		    zfs_id_overblockquota(outzfsvfs, DMU_GROUPUSED_OBJECT,
		    gid) ||
		    (projid != ZFS_DEFAULT_PROJID &&
		    zfs_id_overblockquota(outzfsvfs, DMU_PROJECTUSED_OBJECT,
		    projid))) {
			error = SET_ERROR(EDQUOT);
			break;
		}

		nbps = maxblocks;
		last_synced_txg = spa_last_synced_txg(dmu_objset_spa(inos));
		error = dmu_read_l0_bps(inos, inzp->z_id, inoff, size, bps,
		    &nbps);
		if (error != 0) {
			/*
			 * If we are trying to clone a block that was created
			 * in the current transaction group, the error will be
			 * EAGAIN here.  Based on zfs_bclone_wait_dirty either
			 * return a shortened range to the caller so it can
			 * fallback, or wait for the next TXG and check again.
			 * A dedupe always waits: it has no fallback, so the
			 * EAGAIN would surface to the FIDEDUPERANGE caller,
			 * and the comparison already read this very data.
			 */
			if (error == EAGAIN &&
			    (dedup || zfs_bclone_wait_dirty)) {
				txg_wait_flag_t wait_flags =
				    spa_get_failmode(dmu_objset_spa(inos)) ==
				    ZIO_FAILURE_MODE_CONTINUE ?
				    TXG_WAIT_SUSPEND : 0;
				error = txg_wait_synced_flags(
				    dmu_objset_pool(inos), last_synced_txg + 1,
				    wait_flags);
				if (error == 0)
					continue;
				ASSERT3U(error, ==, ESHUTDOWN);
				error = SET_ERROR(EIO);
			}

			break;
		}

		/*
		 * Start a transaction.
		 */
		tx = dmu_tx_create(outos);
		dmu_tx_hold_sa(tx, outzp->z_sa_hdl, ZFS_SEQ_MAY_GROW(outzp));
		db = (dmu_buf_impl_t *)sa_get_db(outzp->z_sa_hdl);
		DB_DNODE_ENTER(db);
		dmu_tx_hold_clone_by_dnode(tx, DB_DNODE(db), outoff, size,
		    inblksz);
		DB_DNODE_EXIT(db);
		zfs_sa_upgrade_txholds(tx, outzp);
		error = dmu_tx_assign(tx, DMU_TX_WAIT);
		if (error != 0) {
			dmu_tx_abort(tx);
			break;
		}

		/*
		 * Copy source znode's block size. This is done only if the
		 * whole znode is locked (see zfs_rangelock_cb()) and only
		 * on the first iteration since zfs_rangelock_reduce() will
		 * shrink down lr_length to the appropriate size.
		 */
		if (outlr->lr_length == UINT64_MAX) {
			zfs_grow_blocksize(outzp, inblksz, tx);

			/*
			 * Block growth may fail for many reasons we can not
			 * predict here. If it happens, the cloning is doomed.
			 */
			if (inblksz != outzp->z_blksz) {
				error = SET_ERROR(EINVAL);
				dmu_tx_commit(tx);
				break;
			}

			/*
			 * Round range lock up to the block boundary, so we
			 * prevent appends until we are done.
			 */
			zfs_rangelock_reduce(outlr, outoff,
			    ((len - 1) / inblksz + 1) * inblksz);
		}

		error = dmu_brt_clone(outos, outzp->z_id, outoff, size, tx,
		    bps, nbps);
		if (error != 0) {
			dmu_tx_commit(tx);
			break;
		}

		/*
		 * A dedupe replaces the destination's blocks with blocks
		 * holding the same bytes, so a page already cached for this
		 * range stays correct and there is nothing to refresh.  The
		 * only page whose contents can differ from the DMU is one
		 * carrying an mmap store made after the pre-compare flush,
		 * and refreshing that would discard the store, which is a
		 * modification a dedupe must never make.
		 */
		if (!dedup &&
		    zn_has_cached_data(outzp, outoff, outoff + size - 1)) {
			update_pages(outzp, outoff, size, outos);
		}

		if (!dedup) {
			zfs_clear_setid_bits_if_necessary(outzfsvfs, outzp, cr,
			    &clear_setid_bits_txg, tx);

			zfs_tstamp_update_setup(outzp, CONTENT_MODIFIED, mtime,
			    ctime);

			if (outzp->z_is_sa)
				outzp->z_has_seq = B_TRUE;

			/*
			 * Update the file size (zp_size) if it has changed;
			 * account for possible concurrent updates.
			 */
			while ((outsize = outzp->z_size) < outoff + size) {
				(void) atomic_cas_64(&outzp->z_size, outsize,
				    outoff + size);
			}
		} else {
			/* A dedupe can only ever rewrite blocks in place. */
			ASSERT3U(outoff + size, <=, outzp->z_size);
		}

		if (count > 0) {
			ASSERT3S(count, <=, ARRAY_SIZE(bulk));
			error = sa_bulk_update(outzp->z_sa_hdl, bulk, count,
			    tx);
		}

		/*
		 * A dedupe is not logged.  It replaces the destination's blocks
		 * with blocks holding the very same bytes, so if the txg is
		 * lost to a crash and never replayed, the destination simply
		 * keeps its own copy of that identical content: nothing a
		 * reader can observe is lost, only the sharing.  Sharing is all
		 * FIDEDUPERANGE promises anyway, and it is the cheaper of the
		 * alternatives.  Logging a plain TX_CLONE_RANGE would restamp
		 * the destination's mtime/ctime on replay, and telling a dedupe
		 * apart with a record of its own would be an on-disk ZIL format
		 * change: an old kernel meeting an unknown txtype fails
		 * zil_parse(), and zil_replay() then destroys the whole log,
		 * discarding every later record - including synced writes.
		 */
		if (!dedup) {
			zfs_log_clone_range(zilog, tx, TX_CLONE_RANGE, outzp,
			    outoff, size, inblksz, bps, nbps);
		}

		dmu_tx_commit(tx);

		if (error != 0)
			break;

		inoff += size;
		outoff += size;
		len -= size;
		done += size;

		if (issig()) {
			error = SET_ERROR(EINTR);
			break;
		}
	}

	vmem_free(bps, sizeof (bps[0]) * maxblocks);
	zfs_znode_update_vfs(outzp);

	*donep = done;

	return (error);
}

/*
 * Clone part of inzp file into outzp file. This must be done under range locks
 * held on both files so as to prevent it from being modified while the clone
 * takes place.
 *
 * Copies bytes from inzp->inoffp to outzp->outoffp, up to *lenp bytes.  On
 * entry *lenp holds the requested length; on successful return it holds the
 * number of bytes actually cloned, and inoffp/outoffp are advanced by that
 * amount.
 *
 * If we make no progress at all, then the caller made a mistake or asked for
 * something impossible, and EINVAL (or another appropriate error) is returned.
 *
 * Note, it doesn't return how many bytes are left to be copied.
 * On errors which are caused by any file system limitations or
 * BRT limitations EINVAL is returned. In the most cases a user
 * requested bad parameters, it could be possible to clone the file but
 * some parameters don't match the requirements.
 */
int
zfs_clone_range(znode_t *inzp, uint64_t *inoffp, znode_t *outzp,
    uint64_t *outoffp, uint64_t *lenp, cred_t *cr)
{
	zfsvfs_t	*inzfsvfs = ZTOZSB(inzp);
	zfsvfs_t	*outzfsvfs = ZTOZSB(outzp);
	zfs_locked_range_t *inlr, *outlr;
	zilog_t		*zilog;
	uint64_t	inoff = *inoffp;
	uint64_t	outoff = *outoffp;
	uint64_t	len = *lenp;
	uint64_t	done = 0;
	int		error;

	/*
	 * We need to call zfs_enter() potentially on two different datasets,
	 * so we need a dedicated function for that.
	 */
	error = zfs_enter_two(inzfsvfs, outzfsvfs, FTAG);
	if (error != 0)
		return (error);

	/*
	 * Read z_log only after entering, so a suspend/resume (rollback,
	 * zfs receive -F) cannot leave us with a stale or freed zilog.
	 */
	zilog = outzfsvfs->z_log;

	error = zfs_clone_range_precheck(inzp, outzp);
	if (error != 0)
		goto out;

	/*
	 * The range to clone must lie within the source file.  Clamp it to the
	 * source EOF and treat an empty range as a no-op.
	 */
	if (inoff >= inzp->z_size) {
		*lenp = 0;
		goto out;
	}
	if (len > inzp->z_size - inoff)
		len = inzp->z_size - inoff;
	if (len == 0) {
		*lenp = 0;
		goto out;
	}

	/*
	 * Callers might not be able to detect properly that we are read-only,
	 * so check it explicitly here.
	 */
	if (zfs_is_readonly(outzfsvfs)) {
		error = SET_ERROR(EROFS);
		goto out;
	}

	/*
	 * If immutable then return EPERM.  Intentionally allow ZFS_READONLY
	 * through here.  See zfs_zaccess_common().
	 */
	if ((outzp->z_pflags & ZFS_IMMUTABLE) != 0) {
		error = SET_ERROR(EPERM);
		goto out;
	}

	/*
	 * No overlapping if we are cloning within the same file.
	 */
	if (inzp == outzp) {
		if (inoff < outoff + len && outoff < inoff + len) {
			error = SET_ERROR(EINVAL);
			goto out;
		}
	}

	/* Flush any mmap()'d data to disk */
	if (zn_has_cached_data(inzp, inoff, inoff + len - 1))
		zn_flush_cached_data(inzp, B_TRUE);

	/*
	 * Maintain predictable lock order.
	 */
	if (inzp == outzp) {
		/*
		 * Within one file, one writer lock spans both
		 * ranges.  Two locks on the same znode can deadlock
		 * this thread against itself: a write that grows the
		 * file's block size grows its lock to cover the
		 * whole file (see zfs_rlock.c), which then conflicts
		 * with the other.  The source block pointers are
		 * read before the lock is reduced, so the source
		 * stays covered.
		 */
		uint64_t lo = MIN(inoff, outoff);
		uint64_t hi = MAX(inoff + len, outoff + len);
		outlr = zfs_rangelock_enter(&outzp->z_rangelock, lo,
		    hi - lo, RL_WRITER);
		inlr = NULL;
	} else if (inzp < outzp) {
		inlr = zfs_rangelock_enter(&inzp->z_rangelock, inoff, len,
		    RL_READER);
		outlr = zfs_rangelock_enter(&outzp->z_rangelock, outoff, len,
		    RL_WRITER);
	} else {
		outlr = zfs_rangelock_enter(&outzp->z_rangelock, outoff, len,
		    RL_WRITER);
		inlr = zfs_rangelock_enter(&inzp->z_rangelock, inoff, len,
		    RL_READER);
	}

	error = zfs_clone_range_locked(inzp, inoff, outzp, outoff, len, cr,
	    outlr, B_FALSE, &done);

	zfs_rangelock_exit(outlr);
	if (inlr != NULL)
		zfs_rangelock_exit(inlr);

	if (done > 0) {
		/*
		 * If we have made at least partial progress, reset the error.
		 */
		error = 0;

		ZFS_ACCESSTIME_STAMP(inzfsvfs, inzp);

		if (outzfsvfs->z_os->os_sync == ZFS_SYNC_ALWAYS)
			error = zil_commit(zilog, outzp->z_id);

		*inoffp += done;
		*outoffp += done;
		*lenp = done;
	} else {
		/*
		 * If we made no progress, there must be a good reason.
		 * EOF is handled explicitly above, before the loop.
		 */
		ASSERT3S(error, !=, 0);
	}

out:
	zfs_exit_two(inzfsvfs, outzfsvfs, FTAG);

	return (error);
}

/*
 * Compare two ranges by copying them out of the DMU into buffers and comparing
 * those.  Only used where the dbuf comparison below cannot go; see there.
 */
static int
zfs_dedupe_range_copy_memcmp(znode_t *inzp, uint64_t inoff, znode_t *outzp,
    uint64_t outoff, uint64_t len, boolean_t *eqp)
{
	objset_t	*inos = ZTOZSB(inzp)->z_os;
	objset_t	*outos = ZTOZSB(outzp)->z_os;
	uint64_t	chunk = MIN(len, (uint64_t)zfs_vnops_read_chunk_size);
	char		*inbuf, *outbuf;
	int		error = 0;

	*eqp = B_FALSE;

	inbuf = vmem_alloc(chunk, KM_SLEEP);
	outbuf = vmem_alloc(chunk, KM_SLEEP);

	while (len > 0) {
		uint64_t n = MIN(len, chunk);

		if (issig()) {
			error = SET_ERROR(EINTR);
			break;
		}

		error = dmu_read(inos, inzp->z_id, inoff, n, inbuf,
		    DMU_READ_PREFETCH);
		if (error != 0)
			break;
		error = dmu_read(outos, outzp->z_id, outoff, n, outbuf,
		    DMU_READ_PREFETCH);
		if (error != 0)
			break;

		if (memcmp(inbuf, outbuf, n) != 0)
			break;

		inoff += n;
		outoff += n;
		len -= n;
	}

	if (error == 0 && len == 0)
		*eqp = B_TRUE;

	vmem_free(inbuf, chunk);
	vmem_free(outbuf, chunk);

	return (error);
}

/*
 * Compare [inoff, inoff + len) in inzp with [outoff, outoff + len) in outzp by
 * reading the committed on-disk data through the DMU (the same data
 * zfs_clone_range_locked() will clone) and byte-comparing it.  Both files must
 * be range locked by the caller.  On return *eqp is set if the ranges are
 * byte-for-byte equal.  This is the authoritative (and always-correct) path;
 * the DMU serves holes as zeros and decrypts transparently, so it needs no
 * special-casing for encryption, holes or embedded blocks.
 *
 * The comparison is done on the dbufs themselves.  dmu_read() would only hold
 * the very same buffers and memcpy() out of them, so going through it would
 * cost two allocations and a copy of every byte, for nothing: we are not
 * keeping the data, only looking at it.  Our range locks keep it still while
 * we do.
 */
static int
zfs_dedupe_range_memcmp(znode_t *inzp, uint64_t inoff, znode_t *outzp,
    uint64_t outoff, uint64_t len, boolean_t *eqp)
{
	objset_t	*inos = ZTOZSB(inzp)->z_os;
	objset_t	*outos = ZTOZSB(outzp)->z_os;
	dnode_t		*indn, *outdn;
	uint64_t	blksz, chunk;
	boolean_t	diff = B_FALSE;
	int		error;

	*eqp = B_FALSE;

	error = dnode_hold(inos, inzp->z_id, FTAG, &indn);
	if (error != 0)
		return (error);
	error = dnode_hold(outos, outzp->z_id, FTAG, &outdn);
	if (error != 0) {
		dnode_rele(indn, FTAG);
		return (error);
	}

	/*
	 * dmu_buf_hold_array_by_dnode() refuses to look past a single block
	 * whose size is not a power of two, calling zfs_panic_recover() rather
	 * than serving the tail as zeros the way dmu_read() does.  A ZPL file
	 * should never present that shape - zfs_grow_blocksize() only leaves a
	 * block size alone once the file has outgrown it, so an odd-sized block
	 * means the file still fits inside it and the range cannot reach past
	 * it.  Should one turn up regardless, fall back to copying rather than
	 * add another way to take the pool down.  Our caller reaches
	 * dmu_read_l0_bps() first, which holds the same buffers and would trip
	 * over such a file before we ever got here, so in practice this only
	 * catches what gets past that on a pool running with zfs_recover set.
	 */
	if ((indn->dn_datablkshift == 0 && inoff + len > indn->dn_datablksz) ||
	    (outdn->dn_datablkshift == 0 &&
	    outoff + len > outdn->dn_datablksz)) {
		dnode_rele(outdn, FTAG);
		dnode_rele(indn, FTAG);
		return (zfs_dedupe_range_copy_memcmp(inzp, inoff, outzp,
		    outoff, len, eqp));
	}

	/*
	 * A block is read and decompressed in full whatever part of it we ask
	 * for, so a chunk below the block size would buy nothing and only hold
	 * the same buffer once per piece of it.  Take whole blocks, at least
	 * one, capped the way zfs_read() caps its own chunk so that a large
	 * request does not pin an unbounded stretch of the ARC.
	 *
	 * zfs_dedupe_range() has already refused an unaligned range, so each
	 * buffer we are handed begins at the offset we are comparing.
	 */
	blksz = indn->dn_datablksz;
	ASSERT3U(outdn->dn_datablksz, ==, blksz);
	chunk = MAX(1, MIN(zfs_vnops_read_chunk_size, DMU_MAX_ACCESS / 2) /
	    blksz) * blksz;

	while (len > 0 && !diff) {
		uint64_t n = MIN(len, chunk);
		uint64_t left = n;
		dmu_buf_t **indbp, **outdbp;
		int innum, outnum;

		if (issig()) {
			error = SET_ERROR(EINTR);
			break;
		}

		error = dmu_buf_hold_array_by_dnode(indn, inoff, n, TRUE,
		    FTAG, &innum, &indbp, DMU_READ_PREFETCH);
		if (error != 0)
			break;
		error = dmu_buf_hold_array_by_dnode(outdn, outoff, n, TRUE,
		    FTAG, &outnum, &outdbp, DMU_READ_PREFETCH);
		if (error != 0) {
			dmu_buf_rele_array(indbp, innum, FTAG);
			break;
		}

		/*
		 * The two files share a block size and both offsets are block
		 * aligned, so the two arrays cover the span the same way.
		 * Should that assumption ever break, refuse rather than walk
		 * the shorter array out of bounds.
		 */
		ASSERT3S(innum, ==, outnum);
		if (innum != outnum) {
			dmu_buf_rele_array(outdbp, outnum, FTAG);
			dmu_buf_rele_array(indbp, innum, FTAG);
			error = SET_ERROR(EINVAL);
			break;
		}

		for (int i = 0; i < innum; i++) {
			uint64_t tocmp;

			ASSERT3U(indbp[i]->db_offset, ==, inoff);
			ASSERT3U(outdbp[i]->db_offset, ==, outoff);
			ASSERT3P(indbp[i]->db_data, !=, NULL);
			ASSERT3P(outdbp[i]->db_data, !=, NULL);

			tocmp = MIN(indbp[i]->db_size, left);

			if (memcmp(indbp[i]->db_data, outdbp[i]->db_data,
			    tocmp) != 0) {
				diff = B_TRUE;
				break;
			}

			inoff += tocmp;
			outoff += tocmp;
			left -= tocmp;
		}

		dmu_buf_rele_array(outdbp, outnum, FTAG);
		dmu_buf_rele_array(indbp, innum, FTAG);

		len -= (n - left);
	}

	if (error == 0 && len == 0 && !diff)
		*eqp = B_TRUE;

	dnode_rele(outdn, FTAG);
	dnode_rele(indn, FTAG);

	return (error);
}

/*
 * Decide, from the two L0 block pointers alone, whether they already name the
 * same data: the same physical block, or two holes.  This is what a
 * previously cloned or deduped pair of ranges looks like, and it is the one
 * case where the clone step itself would have nothing left to do.
 */
static boolean_t
zfs_dedupe_bp_same_block(const blkptr_t *bi, const blkptr_t *bo)
{
	/*
	 * Two holes are both a run of zeros over the same block, so they are
	 * equal without either one naming any storage.  Their sizes are not
	 * worth comparing and must not be: our caller pairs the two files
	 * block for block over a block size they share, but a block that was
	 * never written at all reaches us as an all-zero block pointer (see
	 * dmu_read_l0_bps()), whose BP_GET_LSIZE() decodes to the minimum
	 * block size rather than to the file's.  Reading a hole to compare
	 * its zeros against another hole's zeros is exactly what this avoids.
	 */
	if (BP_IS_HOLE(bi) && BP_IS_HOLE(bo))
		return (B_TRUE);

	if (BP_IS_HOLE(bi) || BP_IS_HOLE(bo) ||
	    BP_IS_EMBEDDED(bi) || BP_IS_EMBEDDED(bo) ||
	    BP_IS_REDACTED(bi) || BP_IS_REDACTED(bo))
		return (B_FALSE);

	/*
	 * One allocation is one DVA born in one txg, so a matching first DVA
	 * at a matching physical birth means both pointers name the same
	 * physical block - typically because the ranges were already cloned
	 * or deduped.  This needs no checksum and holds even where the
	 * checksum test below cannot (encryption, weak checksums): the same
	 * block under the same key decrypts to the same data.  BP_EQUAL() is
	 * too strict for this: it also compares the logical births, and
	 * cloning restamps those on every clone.
	 */
	return (BP_GET_PHYSICAL_BIRTH(bi) == BP_GET_PHYSICAL_BIRTH(bo) &&
	    DVA_EQUAL(&bi->blk_dva[0], &bo->blk_dva[0]));
}

/*
 * Decide, from the two L0 block pointers alone, whether the blocks provably
 * hold identical logical data.  Modeled on the nopwrite guard set in
 * zio_nop_write(): a match of a cryptographically strong (dedup-grade)
 * checksum is trusted to imply data equality, exactly as ZFS dedup relies on
 * it.  This can only ever prove equality; a mismatch (or any non-provable
 * case) is inconclusive and the caller must fall back to reading the data.
 * Embedded blocks are the exception to needing a checksum: their payload
 * lives in the pointer itself, so two of them are compared directly.
 *
 * All of the following must hold for a checksum match to be trusted:
 *   - neither BP is embedded, a hole, or redacted (no usable blk_cksum);
 *   - neither BP is encrypted (per-block IV/salt make the checksum data-
 *     independent, so equal plaintext yields different checksums);
 *   - both BPs use the same checksum function and it carries the DEDUP flag
 *     (sha256/sha512/skein/blake3; fletcher and edonr are excluded);
 *   - compression, PSIZE, LSIZE and byteorder match, since blk_cksum is
 *     computed over the physical (post-compression) bytes.
 * Salted checksums (skein/blake3) are safe here because block cloning is
 * always intra-pool (zfs_clone_range_precheck() returns EXDEV otherwise) and
 * the salt is per-pool, so identical data yields identical checksums.
 */
static boolean_t
zfs_dedupe_bp_provably_equal(const blkptr_t *bi, const blkptr_t *bo)
{
	enum zio_checksum ck;

	if (zfs_dedupe_bp_same_block(bi, bo))
		return (B_TRUE);

	/*
	 * Two embedded blocks carry their (possibly compressed) payload in
	 * the pointers themselves.  The same payload bytes under the same
	 * compression function decompress to the same data, so equality is
	 * provable without decompressing anything.  A payload mismatch
	 * proves nothing - equal data can compress to different bytes - and
	 * falls through to the read path like any other unprovable case.
	 */
	if (BP_IS_EMBEDDED(bi) && BP_IS_EMBEDDED(bo)) {
		uint8_t pi[BPE_PAYLOAD_SIZE], po[BPE_PAYLOAD_SIZE];

		if (BPE_GET_ETYPE(bi) != BP_EMBEDDED_TYPE_DATA ||
		    BPE_GET_ETYPE(bo) != BP_EMBEDDED_TYPE_DATA ||
		    BP_GET_COMPRESS(bi) != BP_GET_COMPRESS(bo) ||
		    BPE_GET_LSIZE(bi) != BPE_GET_LSIZE(bo) ||
		    BPE_GET_PSIZE(bi) != BPE_GET_PSIZE(bo) ||
		    BP_GET_BYTEORDER(bi) != BP_GET_BYTEORDER(bo))
			return (B_FALSE);

		/*
		 * The psize bit field can encode more than the payload area
		 * holds.  No valid pointer does, but this is cheaper than
		 * trusting that.
		 */
		if (BPE_GET_PSIZE(bi) > BPE_PAYLOAD_SIZE)
			return (B_FALSE);

		decode_embedded_bp_compressed(bi, pi);
		decode_embedded_bp_compressed(bo, po);

		return (memcmp(pi, po, BPE_GET_PSIZE(bi)) == 0);
	}

	if (BP_IS_EMBEDDED(bi) || BP_IS_EMBEDDED(bo) ||
	    BP_IS_HOLE(bi) || BP_IS_HOLE(bo) ||
	    BP_IS_REDACTED(bi) || BP_IS_REDACTED(bo) ||
	    BP_IS_ENCRYPTED(bi) || BP_IS_ENCRYPTED(bo))
		return (B_FALSE);

	ck = BP_GET_CHECKSUM(bi);
	if (ck != BP_GET_CHECKSUM(bo))
		return (B_FALSE);
	if ((zio_checksum_table[ck].ci_flags & ZCHECKSUM_FLAG_DEDUP) == 0)
		return (B_FALSE);

	if (BP_GET_COMPRESS(bi) != BP_GET_COMPRESS(bo) ||
	    BP_GET_PSIZE(bi) != BP_GET_PSIZE(bo) ||
	    BP_GET_LSIZE(bi) != BP_GET_LSIZE(bo) ||
	    BP_GET_BYTEORDER(bi) != BP_GET_BYTEORDER(bo))
		return (B_FALSE);

	return (ZIO_CHECKSUM_EQUAL(bi->blk_cksum, bo->blk_cksum));
}

/*
 * The mirror of zfs_dedupe_bp_provably_equal(): decide, from the two L0 block
 * pointers alone, whether the blocks provably hold *different* logical data,
 * so the caller can report the ranges as differing without reading them.  This
 * can only ever prove inequality; anything not provable is inconclusive and
 * the caller must fall back to reading the data.
 *
 * A checksum covers the physical bytes, so unequal checksums prove only that
 * the physical bytes differ.  Carrying that back to the logical data needs the
 * block stored verbatim, hence the insistence on compression being off: two
 * blocks holding identical data can perfectly well compress to different
 * bytes, say after the compression property was changed, and zstd records only
 * its algorithm in the BP, not its level.  Encrypted blocks are excluded for
 * the same reason as in the equality test, to the opposite effect: per-block
 * IVs give identical plaintext different ciphertext, so unequal checksums
 * there would say nothing.  Holes and embedded blocks carry no comparable
 * checksum at all.
 *
 * Unlike the equality test this does not need a dedup-grade checksum.  Even a
 * weak one is deterministic, so equal data always gives equal checksums, and
 * unequal checksums therefore always mean unequal data.
 */
static boolean_t
zfs_dedupe_bp_provably_unequal(const blkptr_t *bi, const blkptr_t *bo)
{
	if (BP_IS_EMBEDDED(bi) || BP_IS_EMBEDDED(bo) ||
	    BP_IS_HOLE(bi) || BP_IS_HOLE(bo) ||
	    BP_IS_REDACTED(bi) || BP_IS_REDACTED(bo) ||
	    BP_IS_ENCRYPTED(bi) || BP_IS_ENCRYPTED(bo))
		return (B_FALSE);

	if (BP_GET_COMPRESS(bi) != ZIO_COMPRESS_OFF ||
	    BP_GET_COMPRESS(bo) != ZIO_COMPRESS_OFF)
		return (B_FALSE);

	if (BP_GET_CHECKSUM(bi) != BP_GET_CHECKSUM(bo) ||
	    BP_GET_LSIZE(bi) != BP_GET_LSIZE(bo) ||
	    BP_GET_BYTEORDER(bi) != BP_GET_BYTEORDER(bo))
		return (B_FALSE);

	return (!ZIO_CHECKSUM_EQUAL(bi->blk_cksum, bo->blk_cksum));
}

/*
 * Compare [inoff, inoff + len) in inzp with [outoff, outoff + len) in outzp,
 * setting *samep if the ranges are byte-for-byte equal.  Both files must be
 * range locked by the caller.
 *
 * As a fast path, blocks are compared by their block-pointer checksums, which
 * can settle a block either way without reading it: equal, when a strong
 * checksum matches (zfs_dedupe_bp_provably_equal(), mirroring how ZFS dedup
 * trusts checksums), or unequal, when checksums over verbatim-stored blocks
 * differ (zfs_dedupe_bp_provably_unequal()).  Whatever neither can settle - a
 * weak or mismatched checksum, encryption, a hole, compression, a dirty (not
 * yet synced) block, etc. - falls back to reading and byte-comparing the data.
 * Both tests only ever assert what the block pointers prove, so the result is
 * identical to a full byte compare.
 *
 * A trailing block the range only partly covers is no different: the two files
 * share a block size and both offsets are block aligned, so the tail occupies
 * the same intra-block span in each, and equal whole-block checksums prove the
 * whole blocks equal, hence any common prefix of them equal.  Only the bytes
 * inside the range are ever compared.
 *
 * When the ranges match, *sharedp additionally reports whether every block
 * pair already names the same storage (or is a pair of holes), in which case
 * the clone step would only reinstall pointers that are already in place and
 * the caller can skip it.
 */
static int
zfs_dedupe_range_compare(znode_t *inzp, uint64_t inoff, znode_t *outzp,
    uint64_t outoff, uint64_t len, boolean_t *samep, boolean_t *sharedp)
{
	objset_t	*inos = ZTOZSB(inzp)->z_os;
	objset_t	*outos = ZTOZSB(outzp)->z_os;
	uint_t		blksz = inzp->z_blksz;
	blkptr_t	*bps_in = NULL, *bps_out = NULL;
	size_t		maxblocks;
	uint64_t	off = 0;
	boolean_t	eq;
	boolean_t	shared = B_TRUE;
	int		error = 0;

	*samep = B_FALSE;
	*sharedp = B_FALSE;

	/*
	 * zfs_dedupe_range() has already rejected differing block sizes and
	 * unaligned offsets, so the two files' blocks line up one to one.
	 */
	ASSERT3U(outzp->z_blksz, ==, blksz);
	ASSERT0(inoff % blksz);
	ASSERT0(outoff % blksz);

	/*
	 * Bound the block-pointer batch so the temporary arrays stay small and
	 * each dmu_read_l0_bps() stays within DMU_MAX_ACCESS.
	 */
	maxblocks = MAX(1, MIN((uint64_t)16, (DMU_MAX_ACCESS >> 1) / blksz));
	bps_in = kmem_alloc(sizeof (blkptr_t) * maxblocks, KM_SLEEP);
	bps_out = kmem_alloc(sizeof (blkptr_t) * maxblocks, KM_SLEEP);

	while (off < len) {
		uint64_t span = MIN(len - off, (uint64_t)maxblocks * blksz);
		size_t nin = maxblocks, nout = maxblocks;
		int e1, e2;

		if (issig()) {
			error = SET_ERROR(EINTR);
			goto out;
		}

		e1 = dmu_read_l0_bps(inos, inzp->z_id, inoff + off, span,
		    bps_in, &nin);
		e2 = dmu_read_l0_bps(outos, outzp->z_id, outoff + off, span,
		    bps_out, &nout);

		if (e1 != 0 || e2 != 0 || nin != nout) {
			/*
			 * A block created in the current txg (EAGAIN), or any
			 * other reason we could not get stable BPs, just means
			 * we compare this span by reading it; the data path is
			 * always correct, including for dirty blocks.
			 */
			shared = B_FALSE;
			error = zfs_dedupe_range_memcmp(inzp, inoff + off,
			    outzp, outoff + off, span, &eq);
			if (error != 0 || !eq)
				goto out;
			off += span;
			continue;
		}

		for (size_t i = 0; i < nin; i++) {
			uint64_t cmpoff = off + (uint64_t)i * blksz;
			uint64_t cmplen = MIN((uint64_t)blksz, len - cmpoff);

			if (zfs_dedupe_bp_same_block(&bps_in[i], &bps_out[i]))
				continue;
			shared = B_FALSE;

			if (zfs_dedupe_bp_provably_equal(&bps_in[i],
			    &bps_out[i]))
				continue;

			/*
			 * Blocks that provably differ make the whole range
			 * differ - but only when the range covers all of this
			 * one.  Two differing blocks can still agree over the
			 * part a trailing sub-block request covers.
			 */
			if (cmplen == blksz &&
			    zfs_dedupe_bp_provably_unequal(&bps_in[i],
			    &bps_out[i]))
				goto out;

			/*
			 * Not provable from BPs: compare just this block, or
			 * only the part of it the range covers.
			 */
			error = zfs_dedupe_range_memcmp(inzp, inoff + cmpoff,
			    outzp, outoff + cmpoff, cmplen, &eq);
			if (error != 0 || !eq)
				goto out;
		}

		off += span;
	}

	*samep = B_TRUE;
	*sharedp = shared;

out:
	kmem_free(bps_in, sizeof (blkptr_t) * maxblocks);
	kmem_free(bps_out, sizeof (blkptr_t) * maxblocks);

	return (error);
}

/*
 * Deduplicate a range of outzp against inzp: if [inoff, inoff + *lenp) and
 * [outoff, outoff + *lenp) hold identical data, clone the source blocks over
 * the destination so the two ranges share storage.  Implements the Linux
 * FIDEDUPERANGE ioctl on top of block cloning.
 *
 * The comparison and the clone are performed while holding a RL_READER range
 * lock on inzp and a RL_WRITER range lock on outzp, so the blocks that get
 * cloned are exactly the bytes that were compared.
 *
 * On return:
 *   error != 0            - the request failed;
 *   *samep == B_FALSE     - the ranges differ, nothing was deduped;
 *   *samep == B_TRUE      - the ranges matched and *lenp holds the number of
 *                           bytes deduped (0 if a shortened range aligned away
 *                           to nothing).
 *
 * Both ranges must lie within their files.  A trailing partial block is
 * aligned away when can_shorten is set, and rejected with EINVAL when it is
 * not, so a dedupe can still cover less than the caller asked for.
 */
int
zfs_dedupe_range(znode_t *inzp, uint64_t inoff, znode_t *outzp, uint64_t outoff,
    uint64_t *lenp, cred_t *cr, boolean_t can_shorten, boolean_t *samep)
{
	zfsvfs_t	*inzfsvfs = ZTOZSB(inzp);
	zfsvfs_t	*outzfsvfs = ZTOZSB(outzp);
	zfs_locked_range_t *inlr = NULL, *outlr = NULL;
	uint64_t	len = *lenp;
	uint64_t	done = 0;
	uint_t		inblksz;
	boolean_t	shared = B_FALSE;
	int		error;

	*samep = B_FALSE;

	error = zfs_enter_two(inzfsvfs, outzfsvfs, FTAG);
	if (error != 0)
		return (error);

	error = zfs_clone_range_precheck(inzp, outzp);
	if (error != 0)
		goto out;

	/*
	 * Deduplicating nothing trivially succeeds.
	 */
	if (len == 0) {
		*lenp = 0;
		*samep = B_TRUE;
		goto out;
	}

	/*
	 * Unlike a clone, a dedupe request is not silently clamped to the
	 * source EOF: the whole compared range must lie within the source.
	 */
	if (inoff >= inzp->z_size || len > inzp->z_size - inoff) {
		error = SET_ERROR(EINVAL);
		goto out;
	}

	/*
	 * The destination range must lie within the destination too.  Trimming
	 * it to the destination EOF would be invisible to the caller, because
	 * the ioctl reports the length it was asked for rather than the one a
	 * filesystem returns, so a shortened dedupe would be indistinguishable
	 * from a complete one.  generic_remap_checks() rejects this for the
	 * filesystems that use it; do the same here.
	 */
	if (outoff >= outzp->z_size || len > outzp->z_size - outoff) {
		error = SET_ERROR(EINVAL);
		goto out;
	}

	/*
	 * Callers might not be able to detect properly that we are read-only,
	 * so check it explicitly here.
	 */
	if (zfs_is_readonly(outzfsvfs)) {
		error = SET_ERROR(EROFS);
		goto out;
	}

	/*
	 * A dedupe leaves the content as it was, but it still frees the
	 * destination's blocks and replaces them with shared ones, so honor
	 * immutable here the way every other data-writing path does.
	 */
	if ((outzp->z_pflags & ZFS_IMMUTABLE) != 0) {
		error = SET_ERROR(EPERM);
		goto out;
	}

	/*
	 * No overlapping if we are deduping within the same file.
	 */
	if (inzp == outzp) {
		if (inoff < outoff + len && outoff < inoff + len) {
			error = SET_ERROR(EINVAL);
			goto out;
		}
	}

	/* Flush any mmap()'d data on both files before we compare it. */
	if (zn_has_cached_data(inzp, inoff, inoff + len - 1))
		zn_flush_cached_data(inzp, B_TRUE);
	if (zn_has_cached_data(outzp, outoff, outoff + len - 1))
		zn_flush_cached_data(outzp, B_TRUE);

	/*
	 * Maintain predictable lock order.
	 */
	if (inzp < outzp || (inzp == outzp && inoff < outoff)) {
		inlr = zfs_rangelock_enter(&inzp->z_rangelock, inoff, len,
		    RL_READER);
		outlr = zfs_rangelock_enter(&outzp->z_rangelock, outoff, len,
		    RL_WRITER);
	} else {
		outlr = zfs_rangelock_enter(&outzp->z_rangelock, outoff, len,
		    RL_WRITER);
		inlr = zfs_rangelock_enter(&inzp->z_rangelock, inoff, len,
		    RL_READER);
	}

	/*
	 * The clone step can only replace whole blocks of the destination with
	 * whole blocks of the source, so the two files must agree on the block
	 * size and both offsets must be block-aligned; those can never be
	 * shortened away.  A trailing partial block is handled after the
	 * comparison below.  Block sizes cannot change under us here: growing
	 * one takes the whole-file rangelock, which our holds exclude.
	 */
	inblksz = inzp->z_blksz;
	if (inblksz != outzp->z_blksz) {
		error = SET_ERROR(EINVAL);
		goto unlock;
	}
	if ((inoff % inblksz) != 0 || (outoff % inblksz) != 0) {
		error = SET_ERROR(EINVAL);
		goto unlock;
	}
	if ((len % inblksz) != 0 &&
	    (len < inzp->z_size - inoff || len < outzp->z_size - outoff) &&
	    !can_shorten) {
		error = SET_ERROR(EINVAL);
		goto unlock;
	}

	/*
	 * Compare the whole requested range under the locks, before aligning
	 * it down for the clone.  Comparing the full range is what makes a
	 * differing trailing sub-block report DIFFERS rather than being
	 * silently dropped by the alignment below (which, with a large
	 * recordsize, can round the whole request away to nothing).
	 */
	error = zfs_dedupe_range_compare(inzp, inoff, outzp, outoff, len,
	    samep, &shared);
	if (error != 0 || !*samep)
		goto unlock;

	/*
	 * The ranges are identical.  Align a trailing partial block away for
	 * the clone step; the ranges still compared equal, so *samep stays
	 * set even if this leaves nothing to share.
	 */
	if ((len % inblksz) != 0 &&
	    (len < inzp->z_size - inoff || len < outzp->z_size - outoff))
		len = P2ALIGN_TYPED(len, inblksz, uint64_t);
	if (len == 0)
		goto unlock;

	/*
	 * When every destination block already shares its source block (or
	 * both are holes), the ranges were cloned or deduped before and the
	 * clone would only reinstall the pointers the destination already
	 * has - dirtying every block, restamping its logical birth (which
	 * makes the next incremental send carry it again) and churning the
	 * BRT for an end state that is already on disk.  Dedupe tools cannot
	 * see ZFS-level sharing, so they will keep finding the same identical
	 * files; make rededuping them free.
	 */
	if (shared) {
		done = len;
		goto unlock;
	}

	error = zfs_clone_range_locked(inzp, inoff, outzp, outoff, len, cr,
	    outlr, B_TRUE, &done);

unlock:
	zfs_rangelock_exit(outlr);
	zfs_rangelock_exit(inlr);

	if (error == 0) {
		if (done > 0) {
			/*
			 * No zil_commit() here: a dedupe logs nothing, so
			 * there is no itx to wait for even with sync=always.
			 */
			ZFS_ACCESSTIME_STAMP(inzfsvfs, inzp);
		}
		*lenp = done;
	}

out:
	zfs_exit_two(inzfsvfs, outzfsvfs, FTAG);

	return (error);
}

/*
 * Usual pattern would be to call zfs_clone_range() from zfs_replay_clone(),
 * but we cannot do that, because when replaying we don't have source znode
 * available. This is why we need a dedicated replay function.
 */
int
zfs_clone_range_replay(znode_t *zp, uint64_t off, uint64_t len, uint64_t blksz,
    const blkptr_t *bps, size_t nbps)
{
	zfsvfs_t	*zfsvfs;
	dmu_buf_impl_t	*db;
	dmu_tx_t	*tx;
	int		error;
	int		count = 0;
	sa_bulk_attr_t	bulk[5];
	uint64_t	mtime[2], ctime[2];

	ASSERT3U(off, <, MAXOFFSET_T);
	ASSERT3U(len, >, 0);
	ASSERT3U(nbps, >, 0);

	zfsvfs = ZTOZSB(zp);

	ASSERT(spa_feature_is_enabled(dmu_objset_spa(zfsvfs->z_os),
	    SPA_FEATURE_BLOCK_CLONING));

	if ((error = zfs_enter_verify_zp(zfsvfs, zp, FTAG)) != 0)
		return (error);

	ASSERT(zfsvfs->z_replay);
	ASSERT(!zfs_is_readonly(zfsvfs));

	if ((off % blksz) != 0) {
		zfs_exit(zfsvfs, FTAG);
		return (SET_ERROR(EINVAL));
	}

	/*
	 * Start a transaction.
	 */
	tx = dmu_tx_create(zfsvfs->z_os);

	dmu_tx_hold_sa(tx, zp->z_sa_hdl, ZFS_SEQ_MAY_GROW(zp));
	db = (dmu_buf_impl_t *)sa_get_db(zp->z_sa_hdl);
	DB_DNODE_ENTER(db);
	dmu_tx_hold_clone_by_dnode(tx, DB_DNODE(db), off, len, blksz);
	DB_DNODE_EXIT(db);
	zfs_sa_upgrade_txholds(tx, zp);
	error = dmu_tx_assign(tx, DMU_TX_WAIT);
	if (error != 0) {
		dmu_tx_abort(tx);
		zfs_exit(zfsvfs, FTAG);
		return (error);
	}

	if (zp->z_blksz < blksz)
		zfs_grow_blocksize(zp, blksz, tx);

	if (blksz != zp->z_blksz) {
		error = SET_ERROR(EINVAL);
		dmu_tx_commit(tx);
		zfs_exit(zfsvfs, FTAG);
		return (error);
	}

	error = dmu_brt_clone(zfsvfs->z_os, zp->z_id, off, len, tx, bps, nbps);
	if (error != 0) {
		dmu_tx_commit(tx);
		zfs_exit(zfsvfs, FTAG);
		return (error);
	}

	SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_MTIME(zfsvfs), NULL, &mtime, 16);
	SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_CTIME(zfsvfs), NULL, &ctime, 16);
	SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_SIZE(zfsvfs), NULL,
	    &zp->z_size, 8);
	SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_FLAGS(zfsvfs), NULL,
	    &zp->z_pflags, 8);
	zfs_tstamp_update_setup(zp, CONTENT_MODIFIED, mtime, ctime);
	ZFS_PERSIST_SEQ(zp, bulk, count);

	if (zp->z_size < off + len)
		zp->z_size = off + len;

	ASSERT3S(count, <=, ARRAY_SIZE(bulk));
	error = sa_bulk_update(zp->z_sa_hdl, bulk, count, tx);

	/*
	 * zil_replaying() not only check if we are replaying ZIL, but also
	 * updates the ZIL header to record replay progress.
	 */
	VERIFY(zil_replaying(zfsvfs->z_log, tx));

	dmu_tx_commit(tx);

	zfs_znode_update_vfs(zp);

	zfs_exit(zfsvfs, FTAG);

	return (error);
}

EXPORT_SYMBOL(zfs_access);
EXPORT_SYMBOL(zfs_fsync);
EXPORT_SYMBOL(zfs_holey);
EXPORT_SYMBOL(zfs_read);
EXPORT_SYMBOL(zfs_write);
EXPORT_SYMBOL(zfs_getsecattr);
EXPORT_SYMBOL(zfs_setsecattr);
EXPORT_SYMBOL(zfs_clone_range);
EXPORT_SYMBOL(zfs_clone_range_replay);
EXPORT_SYMBOL(zfs_dedupe_range);

ZFS_MODULE_PARAM(zfs_vnops, zfs_vnops_, read_chunk_size, U64, ZMOD_RW,
	"Bytes to read per chunk");

ZFS_MODULE_PARAM(zfs, zfs_, bclone_enabled, INT, ZMOD_RW,
	"Enable block cloning");

ZFS_MODULE_PARAM(zfs, zfs_, bclone_strict_properties, INT, ZMOD_RW,
	"Restrict cross-dataset cloning with different properties");

ZFS_MODULE_PARAM(zfs, zfs_, bclone_wait_dirty, INT, ZMOD_RW,
	"Wait for dirty blocks when cloning");

ZFS_MODULE_PARAM(zfs, zfs_, dio_enabled, INT, ZMOD_RW,
	"Enable Direct I/O");

ZFS_MODULE_PARAM(zfs, zfs_, dio_strict, INT, ZMOD_RW,
	"Return errors on misaligned Direct I/O");