--- old/usr/src/uts/common/fs/zfs/zfs_vnops.c	Mon Aug  5 17:27:12 2019
+++ new/usr/src/uts/common/fs/zfs/zfs_vnops.c	Mon Aug  5 17:27:12 2019
@@ -21,6 +21,8 @@
 
 /*
  * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
+ * Portions Copyright 2007 Jeremy Teo
+ * Portions Copyright 2010 Robert Milkowski
  * Copyright (c) 2012, 2017 by Delphix. All rights reserved.
  * Copyright (c) 2014 Integros [integros.com]
  * Copyright 2015 Joyent, Inc.
@@ -27,9 +29,6 @@
  * Copyright 2017 Nexenta Systems, Inc.
  */
 
-/* Portions Copyright 2007 Jeremy Teo */
-/* Portions Copyright 2010 Robert Milkowski */
-
 #include <sys/types.h>
 #include <sys/param.h>
 #include <sys/time.h>
@@ -83,6 +82,7 @@
 #include <sys/kidmap.h>
 #include <sys/cred.h>
 #include <sys/attr.h>
+#include <sys/dsl_prop.h>
 #include <sys/zil.h>
 
 /*
@@ -135,7 +135,7 @@
  *
  *	If dmu_tx_assign() returns ERESTART and zfsvfs->z_assign is TXG_NOWAIT,
  *	then drop all locks, call dmu_tx_wait(), and try again.  On subsequent
- *	calls to dmu_tx_assign(), pass TXG_NOTHROTTLE in addition to TXG_NOWAIT,
+ *	calls to dmu_tx_assign(), pass TXG_WAITED rather than TXG_NOWAIT,
  *	to indicate that this operation has already called dmu_tx_wait().
  *	This will ensure that we don't retry forever, waiting a short bit
  *	each time.
@@ -160,7 +160,7 @@
  *	rw_enter(...);			// grab any other locks you need
  *	tx = dmu_tx_create(...);	// get DMU tx
  *	dmu_tx_hold_*();		// hold each object you might modify
- *	error = dmu_tx_assign(tx, (waited ? TXG_NOTHROTTLE : 0) | TXG_NOWAIT);
+ *	error = dmu_tx_assign(tx, waited ? TXG_WAITED : TXG_NOWAIT);
  *	if (error) {
  *		rw_exit(...);		// drop locks
  *		zfs_dirent_unlock(dl);	// unlock directory entry
@@ -187,6 +187,47 @@
  *	return (error);			// done, report error
  */
 
+/* set this tunable to zero to disable asynchronous freeing of files */
+boolean_t zfs_do_async_free = B_TRUE;
+
+/*
+ * This value will be multiplied by zfs_dirty_data_max to determine
+ * the threshold past which we will call zfs_inactive_impl() async.
+ *
+ * Selecting the multiplier is a balance between how long we're willing to wait
+ * for delete/free to complete (get shell back, have a NFS thread captive, etc)
+ * and reducing the number of active requests in the backing taskq.
+ *
+ * 4 GiB (zfs_dirty_data_max default) * 16 (multiplier default) = 64 GiB
+ * meaning by default we will call zfs_inactive_impl async for vnodes > 64 GiB
+ *
+ * WARNING: Setting this tunable to zero will enable asynchronous freeing for
+ * all files which can have undesirable side effects.
+ */
+uint16_t zfs_inactive_async_multiplier = 16;
+
+int nms_worm_transition_time = 30;
+int
+zfs_worm_in_trans(znode_t *zp)
+{
+	zfsvfs_t		*zfsvfs = zp->z_zfsvfs;
+	timestruc_t		now;
+	sa_bulk_attr_t		bulk[2];
+	uint64_t		ctime[2];
+	int			count = 0;
+
+	if (!nms_worm_transition_time)
+		return (0);
+
+	gethrestime(&now);
+	SA_ADD_BULK_ATTR(bulk, count, SA_ZPL_CTIME(zfsvfs), NULL,
+	    &ctime, sizeof (ctime));
+	if (sa_bulk_lookup(zp->z_sa_hdl, bulk, count) != 0)
+		return (0);
+
+	return ((uint64_t)now.tv_sec - ctime[0] < nms_worm_transition_time);
+}
+
 /* ARGSUSED */
 static int
 zfs_open(vnode_t **vpp, int flag, cred_t *cr, caller_context_t *ct)
@@ -227,12 +268,13 @@
 {
 	znode_t	*zp = VTOZ(vp);
 	zfsvfs_t *zfsvfs = zp->z_zfsvfs;
+	pid_t caller_pid = (ct != NULL) ? ct->cc_pid : ddi_get_pid();
 
 	/*
 	 * Clean up any locks held by this process on the vp.
 	 */
-	cleanlocks(vp, ddi_get_pid(), 0);
-	cleanshares(vp, ddi_get_pid());
+	cleanlocks(vp, caller_pid, 0);
+	cleanshares(vp, caller_pid);
 
 	ZFS_ENTER(zfsvfs);
 	ZFS_VERIFY_ZP(zp);
@@ -486,6 +528,164 @@
 	return (error);
 }
 
+
+/*
+ * ZFS I/O rate throttling
+ */
+
+#define	DELAY_SHIFT 24
+
+typedef struct zfs_rate_delay {
+	uint_t rl_rate;
+	hrtime_t rl_delay;
+} zfs_rate_delay_t;
+
+/*
+ * The time we'll attempt to cv_wait (below), in nSec.
+ * This should be no less than the minimum time it normally takes
+ * to block a thread and wake back up after the timeout fires.
+ *
+ * Each table entry represents the delay for each 4MB of bandwith.
+ * we reduce the delay as the size fo the I/O increases.
+ */
+zfs_rate_delay_t zfs_rate_delay_table[] = {
+	{0, 100000},
+	{1, 100000},
+	{2, 100000},
+	{3, 100000},
+	{4, 100000},
+	{5, 50000},
+	{6, 50000},
+	{7, 50000},
+	{8, 50000},
+	{9, 25000},
+	{10, 25000},
+	{11, 25000},
+	{12, 25000},
+	{13, 12500},
+	{14, 12500},
+	{15, 12500},
+	{16, 12500},
+	{17, 6250},
+	{18, 6250},
+	{19, 6250},
+	{20, 6250},
+	{21, 3125},
+	{22, 3125},
+	{23, 3125},
+	{24, 3125},
+};
+
+#define	MAX_RATE_TBL_ENTRY 24
+
+/*
+ * The delay we use should be reduced based on the size of the iorate
+ * for higher iorates we want a shorter delay.
+ */
+static inline hrtime_t
+zfs_get_delay(ssize_t iorate)
+{
+	uint_t rate = iorate >> DELAY_SHIFT;
+
+	if (rate > MAX_RATE_TBL_ENTRY)
+		rate = MAX_RATE_TBL_ENTRY;
+	return (zfs_rate_delay_table[rate].rl_delay);
+}
+
+/*
+ * ZFS I/O rate throttling
+ * See "Token Bucket" on Wikipedia
+ *
+ * This is "Token Bucket" with some modifications to avoid wait times
+ * longer than a couple seconds, so that we don't trigger NFS retries
+ * or similar.  This does mean that concurrent requests might take us
+ * over the rate limit, but that's a lesser evil.
+ */
+static void
+zfs_rate_throttle(zfsvfs_t *zfsvfs, ssize_t iosize)
+{
+	zfs_rate_state_t *rate = &zfsvfs->z_rate;
+	hrtime_t now, delta; /* nanoseconds */
+	int64_t refill;
+
+	VERIFY(rate->rate_cap > 0);
+	mutex_enter(&rate->rate_lock);
+
+	/*
+	 * If another thread is already waiting, we must queue up behind them.
+	 * We'll wait up to 1 sec here.  We normally will resume by cv_signal,
+	 * so we don't need fine timer resolution on this wait.
+	 */
+	if (rate->rate_token_bucket < 0) {
+		rate->rate_waiters++;
+		(void) cv_timedwait_hires(
+		    &rate->rate_wait_cv, &rate->rate_lock,
+		    NANOSEC, TR_CLOCK_TICK, 0);
+		rate->rate_waiters--;
+	}
+
+	/*
+	 * How long since we last updated the bucket?
+	 */
+	now = gethrtime();
+	delta = now - rate->rate_last_update;
+	rate->rate_last_update = now;
+	if (delta < 0)
+		delta = 0; /* paranoid */
+
+	/*
+	 * Add "tokens" for time since last update,
+	 * being careful about possible overflow.
+	 */
+	refill = (delta * rate->rate_cap) / NANOSEC;
+	if (refill < 0 || refill > rate->rate_cap)
+		refill = rate->rate_cap; /* overflow */
+	rate->rate_token_bucket += refill;
+	if (rate->rate_token_bucket > rate->rate_cap)
+		rate->rate_token_bucket = rate->rate_cap;
+
+	/*
+	 * Withdraw tokens for the current I/O.* If this makes us overdrawn,
+	 * wait an amount of time proportionate to the overdraft.  However,
+	 * as a sanity measure, never wait more than 1 sec, and never try to
+	 * wait less than the time it normally takes to block and reschedule.
+	 *
+	 * Leave the bucket negative while we wait so other threads know to
+	 * queue up. In here, "refill" is the debt we're waiting to pay off.
+	 */
+	rate->rate_token_bucket -= iosize;
+	if (rate->rate_token_bucket < 0) {
+		hrtime_t zfs_rate_wait = 0;
+
+		refill = rate->rate_token_bucket;
+		DTRACE_PROBE2(zfs_rate_over, zfsvfs_t *, zfsvfs,
+		    int64_t, refill);
+
+		if (rate->rate_cap <= 0)
+			goto nocap;
+
+		delta = (refill * NANOSEC) / rate->rate_cap;
+		delta = MIN(delta, NANOSEC);
+
+		zfs_rate_wait = zfs_get_delay(rate->rate_cap);
+
+		if (delta > zfs_rate_wait) {
+			(void) cv_timedwait_hires(
+			    &rate->rate_wait_cv, &rate->rate_lock,
+			    delta, TR_CLOCK_TICK, 0);
+		}
+
+		rate->rate_token_bucket += refill;
+	}
+nocap:
+	if (rate->rate_waiters > 0) {
+		cv_signal(&rate->rate_wait_cv);
+	}
+
+	mutex_exit(&rate->rate_lock);
+}
+
+
 offset_t zfs_read_chunk_size = 1024 * 1024; /* Tunable */
 
 /*
@@ -552,6 +752,12 @@
 	}
 
 	/*
+	 * ZFS I/O rate throttling
+	 */
+	if (zfsvfs->z_rate.rate_cap)
+		zfs_rate_throttle(zfsvfs, uio->uio_resid);
+
+	/*
 	 * If we're in FRSYNC mode, sync out this znode before reading it.
 	 */
 	if (ioflag & FRSYNC || zfsvfs->z_os->os_sync == ZFS_SYNC_ALWAYS)
@@ -715,8 +921,12 @@
 	if ((zp->z_pflags & ZFS_IMMUTABLE) ||
 	    ((zp->z_pflags & ZFS_APPENDONLY) && !(ioflag & FAPPEND) &&
 	    (uio->uio_loffset < zp->z_size))) {
-		ZFS_EXIT(zfsvfs);
-		return (SET_ERROR(EPERM));
+		/* Make sure we're not a WORM before returning EPERM. */
+		if (!(zp->z_pflags & ZFS_IMMUTABLE) ||
+		    !zp->z_zfsvfs->z_isworm) {
+			ZFS_EXIT(zfsvfs);
+			return (SET_ERROR(EPERM));
+		}
 	}
 
 	zilog = zfsvfs->z_log;
@@ -741,6 +951,12 @@
 	}
 
 	/*
+	 * ZFS I/O rate throttling
+	 */
+	if (zfsvfs->z_rate.rate_cap)
+		zfs_rate_throttle(zfsvfs, uio->uio_resid);
+
+	/*
 	 * Pre-fault the pages to ensure slow (eg NFS) pages
 	 * don't hold up txg.
 	 * Skip this if uio contains loaned arc_buf.
@@ -1015,6 +1231,7 @@
 	return (0);
 }
 
+/* ARGSUSED */
 void
 zfs_get_done(zgd_t *zgd, int error)
 {
@@ -1032,9 +1249,6 @@
 	 */
 	VN_RELE_ASYNC(ZTOV(zp), dsl_pool_vnrele_taskq(dmu_objset_pool(os)));
 
-	if (error == 0 && zgd->zgd_bp)
-		zil_lwb_add_block(zgd->zgd_lwb, zgd->zgd_bp);
-
 	kmem_free(zgd, sizeof (zgd_t));
 }
 
@@ -1158,11 +1372,7 @@
 				 * 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,
-				 * so there's no need to zio_flush() its
-				 * vdevs (flushing would needlesly hurt
-				 * performance, and doesn't work on
-				 * indirect vdevs).
+				 * it is the old, currently-on-disk BP.
 				 */
 				zgd->zgd_bp = NULL;
 				BP_ZERO(bp);
@@ -1245,7 +1455,7 @@
     int flags, vnode_t *rdir, cred_t *cr,  caller_context_t *ct,
     int *direntflags, pathname_t *realpnp)
 {
-	znode_t *zdp = VTOZ(dvp);
+	znode_t *zp, *zdp = VTOZ(dvp);
 	zfsvfs_t *zfsvfs = zdp->z_zfsvfs;
 	int	error = 0;
 
@@ -1363,6 +1573,14 @@
 	error = zfs_dirlook(zdp, nm, vpp, flags, direntflags, realpnp);
 	if (error == 0)
 		error = specvp_check(vpp, cr);
+	if (*vpp) {
+		zp = VTOZ(*vpp);
+		if (!(zp->z_pflags & ZFS_IMMUTABLE) &&
+		    ((*vpp)->v_type != VDIR) &&
+		    zfsvfs->z_isworm && !zfs_worm_in_trans(zp)) {
+			zp->z_pflags |= ZFS_IMMUTABLE;
+		}
+	}
 
 	ZFS_EXIT(zfsvfs);
 	return (error);
@@ -1398,6 +1616,7 @@
     int mode, vnode_t **vpp, cred_t *cr, int flag, caller_context_t *ct,
     vsecattr_t *vsecp)
 {
+	int		imm_was_set = 0;
 	znode_t		*zp, *dzp = VTOZ(dvp);
 	zfsvfs_t	*zfsvfs = dzp->z_zfsvfs;
 	zilog_t		*zilog;
@@ -1483,6 +1702,12 @@
 	if (zp == NULL) {
 		uint64_t txtype;
 
+		if ((dzp->z_pflags & ZFS_IMMUTABLE) &&
+		    dzp->z_zfsvfs->z_isworm) {
+			imm_was_set = 1;
+			dzp->z_pflags &= ~ZFS_IMMUTABLE;
+		}
+
 		/*
 		 * Create a new file object and update the directory
 		 * to reference it.
@@ -1490,9 +1715,14 @@
 		if (error = zfs_zaccess(dzp, ACE_ADD_FILE, 0, B_FALSE, cr)) {
 			if (have_acl)
 				zfs_acl_ids_free(&acl_ids);
+			if (imm_was_set)
+				dzp->z_pflags |= ZFS_IMMUTABLE;
 			goto out;
 		}
 
+		if (imm_was_set)
+			dzp->z_pflags |= ZFS_IMMUTABLE;
+
 		/*
 		 * We only support the creation of regular files in
 		 * extended attribute directories.
@@ -1532,8 +1762,7 @@
 			dmu_tx_hold_write(tx, DMU_NEW_OBJECT,
 			    0, acl_ids.z_aclp->z_acl_bytes);
 		}
-		error = dmu_tx_assign(tx,
-		    (waited ? TXG_NOTHROTTLE : 0) | TXG_NOWAIT);
+		error = dmu_tx_assign(tx, waited ? TXG_WAITED : TXG_NOWAIT);
 		if (error) {
 			zfs_dirent_unlock(dl);
 			if (error == ERESTART) {
@@ -1552,6 +1781,9 @@
 		if (fuid_dirtied)
 			zfs_fuid_sync(zfsvfs, tx);
 
+		if (imm_was_set)
+			zp->z_pflags |= ZFS_IMMUTABLE;
+
 		(void) zfs_link_create(dl, zp, tx, ZNEW);
 		txtype = zfs_log_create_txtype(Z_FILE, vsecp, vap);
 		if (flag & FIGNORECASE)
@@ -1584,13 +1816,30 @@
 			error = SET_ERROR(EISDIR);
 			goto out;
 		}
+		if ((flag & FWRITE) &&
+		    dzp->z_zfsvfs->z_isworm) {
+			error = EPERM;
+			goto out;
+		}
+
+		if (!(flag & FAPPEND) &&
+		    (zp->z_pflags & ZFS_IMMUTABLE) &&
+		    dzp->z_zfsvfs->z_isworm) {
+			imm_was_set = 1;
+			zp->z_pflags &= ~ZFS_IMMUTABLE;
+		}
 		/*
 		 * Verify requested access to file.
 		 */
 		if (mode && (error = zfs_zaccess_rwx(zp, mode, aflags, cr))) {
+			if (imm_was_set)
+				zp->z_pflags |= ZFS_IMMUTABLE;
 			goto out;
 		}
 
+		if (imm_was_set)
+			zp->z_pflags |= ZFS_IMMUTABLE;
+
 		mutex_enter(&dzp->z_lock);
 		dzp->z_seq++;
 		mutex_exit(&dzp->z_lock);
@@ -1697,6 +1946,11 @@
 
 	vp = ZTOV(zp);
 
+	if (zp->z_zfsvfs->z_isworm) {
+		error = SET_ERROR(EPERM);
+		goto out;
+	}
+
 	if (error = zfs_zaccess_delete(dzp, zp, cr)) {
 		goto out;
 	}
@@ -1763,7 +2017,7 @@
 	 */
 	dmu_tx_mark_netfree(tx);
 
-	error = dmu_tx_assign(tx, (waited ? TXG_NOTHROTTLE : 0) | TXG_NOWAIT);
+	error = dmu_tx_assign(tx, waited ? TXG_WAITED : TXG_NOWAIT);
 	if (error) {
 		zfs_dirent_unlock(dl);
 		VN_RELE(vp);
@@ -1890,6 +2144,7 @@
 zfs_mkdir(vnode_t *dvp, char *dirname, vattr_t *vap, vnode_t **vpp, cred_t *cr,
     caller_context_t *ct, int flags, vsecattr_t *vsecp)
 {
+	int		imm_was_set = 0;
 	znode_t		*zp, *dzp = VTOZ(dvp);
 	zfsvfs_t	*zfsvfs = dzp->z_zfsvfs;
 	zilog_t		*zilog;
@@ -1969,7 +2224,15 @@
 		return (error);
 	}
 
+	if ((dzp->z_pflags & ZFS_IMMUTABLE) &&
+	    dzp->z_zfsvfs->z_isworm) {
+		imm_was_set = 1;
+		dzp->z_pflags &= ~ZFS_IMMUTABLE;
+	}
+
 	if (error = zfs_zaccess(dzp, ACE_ADD_SUBDIRECTORY, 0, B_FALSE, cr)) {
+		if (imm_was_set)
+			dzp->z_pflags |= ZFS_IMMUTABLE;
 		zfs_acl_ids_free(&acl_ids);
 		zfs_dirent_unlock(dl);
 		ZFS_EXIT(zfsvfs);
@@ -1976,6 +2239,9 @@
 		return (error);
 	}
 
+	if (imm_was_set)
+		dzp->z_pflags |= ZFS_IMMUTABLE;
+
 	if (zfs_acl_ids_overquota(zfsvfs, &acl_ids)) {
 		zfs_acl_ids_free(&acl_ids);
 		zfs_dirent_unlock(dl);
@@ -2000,7 +2266,7 @@
 	dmu_tx_hold_sa_create(tx, acl_ids.z_aclp->z_acl_bytes +
 	    ZFS_SA_BASE_ATTR_SIZE);
 
-	error = dmu_tx_assign(tx, (waited ? TXG_NOTHROTTLE : 0) | TXG_NOWAIT);
+	error = dmu_tx_assign(tx, waited ? TXG_WAITED : TXG_NOWAIT);
 	if (error) {
 		zfs_dirent_unlock(dl);
 		if (error == ERESTART) {
@@ -2102,6 +2368,11 @@
 
 	vp = ZTOV(zp);
 
+	if (dzp->z_zfsvfs->z_isworm) {
+		error = SET_ERROR(EPERM);
+		goto out;
+	}
+
 	if (error = zfs_zaccess_delete(dzp, zp, cr)) {
 		goto out;
 	}
@@ -2137,7 +2408,7 @@
 	zfs_sa_upgrade_txholds(tx, zp);
 	zfs_sa_upgrade_txholds(tx, dzp);
 	dmu_tx_mark_netfree(tx);
-	error = dmu_tx_assign(tx, (waited ? TXG_NOTHROTTLE : 0) | TXG_NOWAIT);
+	error = dmu_tx_assign(tx, waited ? TXG_WAITED : TXG_NOWAIT);
 	if (error) {
 		rw_exit(&zp->z_parent_lock);
 		rw_exit(&zp->z_name_lock);
@@ -2794,16 +3065,28 @@
 	xva_init(&tmpxvattr);
 
 	/*
-	 * Immutable files can only alter immutable bit and atime
+	 * Do not allow to alter immutable bit after it is set
 	 */
 	if ((zp->z_pflags & ZFS_IMMUTABLE) &&
-	    ((mask & (AT_SIZE|AT_UID|AT_GID|AT_MTIME|AT_MODE)) ||
-	    ((mask & AT_XVATTR) && XVA_ISSET_REQ(xvap, XAT_CREATETIME)))) {
+	    XVA_ISSET_REQ(xvap, XAT_IMMUTABLE) &&
+	    zp->z_zfsvfs->z_isworm) {
 		ZFS_EXIT(zfsvfs);
 		return (SET_ERROR(EPERM));
 	}
 
 	/*
+	 * Immutable files can only alter atime
+	 */
+	if (((zp->z_pflags & ZFS_IMMUTABLE) || zp->z_zfsvfs->z_isworm) &&
+	    ((mask & (AT_SIZE|AT_UID|AT_GID|AT_MTIME|AT_MODE)) ||
+	    ((mask & AT_XVATTR) && XVA_ISSET_REQ(xvap, XAT_CREATETIME)))) {
+		if (!zp->z_zfsvfs->z_isworm || !zfs_worm_in_trans(zp)) {
+			ZFS_EXIT(zfsvfs);
+			return (SET_ERROR(EPERM));
+		}
+	}
+
+	/*
 	 * Note: ZFS_READONLY is handled in zfs_zaccess_common.
 	 */
 
@@ -3710,7 +3993,7 @@
 
 	zfs_sa_upgrade_txholds(tx, szp);
 	dmu_tx_hold_zap(tx, zfsvfs->z_unlinkedobj, FALSE, NULL);
-	error = dmu_tx_assign(tx, (waited ? TXG_NOTHROTTLE : 0) | TXG_NOWAIT);
+	error = dmu_tx_assign(tx, waited ? TXG_WAITED : TXG_NOWAIT);
 	if (error) {
 		if (zl != NULL)
 			zfs_rename_unlock(&zl);
@@ -3834,6 +4117,7 @@
 	dmu_tx_t	*tx;
 	zfsvfs_t	*zfsvfs = dzp->z_zfsvfs;
 	zilog_t		*zilog;
+	int		imm_was_set = 0;
 	uint64_t	len = strlen(link);
 	int		error;
 	int		zflg = ZNEW;
@@ -3877,12 +4161,20 @@
 		return (error);
 	}
 
+	if ((dzp->z_pflags & ZFS_IMMUTABLE) && dzp->z_zfsvfs->z_isworm) {
+		imm_was_set = 1;
+		dzp->z_pflags &= ~ZFS_IMMUTABLE;
+	}
 	if (error = zfs_zaccess(dzp, ACE_ADD_FILE, 0, B_FALSE, cr)) {
+		if (imm_was_set)
+			dzp->z_pflags |= ZFS_IMMUTABLE;
 		zfs_acl_ids_free(&acl_ids);
 		zfs_dirent_unlock(dl);
 		ZFS_EXIT(zfsvfs);
 		return (error);
 	}
+	if (imm_was_set)
+		dzp->z_pflags |= ZFS_IMMUTABLE;
 
 	if (zfs_acl_ids_overquota(zfsvfs, &acl_ids)) {
 		zfs_acl_ids_free(&acl_ids);
@@ -3903,7 +4195,7 @@
 	}
 	if (fuid_dirtied)
 		zfs_fuid_txhold(zfsvfs, tx);
-	error = dmu_tx_assign(tx, (waited ? TXG_NOTHROTTLE : 0) | TXG_NOWAIT);
+	error = dmu_tx_assign(tx, waited ? TXG_WAITED : TXG_NOWAIT);
 	if (error) {
 		zfs_dirent_unlock(dl);
 		if (error == ERESTART) {
@@ -4124,7 +4416,7 @@
 	dmu_tx_hold_zap(tx, dzp->z_id, TRUE, name);
 	zfs_sa_upgrade_txholds(tx, szp);
 	zfs_sa_upgrade_txholds(tx, dzp);
-	error = dmu_tx_assign(tx, (waited ? TXG_NOTHROTTLE : 0) | TXG_NOWAIT);
+	error = dmu_tx_assign(tx, waited ? TXG_WAITED : TXG_NOWAIT);
 	if (error) {
 		zfs_dirent_unlock(dl);
 		if (error == ERESTART) {
@@ -4400,15 +4692,18 @@
 	return (error);
 }
 
-/*ARGSUSED*/
-void
-zfs_inactive(vnode_t *vp, cred_t *cr, caller_context_t *ct)
+/*
+ * Returns B_TRUE and exits the z_teardown_inactive_lock
+ * if the znode we are looking at is no longer valid
+ */
+static boolean_t
+zfs_znode_free_invalid(znode_t *zp)
 {
-	znode_t	*zp = VTOZ(vp);
 	zfsvfs_t *zfsvfs = zp->z_zfsvfs;
-	int error;
+	vnode_t *vp = ZTOV(zp);
 
-	rw_enter(&zfsvfs->z_teardown_inactive_lock, RW_READER);
+	ASSERT(rw_read_held(&zfsvfs->z_teardown_inactive_lock));
+
 	if (zp->z_sa_hdl == NULL) {
 		/*
 		 * The fs has been unmounted, or we did a
@@ -4416,7 +4711,7 @@
 		 */
 		if (vn_has_cached_data(vp)) {
 			(void) pvn_vplist_dirty(vp, 0, zfs_null_putapage,
-			    B_INVAL, cr);
+			    B_INVAL, CRED());
 		}
 
 		mutex_enter(&zp->z_lock);
@@ -4425,11 +4720,32 @@
 		VN_RELE_LOCKED(vp);
 		mutex_exit(&vp->v_lock);
 		mutex_exit(&zp->z_lock);
+		VERIFY(atomic_dec_32_nv(&zfsvfs->z_znodes_freeing_cnt) !=
+		    UINT32_MAX);
 		rw_exit(&zfsvfs->z_teardown_inactive_lock);
 		zfs_znode_free(zp);
-		return;
+		return (B_TRUE);
 	}
 
+	return (B_FALSE);
+}
+
+/*
+ * Does the prep work for freeing the znode, then calls zfs_zinactive to do the
+ * actual freeing.
+ * This code used be in zfs_inactive() before the async delete patch came in
+ */
+static void
+zfs_inactive_impl(znode_t *zp)
+{
+	vnode_t	*vp = ZTOV(zp);
+	zfsvfs_t *zfsvfs = zp->z_zfsvfs;
+	int error;
+
+	rw_enter(&zfsvfs->z_teardown_inactive_lock, RW_READER_STARVEWRITER);
+	if (zfs_znode_free_invalid(zp))
+		return; /* z_teardown_inactive_lock already dropped */
+
 	/*
 	 * Attempt to push any data in the page cache.  If this fails
 	 * we will get kicked out later in zfs_zinactive().
@@ -4436,7 +4752,7 @@
 	 */
 	if (vn_has_cached_data(vp)) {
 		(void) pvn_vplist_dirty(vp, 0, zfs_putapage, B_INVAL|B_ASYNC,
-		    cr);
+		    CRED());
 	}
 
 	if (zp->z_atime_dirty && zp->z_unlinked == 0) {
@@ -4458,9 +4774,51 @@
 	}
 
 	zfs_zinactive(zp);
+
+	VERIFY(atomic_dec_32_nv(&zfsvfs->z_znodes_freeing_cnt) != UINT32_MAX);
+
 	rw_exit(&zfsvfs->z_teardown_inactive_lock);
 }
 
+/*
+ * taskq task calls zfs_inactive_impl() so that we can free the znode
+ */
+static void
+zfs_inactive_task(void *task_arg)
+{
+	znode_t *zp = task_arg;
+	ASSERT(zp != NULL);
+	zfs_inactive_impl(zp);
+}
+
+/*ARGSUSED*/
+void
+zfs_inactive(vnode_t *vp, cred_t *cr, caller_context_t *ct)
+{
+	znode_t	*zp = VTOZ(vp);
+	zfsvfs_t *zfsvfs = zp->z_zfsvfs;
+
+	rw_enter(&zfsvfs->z_teardown_inactive_lock, RW_READER_STARVEWRITER);
+
+	VERIFY(atomic_inc_32_nv(&zfsvfs->z_znodes_freeing_cnt) != 0);
+
+	if (zfs_znode_free_invalid(zp))
+		return; /* z_teardown_inactive_lock already dropped */
+
+	if (zfs_do_async_free &&
+	    zp->z_size > zfs_inactive_async_multiplier * zfs_dirty_data_max &&
+	    taskq_dispatch(dsl_pool_vnrele_taskq(
+	    dmu_objset_pool(zp->z_zfsvfs->z_os)), zfs_inactive_task,
+	    zp, TQ_NOSLEEP) != NULL) {
+		rw_exit(&zfsvfs->z_teardown_inactive_lock);
+		return; /* task dispatched, we're done */
+	}
+	rw_exit(&zfsvfs->z_teardown_inactive_lock);
+
+	/* if the taskq dispatch failed - do a sync zfs_inactive_impl() call */
+	zfs_inactive_impl(zp);
+}
+
 /*
  * Bounds-check the seek operation.
  *
