diff --git a/fs/fuse/Makefile b/fs/fuse/Makefile
index f54c504ca6637c..1a95e59e32d26d 100644
--- a/fs/fuse/Makefile
+++ b/fs/fuse/Makefile
@@ -10,7 +10,7 @@ obj-$(CONFIG_FUSE_FS) += fuse.o
 obj-$(CONFIG_CUSE) += cuse.o
 obj-$(CONFIG_VIRTIO_FS) += virtiofs.o
 
-fuse-y := dev.o dir.o file.o inode.o control.o xattr.o acl.o readdir.o ioctl.o fuse_dlm_cache.o compound.o
+fuse-y := dev.o dir.o file.o inode.o control.o xattr.o acl.o readdir.o ioctl.o fuse_dlm_cache.o fuse_range_lock.o compound.o
 fuse-y += iomode.o
 fuse-$(CONFIG_FUSE_DAX) += dax.o
 fuse-$(CONFIG_FUSE_PASSTHROUGH) += passthrough.o
diff --git a/fs/fuse/dev.c b/fs/fuse/dev.c
index 333ce28f4f920c..5ce32a5376543a 100644
--- a/fs/fuse/dev.c
+++ b/fs/fuse/dev.c
@@ -433,9 +433,10 @@ static void flush_bg_queue(struct fuse_conn *fc)
  * This function is called when a request is finished.  Either a reply
  * has arrived or it was aborted (and not yet sent) or some error
  * occurred during communication with userspace, or the device file
- * was closed.  The requester thread is woken up (if still waiting),
- * the 'end' callback is called if given, else the reference to the
- * request is released
+ * was closed.  The 'complete' callback, if given, is called first, on
+ * this thread; only then is the requester thread woken up (if still
+ * waiting), the 'end' callback called if given, else the reference to
+ * the request is released.
  */
 void fuse_request_end(struct fuse_req *req)
 {
@@ -459,6 +460,10 @@ void fuse_request_end(struct fuse_req *req)
 	}
 	WARN_ON(test_bit(FR_PENDING, &req->flags));
 	WARN_ON(test_bit(FR_SENT, &req->flags));
+
+	if (req->args->complete)
+		req->args->complete(fm, req->args, req->out.h.error);
+
 	if (test_bit(FR_BACKGROUND, &req->flags)) {
 		spin_lock(&fc->bg_lock);
 		clear_bit(FR_BACKGROUND, &req->flags);
diff --git a/fs/fuse/dir.c b/fs/fuse/dir.c
index 5f426228f4c76a..c628f01d05c210 100644
--- a/fs/fuse/dir.c
+++ b/fs/fuse/dir.c
@@ -2106,34 +2106,41 @@ int fuse_do_setattr(struct mnt_idmap *idmap, struct dentry *dentry,
 		WARN_ON(!(attr->ia_valid & ATTR_SIZE));
 		WARN_ON(attr->ia_size != 0);
 		if (fc->atomic_o_trunc) {
-			struct percpu_rw_semaphore *wb_sem = fi->wb_inval_rwsem;
+			struct fuse_range_lock rlock;
+			bool range_locked = fc->writeback_cache && fc->dlm;
 
 			/*
 			 * No need to send request to userspace, since actual
 			 * truncation has already been done by OPEN.  But still
 			 * need to truncate page cache.
 			 *
-			 * Revoke and drop under the coherency gate write side,
-			 * like the NOTIFY invalidate path: a gate reader that
-			 * already re-validated its grant must not have the
-			 * lock tree and the cache yanked mid-hold, or it
-			 * would repopulate the truncated range trusting a
-			 * grant that no longer exists.  Waiting for gate
-			 * readers here is safe: we hold i_rwsem exclusive, so
-			 * no gate holder can be waiting on it (the write path
-			 * takes i_rwsem before the gate, the read path never
-			 * takes it).
+			 * Revoke and drop under the full-range IO range lock,
+			 * like the NOTIFY invalidate path
+			 * (fuse_reverse_inval_inode()): a reader/writer that
+			 * already reached READY state must not have the lock
+			 * tree and the cache yanked mid-hold, or it would
+			 * repopulate the truncated range trusting a grant that
+			 * no longer exists.  fuse_range_lock_acquire_ready()
+			 * ignores an overlapping INIT range (a read/write with
+			 * only a DLM request in flight), so waiting here is
+			 * bounded.  Blocking is also safe: we hold i_rwsem
+			 * exclusive, so no cached writer can be waiting on this
+			 * range lock (the write path takes i_rwsem before it,
+			 * the read path never takes i_rwsem at all).  Only
+			 * meaningful under DLM with the writeback cache; see
+			 * fuse_range_lock.h.
 			 */
-			if (wb_sem)
-				percpu_down_write(wb_sem);
+			if (range_locked)
+				fuse_range_lock_acquire_ready(fi, &rlock, 0, ~0ULL,
+							      FUSE_RANGE_LOCK_WRITE);
 			if (fc->dlm && fc->writeback_cache)
 				fuse_dlm_cache_release_locks(fi);
 			spin_lock(&fi->lock);
 			i_size_write(inode, 0);
 			spin_unlock(&fi->lock);
 			truncate_pagecache(inode, 0);
-			if (wb_sem)
-				percpu_up_write(wb_sem);
+			if (range_locked)
+				fuse_range_lock_release(fi, &rlock);
 			goto out;
 		}
 		file = NULL;
@@ -2237,23 +2244,26 @@ int fuse_do_setattr(struct mnt_idmap *idmap, struct dentry *dentry,
 	 */
 	if ((is_truncate || !is_wb) &&
 	    S_ISREG(inode->i_mode) && oldsize != outarg.attr.size) {
-		struct percpu_rw_semaphore *wb_sem = fi->wb_inval_rwsem;
+		struct fuse_range_lock rlock;
+		bool range_locked = fc->writeback_cache && fc->dlm;
 
 		/*
-		 * Revoke and drop under the coherency gate write side; see
+		 * Revoke and drop under the full-range IO range lock; see
 		 * the atomic-O_TRUNC branch above.  i_rwsem is held
-		 * exclusive here as well (setattr), so waiting out gate
-		 * readers cannot deadlock.
+		 * exclusive here as well (setattr), so waiting out
+		 * in-progress READY IO cannot deadlock.  Only meaningful
+		 * under DLM with the writeback cache; see fuse_range_lock.h.
 		 */
-		if (wb_sem)
-			percpu_down_write(wb_sem);
+		if (range_locked)
+			fuse_range_lock_acquire_ready(fi, &rlock, 0, ~0ULL,
+						      FUSE_RANGE_LOCK_WRITE);
 		if (fc->dlm && fc->writeback_cache)
 			fuse_dlm_unlock_range(fi, outarg.attr.size & PAGE_MASK, -1);
 
 		truncate_pagecache(inode, outarg.attr.size);
 		invalidate_inode_pages2(mapping);
-		if (wb_sem)
-			percpu_up_write(wb_sem);
+		if (range_locked)
+			fuse_range_lock_release(fi, &rlock);
 	}
 
 	clear_bit(FUSE_I_SIZE_UNSTABLE, &fi->state);
diff --git a/fs/fuse/file.c b/fs/fuse/file.c
index 7c5b5e8b4ba268..b797460e2b1ad4 100644
--- a/fs/fuse/file.c
+++ b/fs/fuse/file.c
@@ -469,8 +469,8 @@ void fuse_file_release(struct inode *inode, struct fuse_file *ff,
 	 * If this release dropped the last writer, fuse_prepare_release()
 	 * cleared the forced-direct-IO latch (under fi->lock).  Drop any clean
 	 * folios a read racing the latch may have repopulated so they cannot be
-	 * served stale once caching mode resumes.  No inode lock or
-	 * wb_inval_rwsem: release may run on the fuse server thread (async fput
+	 * served stale once caching mode resumes.  No inode lock or IO range
+	 * lock taken: release may run on the fuse server thread (async fput
 	 * from aio completion), where blocking on a contended inode lock could
 	 * stall the connection.  Writes were routed direct while latched, so
 	 * only clean folios exist and this invalidate is server-free; the last
@@ -1227,21 +1227,16 @@ static void fuse_readahead(struct readahead_control *rac)
 
 static ssize_t fuse_direct_read_iter(struct kiocb *iocb, struct iov_iter *to);
 
-/*
- * Bound on re-requesting a revoked DLM grant before a cached read is
- * served unlocked; see fuse_cache_read_iter().
- */
-#define FUSE_DLM_READ_RETRIES 3
-
 static ssize_t fuse_cache_read_iter(struct kiocb *iocb, struct iov_iter *to)
 {
 	struct file *file = iocb->ki_filp;
 	struct inode *inode = file->f_mapping->host;
 	struct fuse_conn *fc = get_fuse_conn(inode);
 	struct fuse_inode *fi = get_fuse_inode(inode);
-	struct percpu_rw_semaphore *wb_sem = fi->wb_inval_rwsem;
+	struct fuse_range_lock rlock;
+	size_t count = iov_iter_count(to);
+	bool range_locked = false;
 	ssize_t res;
-	int lock_err = 0;
 
 	/*
 	 * In auto invalidate mode, always update attributes on read.
@@ -1256,67 +1251,66 @@ static ssize_t fuse_cache_read_iter(struct kiocb *iocb, struct iov_iter *to)
 			return err;
 	}
 
-	/* if we have dlm support acquire a read lock for the area
-	 * we are reading from. */
+	/*
+	 * Reserve the IO range lock in INIT state over the bytes this read
+	 * will touch before requesting the DLM read lock below: an INIT
+	 * range is invisible to fuse_reverse_inval_inode(), so a NOTIFY
+	 * invalidate whose range overlaps ours can run to completion --
+	 * revoking the grant requested below and dropping its page-cache
+	 * range -- without waiting out this (unbounded, cluster round trip)
+	 * request.  Only meaningful under DLM with the writeback cache:
+	 * without DLM there is no round trip to protect against, and the
+	 * truncate / invalidate paths' fuse_range_lock_acquire_ready()
+	 * calls simply find no overlapping node to wait on; without the
+	 * writeback cache there is no DLM-covered write to race against.
+	 */
+	if (fc->writeback_cache && fc->dlm && count) {
+		range_locked = true;
+		fuse_range_lock_acquire_init(fi, &rlock, iocb->ki_pos,
+					     iocb->ki_pos + count - 1,
+					     FUSE_RANGE_LOCK_READ);
+	}
+
+	/*
+	 * If we have dlm support acquire a read lock for the area we are
+	 * reading from.  Passing the range lock through moves it to READY
+	 * as part of processing the reply -- as soon as the grant (or a
+	 * "no DLM" reply) comes back, rather than only later, right below.
+	 */
 	if (fc->writeback_cache && fc->dlm)
-		lock_err = fuse_get_dlm_lock(file, iocb->ki_pos,
-					     iov_iter_count(to),
-					     FUSE_PAGE_LOCK_READ);
+		fuse_get_dlm_lock(file, iocb->ki_pos, iov_iter_count(to),
+				  FUSE_PAGE_LOCK_READ,
+				  range_locked ? &rlock : NULL);
 
 	/*
-	 * Fence the cache-serving read against a NOTIFY invalidate so we never
-	 * hand back a folio the server has just superseded.  The gate read side
-	 * is per-CPU cheap; the NOTIFY holds the write side with priority.
-	 * Re-check the forced-DIO latch under it: if a storm latched us while we
-	 * waited on a pending writer, reroute to direct like the buffered write
-	 * path, so we do not repopulate the cache the latch just dropped.
-	 * wb_sem is NULL on non-writeback+dlm mounts (gate inactive).
+	 * Ensure the range lock is READY before touching the page cache:
+	 * the DLM reply above already moved it there when it was taken;
+	 * this is then just a confirmation.  It still does the transition
+	 * itself when no DLM request was made above (no writeback cache or
+	 * no dlm), and blocks only if a NOTIFY invalidate is currently
+	 * draining an overlapping range -- once granted it fences any *new*
+	 * overlapping invalidate until the range lock is released below.
+	 * An invalidate that instead ran to completion entirely while we
+	 * were still in INIT above (and so invisible to it) has already
+	 * dropped whatever page-cache range it revoked, so no re-validation
+	 * of the DLM grant requested above is needed here: either way, the
+	 * page cache this read is about to see is consistent.  A
+	 * revoked-but-unnoticed grant costs at most an extra cache-miss
+	 * round trip on this or the next read, never stale data.
 	 */
-	if (wb_sem) {
-		int tries = FUSE_DLM_READ_RETRIES;
+	if (range_locked) {
+		fuse_range_lock_mark_ready(fi, &rlock);
 
-retry:
-		percpu_down_read(wb_sem);
 		if (fuse_inode_force_dio(inode)) {
-			percpu_up_read(wb_sem);
+			fuse_range_lock_release(fi, &rlock);
 			return fuse_direct_read_iter(iocb, to);
 		}
-		/*
-		 * The DLM lock was requested before entering the gate, and
-		 * the NOTIFY invalidate we may just have waited on revokes
-		 * locks under the gate write side.  Re-check the grant here
-		 * and re-request with the gate dropped, so a
-		 * FUSE_DLM_WB_LOCK round trip never parks a pending
-		 * invalidate behind our own gate hold.  Once the check
-		 * passes the lock cannot go away for the rest of the gate
-		 * hold.  A failed or unrecorded request falls through
-		 * unlocked, as before: the retry is taken even then (the
-		 * latch must be re-checked under the re-entered gate), so
-		 * lock_err has to stay sticky across it -- seeded by the
-		 * pre-gate request above -- or a grant that failed would
-		 * be re-requested forever.  The retry is also bounded: a
-		 * remote writer can revoke each successful grant before
-		 * the gate is re-entered, and a reader-only inode has no
-		 * force-DIO latch to end such a storm, so after
-		 * FUSE_DLM_READ_RETRIES re-requests the read is served
-		 * unlocked rather than looping without bound.
-		 */
-		if (!lock_err && fc->dlm && tries-- > 0 &&
-		    !fuse_dlm_lock_is_held(fi, iocb->ki_pos,
-					   iov_iter_count(to),
-					   FUSE_PAGE_LOCK_READ)) {
-			percpu_up_read(wb_sem);
-			lock_err = fuse_get_dlm_lock(file, iocb->ki_pos,
-						     iov_iter_count(to),
-						     FUSE_PAGE_LOCK_READ);
-			goto retry;
-		}
 	}
 
 	res = generic_file_read_iter(iocb, to);
 
-	if (wb_sem)
-		percpu_up_read(wb_sem);
+	if (range_locked)
+		fuse_range_lock_release(fi, &rlock);
 
 	return res;
 }
@@ -1902,19 +1896,20 @@ static void fuse_cache_wr_unlock(struct inode *inode, bool exclusive)
  * fc->dlm: the server has no DLM, proceed as a plain cached write.  Any
  * other failure means the cache would be dirtied without DLM coverage -
  * the caller must fail the write instead.  A granted-but-unrecorded
- * lock (positive return) is covered cluster-wide; proceed, but flag it
- * so the in-gate re-validation skips a check an invisible grant could
- * never pass.
+ * lock (positive return) is covered cluster-wide; proceed regardless.
+ *
+ * @rlock: passed straight through to fuse_get_dlm_lock(); NULL unless
+ * @rlock is the range lock the caller will actually touch the page
+ * cache under, since that is what "reply processing marks it READY"
+ * is meant to cover.
  */
 static int fuse_cache_wr_dlm_lock(struct file *file, loff_t pos, size_t len,
-				  bool *unrecorded)
+				  struct fuse_range_lock *rlock)
 {
-	int err = fuse_get_dlm_lock(file, pos, len, FUSE_PAGE_LOCK_WRITE);
+	int err = fuse_get_dlm_lock(file, pos, len, FUSE_PAGE_LOCK_WRITE,
+				    rlock);
 
-	if (err < 0 && err != -ENOSYS)
-		return err;
-	*unrecorded = err > 0;
-	return 0;
+	return (err < 0 && err != -ENOSYS) ? err : 0;
 }
 
 static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
@@ -1927,11 +1922,10 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 	ssize_t err, count;
 	struct fuse_conn *fc = get_fuse_conn(inode);
 	struct fuse_inode *fi = get_fuse_inode(inode);
-	struct percpu_rw_semaphore *wb_sem = fi->wb_inval_rwsem;
+	struct fuse_range_lock rlock;
 	bool writeback = false;
-	bool wb_guard = false;
+	bool range_locked = false;
 	bool exclusive = true;
-	bool dlm_unrecorded = false;
 	loff_t dlm_pos = 0;
 	size_t dlm_len = 0;
 
@@ -1980,22 +1974,41 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 	exclusive = fuse_cache_wr_exclusive_lock(iocb, writeback);
 
 	/*
-	 * Request the DLM write lock before taking i_rwsem: the request is
-	 * an unbounded cluster round trip, and holding the writer-priority
+	 * Reserve the IO range lock in INIT state over the byte range this
+	 * write will (provisionally) touch before requesting the DLM write
+	 * lock below, and before taking i_rwsem: the request is an
+	 * unbounded cluster round trip, and holding the writer-priority
 	 * rwsem across it would park a truncate -- and behind it every
-	 * later writer -- for the duration.  The grant-to-use window this
-	 * leaves open is closed by the in-gate re-validation below.  Only
-	 * the append case must wait for the lock: its range depends on
-	 * i_size, which is stable only under the exclusive inode lock.
+	 * later writer -- for the duration.  An INIT range is invisible to
+	 * fuse_reverse_inval_inode(), so a NOTIFY invalidate overlapping
+	 * this range runs to completion instead of waiting out the
+	 * request.  The grant-to-use window this leaves open is closed by
+	 * the exact-range DLM request once the write's final range is known
+	 * below, which moves the range lock to READY as part of its reply.
+	 * Only the append case must wait for the lock: its range depends
+	 * on i_size, which is stable only under the exclusive inode lock.
 	 */
 	if (writeback && fc->dlm && !(iocb->ki_flags & IOCB_APPEND)) {
 		dlm_pos = iocb->ki_pos;
 		dlm_len = iov_iter_count(from);
 
-		err = fuse_cache_wr_dlm_lock(file, dlm_pos, dlm_len,
-					     &dlm_unrecorded);
-		if (err)
+		range_locked = true;
+		fuse_range_lock_acquire_init(fi, &rlock, dlm_pos,
+					     dlm_pos + dlm_len - 1,
+					     FUSE_RANGE_LOCK_WRITE);
+
+		/*
+		 * NULL rlock: this provisional range is released and
+		 * re-acquired narrower, in INIT state, once the exact write
+		 * range is known below, so it is not the range lock this
+		 * write actually touches the page cache under -- that one
+		 * is requested, and marked READY, further down.
+		 */
+		err = fuse_cache_wr_dlm_lock(file, dlm_pos, dlm_len, NULL);
+		if (err) {
+			fuse_range_lock_release(fi, &rlock);
 			return err;
+		}
 
 		/*
 		 * The request above may have found that the server has no DLM
@@ -2029,8 +2042,14 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 		dlm_pos = i_size_read(inode);
 		dlm_len = iov_iter_count(from);
 
-		err = fuse_cache_wr_dlm_lock(file, dlm_pos, dlm_len,
-					     &dlm_unrecorded);
+		range_locked = true;
+		fuse_range_lock_acquire_init(fi, &rlock, dlm_pos,
+					     dlm_pos + dlm_len - 1,
+					     FUSE_RANGE_LOCK_WRITE);
+
+		/* NULL rlock: provisional range, see the comment above the
+		 * first fuse_cache_wr_dlm_lock() call above. */
+		err = fuse_cache_wr_dlm_lock(file, dlm_pos, dlm_len, NULL);
 		if (err)
 			goto out;
 	}
@@ -2044,31 +2063,39 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 	 * attribute replies move it under fi->lock alone, so
 	 * generic_write_checks() may have put ki_pos past the granted
 	 * range.  Re-lock where the write really lands; dlm_pos tracks it
-	 * so the in-gate re-validation below guards the same range.
+	 * so the exact-range request further down covers the same range.
 	 */
 	if (writeback && fc->dlm && (iocb->ki_flags & IOCB_APPEND) &&
 	    iocb->ki_pos != dlm_pos) {
+
+		fuse_range_lock_release(fi, &rlock);
 		dlm_pos = iocb->ki_pos;
 		dlm_len = count;
+		fuse_range_lock_acquire_init(fi, &rlock, dlm_pos,
+					     dlm_pos + dlm_len - 1,
+					     FUSE_RANGE_LOCK_WRITE);
 
-		err = fuse_cache_wr_dlm_lock(file, dlm_pos, dlm_len,
-					     &dlm_unrecorded);
+		/* NULL rlock: provisional range, see the comment above the
+		 * first fuse_cache_wr_dlm_lock() call above. */
+		err = fuse_cache_wr_dlm_lock(file, dlm_pos, dlm_len, NULL);
 		if (err)
 			goto out;
 	}
 
 	/*
-	 * Kill suid/sgid and stamp the timestamps here, before the gate,
-	 * instead of leaving them next to the write itself.  kiocb_modified()
-	 * -> file_remove_privs() is the one that reaches the server: without
-	 * handle_killpriv[_v2] fuse_setattr() kills the bits by asking it (a
-	 * FUSE_GETATTR to refresh the mode, then a FUSE_SETATTR, which for a
-	 * writeback inode first flushes and freezes writepages), and
-	 * security_inode_killpriv() can drop the capability xattr with another
-	 * round trip.  A server may have to invalidate this inode from inside
-	 * such a handler; its NOTIFY_INVAL_INODE then blocks in
-	 * percpu_down_write() draining a gate reader that is itself waiting for
-	 * the reply.  Nothing held under the gate may wait for the server.
+	 * Kill suid/sgid and stamp the timestamps here, before the range
+	 * lock moves to READY below, instead of leaving them next to the
+	 * write itself.  kiocb_modified() -> file_remove_privs() is the one
+	 * that reaches the server: without handle_killpriv[_v2]
+	 * fuse_setattr() kills the bits by asking it (a FUSE_GETATTR to
+	 * refresh the mode, then a FUSE_SETATTR, which for a writeback
+	 * inode first flushes and freezes writepages), and
+	 * security_inode_killpriv() can drop the capability xattr with
+	 * another round trip.  A server may have to invalidate this inode
+	 * from inside such a handler; its NOTIFY_INVAL_INODE then calls
+	 * fuse_range_lock_acquire_ready(), which does not wait on our INIT
+	 * range.  Nothing held once the range lock is READY may wait for
+	 * the server.
 	 *
 	 * This also runs before the forced-DIO re-route below, so a re-routed
 	 * write repeats it; there is nothing left to do the second time.
@@ -2077,29 +2104,66 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 	if (err)
 		goto out;
 
-	wb_guard = !!wb_sem;
-	if (wb_guard) {
-retry:
-		percpu_down_read(wb_sem);
-		if (fuse_inode_force_dio(inode)) {
-			percpu_up_read(wb_sem);
-			fuse_cache_wr_unlock(inode, exclusive);
-			return fuse_direct_write_iter(iocb, from);
-		}
-		if (writeback && fc->dlm && !dlm_unrecorded &&
-		    !fuse_dlm_lock_is_held(fi, dlm_pos, dlm_len,
-					   FUSE_PAGE_LOCK_WRITE)) {
-			percpu_up_read(wb_sem);
-			err = fuse_cache_wr_dlm_lock(file, dlm_pos, dlm_len,
-						     &dlm_unrecorded);
-			if (err) {
-				/* The gate is already dropped; funnel the
-				 * failure through the one audited exit. */
-				wb_guard = false;
-				goto out;
-			}
-			goto retry;
-		}
+	/*
+	 * Narrow the range lock reservation from the provisional DLM
+	 * request range above (if any) to the exact range this write will
+	 * touch -- generic_write_checks() may have trimmed count below the
+	 * provisional dlm_len -- staying in INIT state so an invalidate
+	 * that is already draining an overlapping range is not waited on
+	 * here either.  Only reacquired while still on the DLM path: a
+	 * fuse_cache_wr_dlm_lock() call above may have found the server
+	 * has no DLM and cleared fc->dlm, in which case the exclusive
+	 * i_rwsem already serializes this write against invalidation and
+	 * the range lock is not needed.
+	 */
+	if (range_locked) {
+		fuse_range_lock_release(fi, &rlock);
+		range_locked = false;
+	}
+	if (writeback && fc->dlm) {
+		dlm_pos = iocb->ki_pos;
+		dlm_len = count;
+
+		range_locked = true;
+		fuse_range_lock_acquire_init(fi, &rlock, dlm_pos,
+					     dlm_pos + dlm_len - 1,
+					     FUSE_RANGE_LOCK_WRITE);
+
+		/*
+		 * Request the DLM write lock for the exact range this write
+		 * will touch, with the range lock itself passed through this
+		 * time: fuse_get_dlm_lock() moves it to READY as part of
+		 * processing the reply (or right away, if the range is
+		 * already covered by the provisional grant above -- see its
+		 * fast path).  Nothing is left to re-validate below.
+		 */
+		err = fuse_cache_wr_dlm_lock(file, dlm_pos, dlm_len, &rlock);
+		if (err)
+			goto out;
+	}
+
+	/*
+	 * Move the range lock to READY.  The DLM request above already made
+	 * this transition as part of processing its reply, so this is just
+	 * a confirmation; it still does the transition itself when nothing
+	 * was reserved above (no DLM), in which case it never blocks.
+	 * Otherwise this blocks only if a NOTIFY invalidate is currently
+	 * draining an overlapping range, and once granted fences any *new*
+	 * overlapping invalidate until the range lock is released below
+	 * (see out:).  This is what gives invalidation an exact conflict
+	 * test instead of draining every in-flight cached reader/writer on
+	 * the inode.  It also serializes us against a concurrent
+	 * fuse_cache_read_iter() on an overlapping range, and against
+	 * another shared-locked writer on an overlapping range.
+	 */
+	if (range_locked)
+		fuse_range_lock_mark_ready(fi, &rlock);
+
+	if (fuse_inode_force_dio(inode)) {
+		if (range_locked)
+			fuse_range_lock_release(fi, &rlock);
+		fuse_cache_wr_unlock(inode, exclusive);
+		return fuse_direct_write_iter(iocb, from);
 	}
 
 	task_io_account_write(count);
@@ -2189,8 +2253,8 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 		written = fuse_perform_write(iocb, from, false);
 	}
 out:
-	if (wb_guard)
-		percpu_up_read(wb_sem);
+	if (range_locked)
+		fuse_range_lock_release(fi, &rlock);
 	fuse_cache_wr_unlock(inode, exclusive);
 	if (written > 0)
 		written = generic_write_sync(iocb, written);
@@ -3140,7 +3204,7 @@ static int fuse_file_mmap(struct file *file, struct vm_area_struct *vma)
 	/*
 	 * If the inode was latched into forced direct IO after a remote-modify
 	 * notification, a mapping needs the page cache, so revert to caching
-	 * mode.  Revert without the inode lock or wb_inval_rwsem: ->mmap runs
+	 * mode.  Revert without the inode lock or IO range lock: ->mmap runs
 	 * under mmap_lock and the buffered write path holds both across a fault
 	 * on the user buffer (which takes mmap_lock), so taking either here
 	 * would invert lock order (ABBA).  Clearing the latch and dropping the
@@ -3980,27 +4044,11 @@ void fuse_init_file_inode(struct inode *inode, unsigned int flags)
 	INIT_LIST_HEAD(&fi->write_files);
 	INIT_LIST_HEAD(&fi->queued_writes);
 	fuse_dlm_cache_init(fi);
+	fuse_range_lock_tree_init(fi);
 	fi->writectr = 0;
 	fi->iocachectr = 0;
 	init_waitqueue_head(&fi->page_waitq);
 	init_waitqueue_head(&fi->direct_io_waitq);
-	/*
-	 * Coherency gate for the forced-direct-IO feature; only writeback+dlm
-	 * regular files need it.  A percpu_rw_semaphore embeds per-CPU state,
-	 * so allocate it out of line and only when the mount can use it rather
-	 * than paying it on every inode.  On failure leave it NULL: the gate
-	 * stays inactive (best-effort invalidate) and the inode is still usable.
-	 */
-	fi->wb_inval_rwsem = NULL;
-	if (fc->writeback_cache && fc->dlm) {
-		struct percpu_rw_semaphore *sem = kmalloc(sizeof(*sem), GFP_KERNEL);
-
-		if (sem && percpu_init_rwsem(sem)) {
-			kfree(sem);
-			sem = NULL;
-		}
-		fi->wb_inval_rwsem = sem;
-	}
 	fi->notify_stamp = jiffies;
 	fi->notify_interval_ewma = FUSE_NOTIFY_EWMA_SEED << FUSE_NOTIFY_EWMA_SHIFT;
 
diff --git a/fs/fuse/fuse_dlm_cache.c b/fs/fuse/fuse_dlm_cache.c
index bc6dbae2d5aeb0..9a76a26d06358f 100644
--- a/fs/fuse/fuse_dlm_cache.c
+++ b/fs/fuse/fuse_dlm_cache.c
@@ -63,7 +63,6 @@ int fuse_dlm_cache_init(struct fuse_inode *inode)
 
 	init_rwsem(&cache->lock);
 	cache->ranges = RB_ROOT_CACHED;
-	cache->revoke_gen = 0;
 
 	return 0;
 }
@@ -85,7 +84,6 @@ void fuse_dlm_cache_release_locks(struct fuse_inode *inode)
 
 	/* Release all locks */
 	down_write(&cache->lock);
-	WRITE_ONCE(cache->revoke_gen, cache->revoke_gen + 1);
 	while ((node = rb_first_cached(&cache->ranges)) != NULL) {
 		range = rb_entry(node, struct fuse_dlm_range, rb);
 		fuse_page_it_remove(range, &cache->ranges);
@@ -168,13 +166,11 @@ static void fuse_dlm_try_merge(struct fuse_dlm_cache *cache, uint64_t start,
 }
 
 /**
- * __fuse_dlm_lock_range - Lock a range of pages
+ * fuse_dlm_lock_range - Lock a range of pages
  * @cache: The page cache
  * @start: Start page offset
  * @end: End page offset
  * @mode: Lock mode (read or write)
- * @genp: If non-NULL, the revocation generation sampled before the grant
- *        was requested; recording fails with -EAGAIN if it has moved
  *
  * Add a locked range on the specified range of pages.
  * If parts of the range are already locked, only add the remaining parts.
@@ -185,9 +181,8 @@ static void fuse_dlm_try_merge(struct fuse_dlm_cache *cache, uint64_t start,
  *
  * Return: 0 on success, negative error code on failure
  */
-static int __fuse_dlm_lock_range(struct fuse_inode *inode, uint64_t start,
-				 uint64_t end, enum fuse_page_lock_mode mode,
-				 const uint64_t *genp)
+int fuse_dlm_lock_range(struct fuse_inode *inode, uint64_t start,
+			uint64_t end, enum fuse_page_lock_mode mode)
 {
 	struct fuse_dlm_cache *cache = &inode->dlm_locked_areas;
 	struct fuse_dlm_range *range, *new_range, *next;
@@ -207,17 +202,6 @@ static int __fuse_dlm_lock_range(struct fuse_inode *inode, uint64_t start,
 
 	down_write(&cache->lock);
 
-	/*
-	 * A revoke was processed after @genp was sampled; the grant this
-	 * record carries may be the very one it targeted (a revoke of a
-	 * not-yet-recorded grant removes nothing and would never be
-	 * retried).  Refuse, the caller re-requests.
-	 */
-	if (genp && cache->revoke_gen != *genp) {
-		up_write(&cache->lock);
-		return -EAGAIN;
-	}
-
 	/* Find all ranges that overlap with [start, end] */
 	range = fuse_page_it_iter_first(&cache->ranges, start, end);
 	while (range) {
@@ -313,35 +297,6 @@ static int __fuse_dlm_lock_range(struct fuse_inode *inode, uint64_t start,
 	return ret;
 }
 
-int fuse_dlm_lock_range(struct fuse_inode *inode, uint64_t start,
-			uint64_t end, enum fuse_page_lock_mode mode)
-{
-	return __fuse_dlm_lock_range(inode, start, end, mode, NULL);
-}
-
-int fuse_dlm_lock_range_gen(struct fuse_inode *inode, uint64_t start,
-			    uint64_t end, enum fuse_page_lock_mode mode,
-			    uint64_t gen)
-{
-	return __fuse_dlm_lock_range(inode, start, end, mode, &gen);
-}
-
-/**
- * fuse_dlm_revoke_gen - sample the revocation generation
- * @inode: the fuse inode
- *
- * Sampled before a FUSE_DLM_WB_LOCK request leaves the client.  The
- * reply and a NOTIFY revoke can be serviced on different threads, so a
- * revoke may be processed between the reply arriving and its grant
- * being recorded.  fuse_dlm_lock_range_gen() re-checks the generation
- * under the cache lock and refuses to record a grant such a revoke may
- * have already killed.
- */
-uint64_t fuse_dlm_revoke_gen(struct fuse_inode *inode)
-{
-	return READ_ONCE(inode->dlm_locked_areas.revoke_gen);
-}
-
 /**
  * fuse_dlm_punch_hole - Punch a hole in a locked range
  * @cache: The page cache
@@ -435,14 +390,6 @@ int fuse_dlm_unlock_range(struct fuse_inode *inode,
 
 	down_write(&cache->lock);
 
-	/*
-	 * Unconditional, even when nothing overlaps: the revoke racing
-	 * with an in-flight grant finds an empty tree precisely because
-	 * the grant is not recorded yet, and the bump is what makes the
-	 * recording side notice (see fuse_dlm_lock_range_gen()).
-	 */
-	WRITE_ONCE(cache->revoke_gen, cache->revoke_gen + 1);
-
 	/* Find all ranges that overlap with [start, end] */
 	range = fuse_page_it_iter_first(&cache->ranges, start, end);
 	while (range) {
@@ -620,12 +567,62 @@ bool fuse_dlm_lock_is_held(struct fuse_inode *fi, loff_t offset,
 	return fuse_dlm_range_is_locked(fi, offset & PAGE_MASK, end, mode);
 }
 
+/* Context for a fuse_get_dlm_lock() request, embedding struct
+ * fuse_args as required by the request API. */
+struct fuse_dlm_lock_args {
+	struct fuse_args args;
+	struct fuse_inode *fi;
+	struct fuse_range_lock *rlock;
+};
+
+/*
+ * fuse_get_dlm_lock_complete - move the caller's range lock to READY
+ *
+ * Called from fuse_request_end() on the thread processing the reply,
+ * before it wakes the thread blocked in fuse_get_dlm_lock() or moves on
+ * to the next message (e.g. an invalidate notification for the same
+ * inode). This is what keeps the range invisible to invalidation for no
+ * longer than necessary: were this instead left to run on the (possibly
+ * not-yet-scheduled) requester thread after waking, an invalidate queued
+ * right behind this reply could be processed first and race ahead of a
+ * grant that, logically, already arrived first.
+ *
+ * Does not re-validate the granted range against what was requested --
+ * fuse_get_dlm_lock() still does that itself after being woken.
+ */
+static void fuse_get_dlm_lock_complete(struct fuse_mount *fm,
+				       struct fuse_args *args, int error)
+{
+	struct fuse_dlm_lock_args *dargs =
+		container_of(args, struct fuse_dlm_lock_args, args);
+
+	if (!dargs->rlock)
+		return;
+
+	/*
+	 * error is 0 (a grant) or -ENOSYS (the server has no DLM, so the
+	 * range is treated as covered) here: either way fuse_get_dlm_lock()
+	 * goes on to consider the range usable. Any other error fails the
+	 * IO instead, and the caller releases the still-INIT rlock directly
+	 * without ever touching the page cache under it -- leave it alone.
+	 */
+	if (error && error != -ENOSYS)
+		return;
+
+	fuse_range_lock_mark_ready(dargs->fi, dargs->rlock);
+}
+
 /**
  * fuse_get_dlm_lock - request a dlm lock from the fuse server
  * @file:   the file being accessed
  * @offset: byte offset into the file (need not be page-aligned)
  * @length: length of the region in bytes (need not be page-aligned)
  * @mode:   FUSE_PAGE_LOCK_READ or FUSE_PAGE_LOCK_WRITE
+ * @rlock:  optional IO range lock reserved by the caller in INIT state;
+ *	moved to READY as part of processing a reply that leaves the range
+ *	covered, before this function's caller is even woken up -- see
+ *	fuse_get_dlm_lock_complete() and the declaration in
+ *	fuse_dlm_cache.h.
  *
  * Return: 0 when the range is covered by a recorded grant on return,
  * FUSE_DLM_GRANT_UNRECORDED when the server granted the lock but
@@ -635,7 +632,8 @@ bool fuse_dlm_lock_is_held(struct fuse_inode *fi, loff_t offset,
  * they would spin.
  */
 int fuse_get_dlm_lock(struct file *file, loff_t offset,
-		      size_t length, enum fuse_page_lock_mode mode)
+		      size_t length, enum fuse_page_lock_mode mode,
+		      struct fuse_range_lock *rlock)
 {
 	struct fuse_file *ff = file->private_data;
 	struct inode *inode = file_inode(file);
@@ -643,17 +641,16 @@ int fuse_get_dlm_lock(struct file *file, loff_t offset,
 	struct fuse_inode *fi = get_fuse_inode(inode);
 	struct fuse_mount *fm = ff->fm;
 
-	FUSE_ARGS(args);
+	struct fuse_dlm_lock_args dargs = { .fi = fi, .rlock = rlock };
+	struct fuse_args *args = &dargs.args;
 	struct fuse_dlm_lock_in inarg;
 	struct fuse_dlm_lock_out outarg;
-	uint64_t gen;
 	int err;
 
 	/* An empty range needs no lock. */
 	if (!length)
 		return 0;
 
-restart:
 	/* note that this can be run from different processes
 	 * at the same time. It is intentionally not protected
 	 * since a DLM implementation in the FUSE server should take care
@@ -661,18 +658,11 @@ int fuse_get_dlm_lock(struct file *file, loff_t offset,
 	 * The early exit uses the same helper the callers re-validate
 	 * with, so this check and a later fuse_dlm_lock_is_held() can
 	 * never disagree about what counts as covered. */
-	if (fuse_dlm_lock_is_held(fi, offset, length, mode))
+	if (fuse_dlm_lock_is_held(fi, offset, length, mode)) {
+		if (rlock)
+			fuse_range_lock_mark_ready(fi, rlock);
 		return 0; /* we already have this area locked */
-
-	/*
-	 * Sample the revocation generation before the request leaves.
-	 * The reply and a NOTIFY revoke are serviced on different
-	 * threads, so a revoke aimed at the grant this request returns
-	 * can be processed before the grant is recorded below --
-	 * recording it anyway would resurrect a dead grant that no later
-	 * NOTIFY will ever remove.
-	 */
-	gen = fuse_dlm_revoke_gen(fi);
+	}
 
 	memset(&inarg, 0, sizeof(inarg));
 	inarg.fh = ff->fh;
@@ -685,15 +675,16 @@ int fuse_get_dlm_lock(struct file *file, loff_t offset,
 	inarg.type = (mode == FUSE_PAGE_LOCK_WRITE) ?
 		FUSE_DLM_LOCK_WRITE : FUSE_DLM_LOCK_READ;
 
-	args.opcode = FUSE_DLM_WB_LOCK;
-	args.nodeid = get_node_id(inode);
-	args.in_numargs = 1;
-	args.in_args[0].size = sizeof(inarg);
-	args.in_args[0].value = &inarg;
-	args.out_numargs = 1;
-	args.out_args[0].size = sizeof(outarg);
-	args.out_args[0].value = &outarg;
-	err = fuse_simple_request(fm, &args);
+	args->opcode = FUSE_DLM_WB_LOCK;
+	args->nodeid = get_node_id(inode);
+	args->in_numargs = 1;
+	args->in_args[0].size = sizeof(inarg);
+	args->in_args[0].value = &inarg;
+	args->out_numargs = 1;
+	args->out_args[0].size = sizeof(outarg);
+	args->out_args[0].value = &outarg;
+	args->complete = fuse_get_dlm_lock_complete;
+	err = fuse_simple_request(fm, args);
 	if (err == -ENOSYS) {
 		/* fuse server does not support dlm, save the info */
 		fc->dlm = 0;
@@ -715,20 +706,7 @@ int fuse_get_dlm_lock(struct file *file, loff_t offset,
 	 * The server granted the lock; record it so
 	 * fuse_dlm_lock_is_held() sees it.
 	 */
-	err = fuse_dlm_lock_range_gen(fi, outarg.start, outarg.end, mode, gen);
-	if (err == -EAGAIN) {
-		/*
-		 * A revoke was processed while the request was in flight;
-		 * the grant may already be dead, so re-request instead of
-		 * recording it.  Retry until a grant survives long enough to
-		 * be recorded: giving up here would hand the caller an error
-		 * for a range no one else holds, and the write path turns
-		 * that into a failed write.  Each pass makes a fresh server
-		 * round trip, so a revoke storm throttles this loop rather
-		 * than spinning it.
-		 */
-		goto restart;
-	}
+	err = fuse_dlm_lock_range(fi, outarg.start, outarg.end, mode);
 
 	/*
 	 * A failure to record (small-allocation -ENOMEM) does not undo
@@ -736,7 +714,8 @@ int fuse_get_dlm_lock(struct file *file, loff_t offset,
 	 * bookkeeping is missing.  Report that as
 	 * FUSE_DLM_GRANT_UNRECORDED so callers neither fail an IO that
 	 * is actually covered nor keep re-requesting a grant that will
-	 * not become visible.
+	 * not become visible.  (rlock, if any, was already moved to READY
+	 * by fuse_get_dlm_lock_complete() while processing the reply.)
 	 */
 	if (err)
 		return FUSE_DLM_GRANT_UNRECORDED;
diff --git a/fs/fuse/fuse_dlm_cache.h b/fs/fuse/fuse_dlm_cache.h
index 30fdbb26bd3daf..af2de115bf2d56 100644
--- a/fs/fuse/fuse_dlm_cache.h
+++ b/fs/fuse/fuse_dlm_cache.h
@@ -13,6 +13,7 @@
 
 
 struct fuse_inode;
+struct fuse_range_lock;
 
 /* Lock modes for page ranges */
 enum fuse_page_lock_mode { FUSE_PAGE_LOCK_READ, FUSE_PAGE_LOCK_WRITE };
@@ -32,13 +33,6 @@ struct fuse_dlm_cache {
 	struct rw_semaphore lock;
 	/* Interval tree of locked ranges */
 	struct rb_root_cached ranges;
-	/*
-	 * Bumped under @lock by every revocation
-	 * (fuse_dlm_unlock_range(), fuse_dlm_cache_release_locks());
-	 * lets fuse_get_dlm_lock() order recording a reply's grant
-	 * against revokes processed while the reply was in flight.
-	 */
-	uint64_t revoke_gen;
 };
 
 /* Initialize a page cache lock manager */
@@ -51,14 +45,6 @@ void fuse_dlm_cache_release_locks(struct fuse_inode *inode);
 int fuse_dlm_lock_range(struct fuse_inode *inode, uint64_t start,
 			uint64_t end, enum fuse_page_lock_mode mode);
 
-/* As above, but refuse (-EAGAIN) if a revoke ran since @gen was sampled */
-int fuse_dlm_lock_range_gen(struct fuse_inode *inode, uint64_t start,
-			    uint64_t end, enum fuse_page_lock_mode mode,
-			    uint64_t gen);
-
-/* Sample the revocation generation (see fuse_dlm_lock_range_gen()) */
-uint64_t fuse_dlm_revoke_gen(struct fuse_inode *inode);
-
 /* Unlock a range of pages */
 int fuse_dlm_unlock_range(struct fuse_inode *inode, uint64_t start,
 			  uint64_t end);
@@ -74,8 +60,22 @@ bool fuse_dlm_lock_is_held(struct fuse_inode *inode, loff_t offset,
 /* Is any part of the file held for write? */
 bool fuse_dlm_write_grant_exists(struct fuse_inode *inode);
 
-/* This is the interface to the filesystem */
+/*
+ * This is the interface to the filesystem.
+ *
+ * @rlock: optional IO range lock, previously reserved by the caller in
+ * INIT state via fuse_range_lock_acquire_init(), covering (at least)
+ * [offset, offset + length - 1].  When non-NULL, it is moved to READY
+ * as part of processing a reply that leaves the range covered -- i.e.
+ * the already-held fast path, a granted lock, or the server having no
+ * DLM at all -- as soon as that outcome is known, which for a request
+ * that reaches the server is before this function's caller is even
+ * woken up (see fuse_get_dlm_lock_complete() in fuse_dlm_cache.c).
+ * Left at INIT on a hard error, since the caller will not touch the
+ * page cache and releases it directly.
+ */
 int fuse_get_dlm_lock(struct file *file, loff_t offset,
-		      size_t length, enum fuse_page_lock_mode mode);
+		      size_t length, enum fuse_page_lock_mode mode,
+		      struct fuse_range_lock *rlock);
 
 #endif /* _FS_FUSE_DLM_CACHE_H */
diff --git a/fs/fuse/fuse_i.h b/fs/fuse/fuse_i.h
index 31580b7834e296..36c5ccdc9a23d7 100644
--- a/fs/fuse/fuse_i.h
+++ b/fs/fuse/fuse_i.h
@@ -33,6 +33,7 @@
 #include <linux/refcount.h>
 #include <linux/user_namespace.h>
 #include "fuse_dlm_cache.h"
+#include "fuse_range_lock.h"
 
 /** Default max number of pages that can be used in a single read request */
 #define FUSE_DEFAULT_MAX_PAGES_PER_REQ 32
@@ -202,21 +203,6 @@ struct fuse_inode {
 			/* dlm locked areas we have sent lock requests for */
 			struct fuse_dlm_cache dlm_locked_areas;
 
-			/*
-			 * Serializes buffered-write page-cache dirtying against
-			 * the forced-direct-IO latch transition driven by
-			 * NOTIFY_INVAL_INODE (fuse_reverse_inval_inode()), which
-			 * may be delivered by the same server thread that still
-			 * owes a reply to an in-flight write holding the inode
-			 * lock.  The buffered writer holds this for read around
-			 * the dirtying and re-checks the latch under it; the
-			 * NOTIFY latch site takes it for write (trylock, never
-			 * blocking) around its page-cache invalidate + latch set.
-			 * Only regular files initialise it -- it shares storage
-			 * with the readdir-cache union arm.
-			 */
-			struct percpu_rw_semaphore *wb_inval_rwsem;
-
 			/*
 			 * Rate of FUSE_NOTIFY_INVAL_INODE data invalidations
 			 * for this whole file: notify_stamp is the jiffies of
@@ -229,6 +215,15 @@ struct fuse_inode {
 			 */
 			unsigned long notify_stamp;
 			unsigned int notify_interval_ewma;
+
+			/*
+			 * Local byte-range lock tree, used to serialize
+			 * concurrent cached reads/writes that overlap, and to
+			 * let range-scoped invalidation (BRL/attr invalidation
+			 * notifications) block only on IO overlapping the
+			 * range being invalidated.
+			 */
+			struct fuse_range_lock_tree io_range_lock;
 		};
 
 		/* readdir cache (directory only) */
@@ -412,6 +407,22 @@ struct fuse_args {
 	struct fuse_in_arg in_args[4];
 	struct fuse_arg out_args[2];
 	void (*end)(struct fuse_mount *fm, struct fuse_args *args, int error);
+	/*
+	 * Called from fuse_request_end(), synchronously, on the thread
+	 * processing the reply -- before that thread wakes a requester
+	 * blocked in request_wait_answer(), runs any FR_BACKGROUND
+	 * completion, or invokes 'end' above. Unlike 'end' (gated on
+	 * FR_ASYNC, and which for background requests runs after the
+	 * request has already been fully torn down), 'complete' runs for
+	 * every request that reaches fuse_request_end(), synchronous or
+	 * not, letting a caller do work that must be visible before the
+	 * requester resumes or the reply-processing thread moves on to the
+	 * next message -- e.g. moving a range lock from INIT to READY as
+	 * part of processing a grant reply, instead of leaving that race
+	 * window open until the (possibly much later, descheduled)
+	 * requester thread gets to run.
+	 */
+	void (*complete)(struct fuse_mount *fm, struct fuse_args *args, int error);
 	/* Used for kvec iter backed by vmalloc address */
 	void *vmap_base;
 };
diff --git a/fs/fuse/fuse_range_lock.c b/fs/fuse/fuse_range_lock.c
new file mode 100644
index 00000000000000..c650f4d28d58d2
--- /dev/null
+++ b/fs/fuse/fuse_range_lock.c
@@ -0,0 +1,263 @@
+// SPDX-License-Identifier: GPL-2.0-only
+/*
+ * FUSE local IO range lock implementation
+ *
+ * See fuse_range_lock.h for a description of what this is used for.
+ */
+#include "fuse_i.h"
+#include "fuse_range_lock.h"
+
+#include <linux/interval_tree_generic.h>
+
+/* Interval tree definitions for IO range locks */
+static inline uint64_t fuse_range_lock_start(struct fuse_range_lock *lock)
+{
+	return lock->start;
+}
+
+static inline uint64_t fuse_range_lock_last(struct fuse_range_lock *lock)
+{
+	return lock->end;
+}
+
+INTERVAL_TREE_DEFINE(struct fuse_range_lock, rb, uint64_t, __subtree_end,
+		   fuse_range_lock_start, fuse_range_lock_last, static,
+		   fuse_range_it);
+
+/**
+ * fuse_range_lock_tree_init - Initialize the range lock manager
+ * @inode: The fuse inode whose range lock tree to initialize
+ */
+void fuse_range_lock_tree_init(struct fuse_inode *inode)
+{
+	struct fuse_range_lock_tree *tree = &inode->io_range_lock;
+
+	spin_lock_init(&tree->lock);
+	tree->root = RB_ROOT_CACHED;
+	init_waitqueue_head(&tree->waitq);
+}
+
+/**
+ * fuse_range_conflicts - Test @lock's range against currently held ranges
+ * @tree: The range lock tree
+ * @lock: The range lock being tested (not yet inserted, or already held)
+ * @ready_only: If true, ignore existing ranges still in INIT state
+ *
+ * A conflict occurs whenever an overlapping range exists (other than
+ * @lock itself) and either @lock or that range is a WRITE lock (READ
+ * ranges may overlap each other freely). When @ready_only is set, a
+ * range still in INIT state (reserved, not yet touching the page cache)
+ * is not considered -- see fuse_range_lock_acquire_ready().
+ *
+ * Caller holds @tree->lock.
+ *
+ * Return: true if @lock's range conflicts with an existing held range.
+ */
+static bool fuse_range_conflicts(struct fuse_range_lock_tree *tree,
+				struct fuse_range_lock *lock,
+				bool ready_only)
+{
+	struct fuse_range_lock *cur;
+
+	cur = fuse_range_it_iter_first(&tree->root, lock->start, lock->end);
+	while (cur) {
+		if (cur != lock &&
+		    (!ready_only || cur->state == FUSE_RANGE_LOCK_READY) &&
+		    (lock->mode == FUSE_RANGE_LOCK_WRITE ||
+		     cur->mode == FUSE_RANGE_LOCK_WRITE))
+			return true;
+		cur = fuse_range_it_iter_next(cur, lock->start, lock->end);
+	}
+
+	return false;
+}
+
+/**
+ * fuse_range_try_lock_init - Try to insert @lock into @tree in INIT state
+ * @tree: The range lock tree
+ * @lock: The range lock to try to acquire
+ *
+ * Conflict tested against every existing range regardless of state, same
+ * as a plain exclusive acquire -- this is what keeps two local IOs on an
+ * overlapping range serialized against each other even while both are
+ * still reserving (INIT), not yet touching the page cache.
+ *
+ * Return: true if @lock was inserted, false if the caller must wait.
+ */
+static bool fuse_range_try_lock_init(struct fuse_range_lock_tree *tree,
+				    struct fuse_range_lock *lock)
+{
+	bool conflict;
+
+	spin_lock(&tree->lock);
+
+	conflict = fuse_range_conflicts(tree, lock, false);
+	if (!conflict) {
+		lock->state = FUSE_RANGE_LOCK_INIT;
+		fuse_range_it_insert(lock, &tree->root);
+	}
+
+	spin_unlock(&tree->lock);
+
+	return !conflict;
+}
+
+/**
+ * fuse_range_try_lock_ready - Try to insert @lock into @tree in READY state
+ * @tree: The range lock tree
+ * @lock: The range lock to try to acquire
+ *
+ * Conflict tested only against existing READY ranges: an overlapping
+ * range still in INIT state is ignored, so this never waits behind a
+ * read/write that has only reserved a range and not yet started
+ * touching the page cache.
+ *
+ * Return: true if @lock was inserted, false if the caller must wait.
+ */
+static bool fuse_range_try_lock_ready(struct fuse_range_lock_tree *tree,
+				     struct fuse_range_lock *lock)
+{
+	bool conflict;
+
+	spin_lock(&tree->lock);
+
+	conflict = fuse_range_conflicts(tree, lock, true);
+	if (!conflict) {
+		lock->state = FUSE_RANGE_LOCK_READY;
+		fuse_range_it_insert(lock, &tree->root);
+	}
+
+	spin_unlock(&tree->lock);
+
+	return !conflict;
+}
+
+/**
+ * fuse_range_try_mark_ready - Try to move an already-held @lock to READY
+ * @tree: The range lock tree
+ * @lock: The (already inserted) range lock to move to READY state
+ *
+ * Return: true if @lock is now READY, false if the caller must wait for
+ * a conflicting READY range to be released.
+ */
+static bool fuse_range_try_mark_ready(struct fuse_range_lock_tree *tree,
+				     struct fuse_range_lock *lock)
+{
+	bool conflict;
+
+	spin_lock(&tree->lock);
+
+	conflict = fuse_range_conflicts(tree, lock, true);
+	if (!conflict)
+		lock->state = FUSE_RANGE_LOCK_READY;
+
+	spin_unlock(&tree->lock);
+
+	return !conflict;
+}
+
+/**
+ * fuse_range_lock_acquire_init - Reserve a byte range lock in INIT state
+ * @inode: The fuse inode
+ * @lock: Caller-allocated storage for the lock (e.g. on the stack)
+ * @start: Start byte offset (inclusive)
+ * @end: End byte offset (inclusive)
+ * @mode: FUSE_RANGE_LOCK_READ or FUSE_RANGE_LOCK_WRITE
+ *
+ * Blocks until [start, end] can be reserved in the requested mode
+ * without conflicting with any other currently held, overlapping range.
+ */
+void fuse_range_lock_acquire_init(struct fuse_inode *inode,
+				 struct fuse_range_lock *lock,
+				 uint64_t start, uint64_t end,
+				 enum fuse_range_lock_mode mode)
+{
+	struct fuse_range_lock_tree *tree = &inode->io_range_lock;
+
+	lock->start = start;
+	lock->end = end;
+	lock->mode = mode;
+
+	wait_event(tree->waitq, fuse_range_try_lock_init(tree, lock));
+}
+
+/**
+ * fuse_range_lock_mark_ready - Move a reserved range lock to READY state
+ * @inode: The fuse inode
+ * @lock: The range lock previously passed to fuse_range_lock_acquire_init()
+ *
+ * Blocks until no other currently held, overlapping READY range
+ * conflicts with @lock.
+ */
+void fuse_range_lock_mark_ready(struct fuse_inode *inode,
+			       struct fuse_range_lock *lock)
+{
+	struct fuse_range_lock_tree *tree = &inode->io_range_lock;
+
+	wait_event(tree->waitq, fuse_range_try_mark_ready(tree, lock));
+}
+
+/**
+ * fuse_range_lock_mark_init - Move a READY range lock back to INIT state
+ * @inode: The fuse inode
+ * @lock: The range lock previously moved to READY state
+ *
+ * Never blocks. Wakes waiters, since an invalidation may be waiting on
+ * @lock's (until now READY) range and can now proceed around it.
+ */
+void fuse_range_lock_mark_init(struct fuse_inode *inode,
+			      struct fuse_range_lock *lock)
+{
+	struct fuse_range_lock_tree *tree = &inode->io_range_lock;
+
+	spin_lock(&tree->lock);
+	lock->state = FUSE_RANGE_LOCK_INIT;
+	spin_unlock(&tree->lock);
+
+	wake_up_all(&tree->waitq);
+}
+
+/**
+ * fuse_range_lock_acquire_ready - Acquire a byte range lock in READY state
+ * @inode: The fuse inode
+ * @lock: Caller-allocated storage for the lock (e.g. on the stack)
+ * @start: Start byte offset (inclusive)
+ * @end: End byte offset (inclusive)
+ * @mode: FUSE_RANGE_LOCK_READ or FUSE_RANGE_LOCK_WRITE
+ *
+ * Blocks until [start, end] can be locked in the requested mode without
+ * conflicting with any other currently held READY, overlapping range.
+ * An overlapping range still in INIT state does not block this.
+ */
+void fuse_range_lock_acquire_ready(struct fuse_inode *inode,
+				  struct fuse_range_lock *lock,
+				  uint64_t start, uint64_t end,
+				  enum fuse_range_lock_mode mode)
+{
+	struct fuse_range_lock_tree *tree = &inode->io_range_lock;
+
+	lock->start = start;
+	lock->end = end;
+	lock->mode = mode;
+
+	wait_event(tree->waitq, fuse_range_try_lock_ready(tree, lock));
+}
+
+/**
+ * fuse_range_lock_release - Release a previously acquired range lock
+ * @inode: The fuse inode
+ * @lock: The range lock previously passed to
+ *	fuse_range_lock_acquire_init() or fuse_range_lock_acquire_ready()
+ */
+void fuse_range_lock_release(struct fuse_inode *inode,
+			    struct fuse_range_lock *lock)
+{
+	struct fuse_range_lock_tree *tree = &inode->io_range_lock;
+
+	spin_lock(&tree->lock);
+	fuse_range_it_remove(lock, &tree->root);
+	spin_unlock(&tree->lock);
+
+	/* Wake everyone; conflicting waiters will simply re-check and sleep. */
+	wake_up_all(&tree->waitq);
+}
diff --git a/fs/fuse/fuse_range_lock.h b/fs/fuse/fuse_range_lock.h
new file mode 100644
index 00000000000000..ae5ee88416f549
--- /dev/null
+++ b/fs/fuse/fuse_range_lock.h
@@ -0,0 +1,181 @@
+/* SPDX-License-Identifier: GPL-2.0-only */
+/*
+ * FUSE local IO range lock
+ *
+ * Interval-tree based byte-range lock, embedded in each fuse_inode,
+ * used to:
+ *
+ *  - Serialize concurrent cached reads/writes that touch overlapping
+ *    byte ranges of the same file, while letting non-overlapping IO
+ *    proceed concurrently.
+ *
+ *  - Let attribute/BRL invalidation (fuse_reverse_inval_inode(),
+ *    truncate in fuse_do_setattr()) block only on in-progress IO that
+ *    overlaps with the range being invalidated, instead of serializing
+ *    with all IO on the inode.
+ *
+ * This is modeled after the interval tree used by fuse_dlm_cache, but
+ * unlike the DLM cache (which only records ranges the client has been
+ * granted a lock for by the server), this tree tracks in-progress local
+ * holders and provides blocking acquire/release semantics.
+ *
+ * Each held range additionally carries a two-state lifecycle:
+ *
+ *  - INIT: a read/write has reserved the range (so another overlapping
+ *    local reader/writer queues behind it, same as before) but has not
+ *    yet touched the page cache -- typically while a fuse_get_dlm_lock()
+ *    request to the server is in flight.  An INIT range is invisible to
+ *    invalidation: fuse_range_lock_acquire_ready() ignores it, so a
+ *    NOTIFY invalidate with an overlapping range never waits on the
+ *    (unbounded, cluster round trip) DLM request.
+ *
+ *  - READY: the holder is about to, or is actively, touching the page
+ *    cache.  Fully exclusive against any overlapping range per the usual
+ *    READ/WRITE compatibility rules, including against other READY
+ *    holders and, unlike INIT, against invalidation.
+ *
+ * A read/write reserves its range with fuse_range_lock_acquire_init(),
+ * does whatever DLM work it needs, then calls fuse_range_lock_mark_ready()
+ * once it is about to touch the page cache; that call itself blocks until
+ * any overlapping READY holder (e.g. an in-progress invalidation that
+ * raced ahead of it) is done.  fuse_range_lock_mark_init() is available
+ * to move a READY range back to INIT, e.g. if a caller must redo DLM work
+ * without letting that block a fresh invalidate on the same range in the
+ * meantime -- it never blocks.  Invalidation instead calls
+ * fuse_range_lock_acquire_ready(), which never waits on an INIT range.
+ * Both sides release with fuse_range_lock_release().
+ *
+ * This locking exists to protect against DLM-covered writeback IO and
+ * invalidation racing on the same range, so callers should only use it
+ * when both the writeback cache and DLM are in use for the connection;
+ * see individual function comments below.
+ */
+
+#ifndef _FS_FUSE_RANGE_LOCK_H
+#define _FS_FUSE_RANGE_LOCK_H
+
+#include <linux/types.h>
+#include <linux/rbtree.h>
+#include <linux/spinlock.h>
+#include <linux/wait.h>
+
+struct fuse_inode;
+
+/* Lock modes for IO range locks */
+enum fuse_range_lock_mode {
+	/* Shared: compatible with other READ holders on overlapping ranges */
+	FUSE_RANGE_LOCK_READ,
+	/* Exclusive: incompatible with any overlapping READ or WRITE holder */
+	FUSE_RANGE_LOCK_WRITE,
+};
+
+/* Lifecycle state of a held range lock; see the file comment above. */
+enum fuse_range_lock_state {
+	/* Reserved, not yet touching the page cache; invisible to
+	 * invalidation. */
+	FUSE_RANGE_LOCK_INIT,
+	/* Actively about to touch, or touching, the page cache; fully
+	 * exclusive, including against invalidation. */
+	FUSE_RANGE_LOCK_READY,
+};
+
+/* Per-inode range lock manager */
+struct fuse_range_lock_tree {
+	/* Protects the interval tree below */
+	spinlock_t lock;
+	/* Interval tree of currently held ranges */
+	struct rb_root_cached root;
+	/* Waiters for a range to become available */
+	wait_queue_head_t waitq;
+};
+
+/*
+ * A single held range lock. The caller owns the storage (typically on
+ * the stack, for the duration of one IO/invalidation call) and passes
+ * it to fuse_range_lock_acquire_init()/fuse_range_lock_acquire_ready()
+ * and to fuse_range_lock_release().
+ */
+struct fuse_range_lock {
+	/* Interval tree node */
+	struct rb_node rb;
+	/* Start byte offset (inclusive) */
+	uint64_t start;
+	/* End byte offset (inclusive) */
+	uint64_t end;
+	/* Subtree end value for interval tree */
+	uint64_t __subtree_end;
+	/* Lock mode */
+	enum fuse_range_lock_mode mode;
+	/* Lifecycle state; see the file comment above */
+	enum fuse_range_lock_state state;
+};
+
+/* Initialize the range lock manager embedded in a fuse_inode */
+void fuse_range_lock_tree_init(struct fuse_inode *inode);
+
+/*
+ * Reserve a range lock on [start, end] (inclusive byte offsets) in the
+ * given mode, in INIT state. Blocks until the range can be reserved
+ * without conflicting with any other currently held, overlapping range
+ * (INIT or READY), same as a plain exclusive acquire. Invisible to
+ * fuse_range_lock_acquire_ready() until fuse_range_lock_mark_ready()
+ * is called.
+ *
+ * Caller must only call this when both the writeback cache and DLM are
+ * in use for @inode's connection; see the file comment above.
+ */
+void fuse_range_lock_acquire_init(struct fuse_inode *inode,
+				 struct fuse_range_lock *lock,
+				 uint64_t start, uint64_t end,
+				 enum fuse_range_lock_mode mode);
+
+/*
+ * Move a range lock reserved by fuse_range_lock_acquire_init() from
+ * INIT to READY state. Blocks until no other overlapping READY holder
+ * conflicts (e.g. an invalidation that raced ahead while this range was
+ * still INIT).
+ *
+ * Caller must only call this when both the writeback cache and DLM are
+ * in use for @inode's connection; see the file comment above.
+ */
+void fuse_range_lock_mark_ready(struct fuse_inode *inode,
+			       struct fuse_range_lock *lock);
+
+/*
+ * Move a range lock back from READY to INIT state, e.g. because the
+ * holder must redo some DLM work before it can touch the page cache
+ * again. Never blocks.
+ *
+ * Caller must only call this when both the writeback cache and DLM are
+ * in use for @inode's connection; see the file comment above.
+ */
+void fuse_range_lock_mark_init(struct fuse_inode *inode,
+			      struct fuse_range_lock *lock);
+
+/*
+ * Acquire a range lock on [start, end] (inclusive byte offsets) in the
+ * given mode, directly in READY state. Blocks until the range can be
+ * locked without conflicting with any other currently held READY,
+ * overlapping range; an overlapping INIT range is ignored. Used by
+ * invalidation, which must not wait on a read/write that has only
+ * reserved a range and not yet started touching the page cache.
+ *
+ * Caller must only call this when both the writeback cache and DLM are
+ * in use for @inode's connection; see the file comment above.
+ */
+void fuse_range_lock_acquire_ready(struct fuse_inode *inode,
+				  struct fuse_range_lock *lock,
+				  uint64_t start, uint64_t end,
+				  enum fuse_range_lock_mode mode);
+
+/*
+ * Release a previously acquired range lock and wake any waiters.
+ *
+ * Caller must only call this to release a lock that was actually
+ * acquired via fuse_range_lock_acquire_init()/_ready(); see the file
+ * comment above.
+ */
+void fuse_range_lock_release(struct fuse_inode *inode,
+			    struct fuse_range_lock *lock);
+
+#endif /* _FS_FUSE_RANGE_LOCK_H */
diff --git a/fs/fuse/inode.c b/fs/fuse/inode.c
index d54676b73abf9e..a560b25ff63ec4 100644
--- a/fs/fuse/inode.c
+++ b/fs/fuse/inode.c
@@ -220,23 +220,6 @@ static void fuse_evict_inode(struct inode *inode)
 		WARN_ON(!list_empty(&fi->queued_writes));
 		fuse_dlm_cache_release_locks(fi);
 	}
-
-	/*
-	 * Free the coherency gate here rather than in ->free_inode: that runs
-	 * from an RCU callback, where percpu_free_rwsem() may sleep in
-	 * rcu_sync_dtor() if the write side has not fully quiesced.  No user
-	 * can remain by eviction time: gate readers hold a file reference and
-	 * a concurrent notify holds an inode reference.  wb_inval_rwsem lives
-	 * in the regular-file union arm and is only ever allocated for regular
-	 * files, so gate on S_ISREG (but not fuse_is_bad() -- bad-marked
-	 * regular files still own a gate); a directory's overlapping
-	 * readdir-cache fields must not be misread.
-	 */
-	if (S_ISREG(inode->i_mode) && fi->wb_inval_rwsem) {
-		percpu_free_rwsem(fi->wb_inval_rwsem);
-		kfree(fi->wb_inval_rwsem);
-		fi->wb_inval_rwsem = NULL;
-	}
 }
 
 static int fuse_reconfigure(struct fs_context *fsc)
@@ -934,11 +917,13 @@ static void fuse_notify_invalidate_range(struct inode *inode, pgoff_t start,
 int fuse_reverse_inval_inode(struct fuse_conn *fc, u64 nodeid,
 			     loff_t offset, loff_t len)
 {
-	struct percpu_rw_semaphore *wb_sem = NULL;
+	struct fuse_range_lock rlock;
 	struct fuse_inode *fi;
 	struct inode *inode;
 	pgoff_t pg_start;
 	pgoff_t pg_end;
+	uint64_t lock_start;
+	uint64_t lock_end;
 
 	inode = fuse_ilookup(fc, nodeid, NULL);
 	if (!inode)
@@ -955,7 +940,7 @@ int fuse_reverse_inval_inode(struct fuse_conn *fc, u64 nodeid,
 
 	fi->attr_version = atomic64_inc_return(&fc->attr_version);
 	spin_unlock(&fi->lock);
-	
+
 	if (fc->inval_inode_entries)
 		fuse_invalidate_inode_entry(inode);
 	else if (fc->expire_inode_entries)
@@ -965,6 +950,11 @@ int fuse_reverse_inval_inode(struct fuse_conn *fc, u64 nodeid,
 	forget_all_cached_acls(inode);
 	security_inode_invalidate_secctx(inode);
 	if (offset >= 0) {
+		bool range_locked = S_ISREG(inode->i_mode) &&
+				   fc->writeback_cache && fc->dlm;
+		bool hot = false, has_writer = false, latched = false;
+		bool track_latch;
+
 		pg_start = offset >> PAGE_SHIFT;
 		if (len <= 0)
 			pg_end = -1;
@@ -972,23 +962,22 @@ int fuse_reverse_inval_inode(struct fuse_conn *fc, u64 nodeid,
 			pg_end = (offset + len - 1) >> PAGE_SHIFT;
 
 		/*
-		 * A data invalidation means another (remote) entity is modifying
-		 * the file.  Two things happen here:
+		 * A data invalidation means another (remote) entity is
+		 * modifying the file.  Two things happen here:
 		 *
-		 * 1. Coherency.  Drop the affected page-cache range so no local
-		 *    read returns a folio the remote modify has superseded.  This
-		 *    runs under the write side of the per-inode coherency gate
-		 *    (wb_inval_rwsem), which fences cache-serving buffered reads
-		 *    and buffered writes out for the whole invalidate.  Unlike the
-		 *    old best-effort trylock this BLOCKS -- the notify has
-		 *    priority: percpu_down_write() parks new gate readers, drains
-		 *    in-flight ones, then invalidates.  A blocking writer here is
-		 *    safe only under a server that services request replies on
-		 *    threads other than the one delivering this notify: the write
-		 *    side waits for gate readers to drain, and a cache-miss read
-		 *    holds the read side across its FUSE_READ round-trip.  redfs'
-		 *    dlm server provides that contract; a server that cannot must
-		 *    not enable writeback+dlm.
+		 * 1. Coherency.  Drop the affected page-cache range so no
+		 *    local read returns a folio the remote modify has
+		 *    superseded.  fuse_range_lock_acquire_ready() blocks
+		 *    only on an overlapping range already in READY state --
+		 *    i.e. one actually touching, or about to touch, the page
+		 *    cache -- and ignores one still in INIT state, i.e. a
+		 *    cached read/write that has only reserved the range
+		 *    while its (possibly unbounded, cluster round trip) DLM
+		 *    request is in flight; see fuse_range_lock.h.  This
+		 *    invalidate is thus never parked behind such a request.
+		 *    Once acquired, the range lock fences overlapping cached
+		 *    reads/writes out for the DLM revoke and page drop
+		 *    below.
 		 *
 		 * 2. Latch.  Keep a moving average (fuse_notify_inval_hot(), under
 		 *    fi->lock, updated for every data invalidation) of how fast
@@ -1006,46 +995,33 @@ int fuse_reverse_inval_inode(struct fuse_conn *fc, u64 nodeid,
 		 *    already-latched inodes run out on the usual exits (last
 		 *    writer closes, or mmap).
 		 *
-		 * The gate (and the average) exist only for writeback+dlm regular
-		 * files; elsewhere wb_sem is NULL and the invalidate runs
-		 * unserialized (best-effort), as before.  An mmapped inode
-		 * keeps the gate -- fuse_cache_read_iter() and
-		 * fuse_cache_write_iter() enter it unconditionally and rely
-		 * on the revoke staying fenced -- but is never latched:
-		 * a mapping needs the page cache, and fuse_file_mmap()
-		 * reverts any latch it races with.
+		 * The average and latch exist only for writeback+dlm regular
+		 * files; elsewhere the invalidate just drops the range
+		 * (best-effort), as before.  An mmapped inode is never
+		 * latched: a mapping needs the page cache, and
+		 * fuse_file_mmap() reverts any latch it races with.
 		 */
-		if (S_ISREG(inode->i_mode) && fc->writeback_cache &&
-		    fc->dlm && !FUSE_IS_DAX(inode) &&
-		    !fuse_inode_backing(fi))
-			wb_sem = fi->wb_inval_rwsem;
+		if (range_locked) {
+			lock_start = offset;
+			lock_end = len <= 0 ? ~0ULL : (uint64_t)offset + len - 1;
 
-		if (wb_sem) {
-			bool hot, has_writer, latched = false;
+			fuse_range_lock_acquire_ready(fi, &rlock, lock_start,
+						      lock_end,
+						      FUSE_RANGE_LOCK_WRITE);
+		}
+
+		if (fc->dlm && fc->writeback_cache)
+			fuse_dlm_revoke_inval_range(fi, offset, len);
 
+		track_latch = S_ISREG(inode->i_mode) && fc->writeback_cache &&
+			      fc->dlm && !FUSE_IS_DAX(inode) &&
+			      !fuse_inode_backing(fi);
+		if (track_latch) {
 			spin_lock(&fi->lock);
 			hot = fuse_notify_inval_hot(fi);
 			has_writer = !list_empty(&fi->write_files);
 			spin_unlock(&fi->lock);
 
-			/*
-			 * Priority write side: park new gate readers,
-			 * drain in-flight ones, then invalidate.  Blocks
-			 * (unlike the old trylock) -- see the contract in
-			 * the comment above.
-			 */
-			percpu_down_write(wb_sem);
-
-			/*
-			 * Revoke the DLM lock range under the gate write
-			 * side, atomically with the page drop: gate readers
-			 * re-validate their grant right after entering, and
-			 * a grant that passed that check must stay visible
-			 * for their whole gate hold.
-			 */
-			if (fc->dlm && fc->writeback_cache)
-				fuse_dlm_revoke_inval_range(fi, offset, len);
-
 			if (enable_notify_dio && hot && has_writer &&
 			    !mapping_mapped(inode->i_mapping) &&
 			    !fuse_inode_force_dio(inode)) {
@@ -1056,32 +1032,25 @@ int fuse_reverse_inval_inode(struct fuse_conn *fc, u64 nodeid,
 				}
 				spin_unlock(&fi->lock);
 			}
+		}
 
-			/*
-			 * Latched: drop the whole mapping (dirty folios
-			 * outside the notified range would be invisible to
-			 * the forced direct reads).  Otherwise just the
-			 * notified range.
-			 */
-			if (fuse_inode_force_dio(inode))
-				fuse_notify_invalidate_range(inode, 0, -1);
-			else
-				fuse_notify_invalidate_range(inode, pg_start,
-							     pg_end);
+		/*
+		 * Latched: drop the whole mapping (dirty folios
+		 * outside the notified range would be invisible to
+		 * the forced direct reads).  Otherwise just the
+		 * notified range.
+		 */
+		if (fuse_inode_force_dio(inode))
+			fuse_notify_invalidate_range(inode, 0, -1);
+		else
+			fuse_notify_invalidate_range(inode, pg_start, pg_end);
 
-			percpu_up_write(wb_sem);
+		if (latched)
+			pr_info_ratelimited("FUSE: inode %llu latched to direct IO on invalidation notify storm\n",
+					    nodeid);
 
-			if (latched)
-				pr_info_ratelimited("FUSE: inode %llu latched to direct IO on invalidation notify storm\n",
-						    nodeid);
-		} else {
-			/* No gate on this inode (DAX, backing, non-regular,
-			 * or the gate allocation failed): drop the lock
-			 * range unserialized (best-effort), as before. */
-			if (fc->dlm && fc->writeback_cache)
-				fuse_dlm_revoke_inval_range(fi, offset, len);
-			fuse_notify_invalidate_range(inode, pg_start, pg_end);
-		}
+		if (range_locked)
+			fuse_range_lock_release(fi, &rlock);
 	}
 	iput(inode);
 	return 0;