diff --git a/fs/fuse/dir.c b/fs/fuse/dir.c
index ce398f117ef593..fe4470e39fc095 100644
--- a/fs/fuse/dir.c
+++ b/fs/fuse/dir.c
@@ -723,6 +723,10 @@ static int fuse_create_open(struct mnt_idmap *idmap, struct inode *dir,
 	memset(&inarg, 0, sizeof(inarg));
 	memset(&outentry, 0, sizeof(outentry));
 	inarg.flags = flags;
+
+	/* The kernel owns append positioning; see fuse_send_open() */
+	if (fm->fc->writeback_cache)
+		inarg.flags &= ~O_APPEND;
 	inarg.mode = mode;
 	inarg.umask = current_umask();
 
@@ -785,10 +789,27 @@ static int fuse_create_open(struct mnt_idmap *idmap, struct inode *dir,
 		fi = get_fuse_inode(inode);
 		fuse_sync_release(fi, ff, flags);
 	} else {
-		if (fm->fc->atomic_o_trunc && trunc)
+		if (fm->fc->atomic_o_trunc && trunc) {
+			/*
+			 * Every grant goes with the cache, as on the
+			 * fuse_open() O_TRUNC path: a record left behind
+			 * would keep naming bytes the folios no longer hold.
+			 */
+			if (fm->fc->dlm && fm->fc->writeback_cache)
+				fuse_dlm_cache_release_locks(
+						get_fuse_inode(inode));
 			truncate_pagecache(inode, 0);
-		else if (!(ff->open_flags & FOPEN_KEEP_CACHE))
-			invalidate_inode_pages2(inode->i_mapping);
+		} else if (!(ff->open_flags & FOPEN_KEEP_CACHE)) {
+			/*
+			 * Only when the drop really emptied the mapping, as
+			 * in fuse_open(): a folio that survived still needs
+			 * its record.
+			 */
+			if (fuse_open_drop_cache(inode) &&
+			    fm->fc->dlm && fm->fc->writeback_cache)
+				fuse_dlm_ranges_dropped(get_fuse_inode(inode),
+							0, U64_MAX);
+		}
 	}
 	return err;
 
@@ -2082,34 +2103,24 @@ 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;
-
 			/*
 			 * 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).
+			 * Dropping every grant here does not need a reader or
+			 * writer fenced out: truncate_pagecache() discards the
+			 * folios rather than writing them, and a write racing
+			 * this is a write racing an O_TRUNC open, which has no
+			 * order to preserve.  i_rwsem is held exclusive
+			 * anyway, so no cached write is in progress.
 			 */
-			if (wb_sem)
-				percpu_down_write(wb_sem);
 			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);
 			goto out;
 		}
 		file = NULL;
@@ -2213,23 +2224,43 @@ 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;
-
 		/*
-		 * Revoke and drop under the coherency gate write side; see
-		 * the atomic-O_TRUNC branch above.  i_rwsem is held
-		 * exclusive here as well (setattr), so waiting out gate
-		 * readers cannot deadlock.
+		 * Revoke past the new size and drop what is beyond it; see
+		 * the atomic-O_TRUNC branch above for why this needs nothing
+		 * fenced out.  i_rwsem is held exclusive here as well.
 		 */
-		if (wb_sem)
-			percpu_down_write(wb_sem);
 		if (fc->dlm && fc->writeback_cache)
-			fuse_dlm_unlock_range(fi, outarg.attr.size & PAGE_MASK, -1);
+			fuse_dlm_unlock_range(fi, outarg.attr.size & PAGE_MASK,
+					      U64_MAX);
 
 		truncate_pagecache(inode, outarg.attr.size);
 		invalidate_inode_pages2(mapping);
-		if (wb_sem)
-			percpu_up_write(wb_sem);
+
+		/*
+		 * The cache above the new size is gone, so the ranges
+		 * describing it have nothing left to say.  From the first
+		 * whole page above it: the page holding the new end of the
+		 * file survives the truncate, and so does its record.
+		 */
+		if (fc->dlm && fc->writeback_cache)
+			fuse_dlm_ranges_dropped(fi, PAGE_ALIGN(outarg.attr.size),
+						U64_MAX);
+
+		/*
+		 * invalidate_inode_pages2() emptied the mapping below the new
+		 * size too (laundering anything dirty first, so those bytes
+		 * are on the server).  A revoked range is kept only to make
+		 * writeback take the grant again before sending the folios
+		 * under it, so with those folios gone it describes nothing
+		 * and would sit in the tree unfreed, keeping its neighbours
+		 * from merging.  Only when the drop really emptied it: a busy
+		 * folio that survived still needs its record, and a fault
+		 * populating after the check keeps its page visible to it.
+		 */
+		if (fc->dlm && fc->writeback_cache && outarg.attr.size &&
+		    !filemap_range_has_page(mapping, 0, outarg.attr.size - 1))
+			fuse_dlm_ranges_dropped(fi, 0,
+						PAGE_ALIGN(outarg.attr.size) - 1);
 	}
 
 	clear_bit(FUSE_I_SIZE_UNSTABLE, &fi->state);
diff --git a/fs/fuse/file.c b/fs/fuse/file.c
index 48247876177b5d..24195db1c34755 100644
--- a/fs/fuse/file.c
+++ b/fs/fuse/file.c
@@ -70,6 +70,17 @@ static int fuse_send_open(struct fuse_mount *fm, u64 nodeid,
 	if (!fm->fc->atomic_o_trunc)
 		inarg.flags &= ~O_TRUNC;
 
+	/*
+	 * With the writeback cache the kernel owns append positioning:
+	 * writeback sends FUSE_WRITE with explicit offsets, and a server
+	 * that opens its backing file O_APPEND has pwrite(2) ignore them
+	 * (Linux appends regardless of offset).  Any re-sent or reordered
+	 * run is then placed at EOF: duplicated data and a growing file.
+	 * Do not hand the flag to the server at all.
+	 */
+	if (fm->fc->writeback_cache)
+		inarg.flags &= ~O_APPEND;
+
 	if (fm->fc->handle_killpriv_v2 &&
 	    (inarg.flags & O_TRUNC) && !capable(CAP_FSETID)) {
 		inarg.open_flags |= FUSE_OPEN_KILL_SUIDGID;
@@ -173,6 +184,10 @@ static int fuse_compound_open_getattr(struct fuse_mount *fm, u64 nodeid,
 	if (!fm->fc->atomic_o_trunc)
 		open_in.flags &= ~O_TRUNC;
 
+	/* The kernel owns append positioning; see fuse_send_open() */
+	if (fm->fc->writeback_cache)
+		open_in.flags &= ~O_APPEND;
+
 	if (fm->fc->handle_killpriv_v2 &&
 	    (open_in.flags & O_TRUNC) && !capable(CAP_FSETID))
 		open_in.open_flags |= FUSE_OPEN_KILL_SUIDGID;
@@ -326,6 +341,34 @@ int fuse_finish_open(struct inode *inode, struct file *file)
 	return 0;
 }
 
+/*
+ * Drop the page cache an open that did not get FOPEN_KEEP_CACHE must not keep,
+ * and report whether the mapping came out empty.
+ *
+ * Serialised on the mapping's invalidate lock.  Every opener runs this same
+ * full-mapping walk and invalidate_inode_pages2_range() takes each folio's
+ * lock in turn, so openers racing on one shared file convoy on every folio
+ * and each pays for the whole walk: seven tasks reopening a cached file were
+ * measured at 4.96s apiece, in lockstep, with the server completely idle and
+ * ~86k lock sleeps per task.  Under the lock the first opener empties the
+ * mapping and the rest fall straight back out of mapping_empty().
+ *
+ * Holding it exclusive also fences faults for the duration, which is what
+ * truncate already does across this same walk, and it keeps the emptiness
+ * test from reading a mapping another opener is halfway through.
+ */
+bool fuse_open_drop_cache(struct inode *inode)
+{
+	bool emptied;
+
+	filemap_invalidate_lock(inode->i_mapping);
+	invalidate_inode_pages2(inode->i_mapping);
+	emptied = !filemap_range_has_page(inode->i_mapping, 0, LLONG_MAX);
+	filemap_invalidate_unlock(inode->i_mapping);
+
+	return emptied;
+}
+
 static void fuse_truncate_update_attr(struct inode *inode, struct file *file)
 {
 	struct fuse_conn *fc = get_fuse_conn(inode);
@@ -383,10 +426,31 @@ static int fuse_open(struct inode *inode, struct file *file)
 	if (is_wb_truncate || dax_truncate)
 		fuse_release_nowrite(inode);
 	if (!err) {
-		if (is_truncate)
+		if (is_truncate) {
+			/*
+			 * Every grant goes with the cache, as on the
+			 * fuse_do_setattr() O_TRUNC path: a record left
+			 * behind would keep naming bytes the folios no
+			 * longer hold.  i_rwsem is held exclusive
+			 * (is_wb_truncate), so no cached write is mid-record.
+			 */
+			if (fc->dlm && fc->writeback_cache)
+				fuse_dlm_cache_release_locks(fi);
 			truncate_pagecache(inode, 0);
-		else if (!(ff->open_flags & FOPEN_KEEP_CACHE))
-			invalidate_inode_pages2(inode->i_mapping);
+		} else if (!(ff->open_flags & FOPEN_KEEP_CACHE)) {
+			/*
+			 * Only when the drop really emptied the mapping; a
+			 * folio that survived still needs its record.  This
+			 * open holds no lock against buffered IO, which can
+			 * repopulate the mapping once the invalidate lock is
+			 * dropped -- but such a folio is populated after the
+			 * drop and keeps its page, so a later reader finds
+			 * both the page and the record it needs.
+			 */
+			if (fuse_open_drop_cache(inode) &&
+			    fc->dlm && fc->writeback_cache)
+				fuse_dlm_ranges_dropped(fi, 0, U64_MAX);
+		}
 	}
 	if (dax_truncate)
 		filemap_invalidate_unlock(inode->i_mapping);
@@ -468,10 +532,10 @@ 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
-	 * from aio completion), where blocking on a contended inode lock could
-	 * stall the connection.  Writes were routed direct while latched, so
+	 * served stale once caching mode resumes.  No inode lock: 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
 	 * writer is gone, so no forced-dio writer can race the drop.
 	 */
@@ -1091,21 +1155,213 @@ static void fuse_send_readpages(struct fuse_io_args *ia, struct file *file)
 	fuse_readpages_end(fm, &ap->args, err);
 }
 
+/*
+ * Populate and send one request past the window mm asked for.
+ *
+ * mm keeps readahead state per open file, so several threads reading one
+ * shared file through their own descriptors each carry a separate idea of
+ * the stream.  Only the first of them to reach a hole allocates anything;
+ * the rest find the folios present and return having issued nothing.  The
+ * PG_readahead trigger that should start the next window is cleared by
+ * whichever thread reaches it first, and that thread's state usually
+ * describes a window some other thread built, so page_cache_async_ra()
+ * takes its interleaved path, finds no hole within read_ahead_kb and
+ * issues nothing at all.  The chain dies there and every window after it
+ * costs a synchronous miss with all the readers waiting on it.
+ *
+ * The inode is what those readers actually share, so keep the lookahead
+ * there instead: one of them claims the range past the window and fills
+ * it while the others skip it.  One request deep is enough to leave a
+ * fetch outstanding for the window just sent to be consumed against.
+ *
+ * Called from inside read_pages(), so the invalidate lock is held shared
+ * by the readahead that got us here.  That is what makes adding folios
+ * safe against a concurrent invalidate.
+ */
+static void fuse_readahead_lookahead(struct file *file, struct inode *inode,
+				     pgoff_t start, unsigned int nr)
+{
+	struct fuse_conn *fc = get_fuse_conn(inode);
+	struct fuse_inode *fi = get_fuse_inode(inode);
+	struct address_space *mapping = inode->i_mapping;
+	gfp_t gfp = readahead_gfp_mask(mapping);
+	struct fuse_io_args *ia;
+	struct fuse_args_pages *ap;
+	loff_t size = i_size_read(inode);
+	unsigned int i, added = 0;
+	pgoff_t last;
+	bool claimed;
+
+	if (!nr || !size)
+		return;
+
+	/* Speculative work is not worth queueing behind a backlog. */
+	if (fc->num_background >= fc->congestion_threshold)
+		return;
+
+	/* Nothing past the end of the file. */
+	last = (size - 1) >> PAGE_SHIFT;
+	if (start > last)
+		return;
+	if (start + nr - 1 > last)
+		nr = last - start + 1;
+
+	/*
+	 * One claimant per range.  A reader that finds the range already
+	 * claimed has nothing to add, and one that jumped elsewhere claims
+	 * afresh, so a re-read from the start is not locked out.
+	 */
+	spin_lock(&fi->lock);
+	claimed = fi->ra_lookahead != start;
+	if (claimed)
+		fi->ra_lookahead = start;
+	spin_unlock(&fi->lock);
+	if (!claimed)
+		return;
+
+	/*
+	 * Same grant the window itself takes: folios the server handed out
+	 * no lock for are folios it will not revoke when a remote node
+	 * writes them.
+	 */
+	if (fc->writeback_cache && fc->dlm) {
+		int err = fuse_get_dlm_lock(file, (loff_t)start << PAGE_SHIFT,
+					    (size_t)nr << PAGE_SHIFT,
+					    FUSE_PAGE_LOCK_READ);
+
+		if (err < 0 && err != -ENOSYS)
+			return;
+	}
+
+	ia = fuse_io_alloc(NULL, nr);
+	if (!ia)
+		return;
+	ap = &ia->ap;
+
+	for (i = 0; i < nr; i++) {
+		struct folio *folio = filemap_alloc_folio(gfp, 0);
+
+		if (!folio)
+			break;
+		/*
+		 * Stop at the first folio already cached, so the request
+		 * stays contiguous and nothing already held is refetched.
+		 */
+		if (filemap_add_folio(mapping, folio, start + i, gfp) < 0) {
+			folio_put(folio);
+			break;
+		}
+		/*
+		 * Hand mm a trigger it can still use.  The reader crossing
+		 * into this range calls page_cache_async_ra(), which looks
+		 * for the first hole ahead and finds the one this request
+		 * stops at, so the next window starts while this one is
+		 * still being consumed.
+		 */
+		if (!added)
+			folio_set_readahead(folio);
+		/* Freshly allocated, so there is no writeback to wait out. */
+		ap->pages[added] = &folio->page;
+		ap->descs[added].length = PAGE_SIZE;
+		added++;
+	}
+
+	if (!added) {
+		fuse_io_free(ia);
+		return;
+	}
+
+	ap->num_pages = added;
+	fuse_send_readpages(ia, file);
+}
+
 static void fuse_readahead(struct readahead_control *rac)
 {
 	struct inode *inode = rac->mapping->host;
 	struct fuse_conn *fc = get_fuse_conn(inode);
-	unsigned int i, max_pages, nr_pages = 0;
+	unsigned int i, max_pages, nr_pages = 0, unit;
+	pgoff_t target_end;
+	bool expanded = false, complete = false;
 
 	if (fuse_is_bad(inode))
 		return;
 
 	max_pages = min_t(unsigned int, fc->max_pages,
-			fc->max_read / PAGE_SIZE);
+			  fc->max_read / PAGE_SIZE);
+
+	/*
+	 * Decide how wide this window is meant to be, rounded up to whole
+	 * requests.  Nothing is allocated here, only the end offset.
+	 *
+	 * mm sizes it from per-fd offset arithmetic, so several threads
+	 * walking one file at a stride each see their own hits as isolated,
+	 * never ramp, and settle on a window barely wider than one read.
+	 * That leaves a request -- and a server round trip -- in the reader's
+	 * path for every read, with the pipeline one deep: measured at
+	 * 73-88KB per request against a server whose threads sat 86% idle.
+	 *
+	 * Requests are the granularity this connection already negotiated, so
+	 * filling out a partial one costs no extra request, only a fuller one,
+	 * and readahead_expand() folds the growth back into ra->size so the
+	 * next window starts from the wider shape instead of collapsing again.
+	 * Only the trailing edge moves, so nothing already read is refetched,
+	 * and the expansion stops at the first folio already cached -- a
+	 * window genuinely bounded by neighbouring data stays bounded.
+	 *
+	 * Never past what read_ahead_kb authorised for this fd: a mount that
+	 * negotiated large requests does not get to prefetch beyond the
+	 * window the admin allowed.
+	 */
+	unit = max_pages;
+	target_end = readahead_index(rac) + readahead_count(rac);
+	if (rac->ra) {
+		unsigned int nr = readahead_count(rac);
+
+		unit = min_t(unsigned int, max_pages, rac->ra->ra_pages);
+		if (unit > 1 && nr % unit)
+			target_end = readahead_index(rac) + roundup(nr, unit);
+	}
+
+	/*
+	 * Readahead fills the page cache past the range the reader locked,
+	 * so take a DLM read grant over the whole window here too.  Folios
+	 * the server handed out no lock for are folios it will not revoke
+	 * when a remote node writes them, and a later read would be served
+	 * from stale cache.  Take the grant before any folio is pulled off
+	 * @rac, so the window that gets populated is the window that is
+	 * covered.
+	 *
+	 * The grant covers the window this call intends rather than the one
+	 * it ends up with, since the folios past what mm built are not
+	 * allocated yet.  readahead_expand() stops at the first folio already
+	 * cached, so a short realisation leaves the tail covered but not
+	 * populated.  That is the harmless direction: coverage without cached
+	 * data serves nothing stale, and a folio already cached is one an
+	 * earlier grant already covers.
+	 *
+	 * Speculative pages are not worth serving uncovered: on a failed
+	 * request drop the window and let read_pages() clean up the folios
+	 * left in @rac.  A server without DLM support answers -ENOSYS and
+	 * clears fc->dlm, which is not a failure.
+	 *
+	 * The round trip is taken before any folio of the window is locked
+	 * and with nothing fenced out, so it holds up this reader and
+	 * nothing else.
+	 */
+	if (fc->writeback_cache && fc->dlm) {
+		size_t len = (size_t)(target_end - readahead_index(rac))
+				<< PAGE_SHIFT;
+		int err = fuse_get_dlm_lock(rac->file, readahead_pos(rac), len,
+					    FUSE_PAGE_LOCK_READ);
+
+		if (err < 0 && err != -ENOSYS)
+			return;
+	}
 
 	for (;;) {
 		struct fuse_io_args *ia;
 		struct fuse_args_pages *ap;
+		unsigned int avail;
 
 		if (fc->num_background >= fc->congestion_threshold &&
 		    rac->ra->async_size >= readahead_count(rac))
@@ -1115,43 +1371,82 @@ static void fuse_readahead(struct readahead_control *rac)
 			 */
 			break;
 
-		nr_pages = readahead_count(rac) - nr_pages;
-		if (nr_pages > max_pages)
-			nr_pages = max_pages;
-		if (nr_pages == 0)
-			break;
+		/*
+		 * @_nr_pages still counts the batch handed out last round,
+		 * which __readahead_batch() retires on its next call, so
+		 * subtracting it gives what is built but not yet sent.
+		 */
+		avail = readahead_count(rac) - nr_pages;
+		if (!avail) {
+			/*
+			 * Everything built is in flight.  Grow the window now
+			 * rather than before the first send.
+			 *
+			 * readahead_expand() allocates, inserts and locks one
+			 * folio per page, and every reader wanting this range
+			 * blocks on the first of them until the request that
+			 * fills it completes.  Building the whole window up
+			 * front puts that walk, hundreds of folios on a mount
+			 * with large requests, ahead of the first byte anyone
+			 * asked for, with nothing in flight to cover it and
+			 * every other reader fenced behind it.  Sending what
+			 * mm built first turns the walk into work done while
+			 * the server is already answering.
+			 *
+			 * _index and _nr_pages are both stale by the
+			 * outstanding batch, so their sum is still the window
+			 * end and the growth lands where it should.
+			 */
+			if (expanded ||
+			    readahead_index(rac) + readahead_count(rac) >=
+				    target_end) {
+				complete = true;
+				break;
+			}
+			expanded = true;
+			readahead_expand(rac, readahead_pos(rac),
+					 (size_t)(target_end -
+						  readahead_index(rac))
+						 << PAGE_SHIFT);
+			avail = readahead_count(rac) - nr_pages;
+			if (!avail) {
+				complete = true;
+				break;
+			}
+		}
+
+		nr_pages = min(avail, max_pages);
 		ia = fuse_io_alloc(NULL, nr_pages);
 		if (!ia)
 			return;
 		ap = &ia->ap;
 		nr_pages = __readahead_batch(rac, ap->pages, nr_pages);
 		for (i = 0; i < nr_pages; i++) {
-			fuse_wait_on_page_writeback(inode,
-						    readahead_index(rac) + i);
+			/* The batch holds a reference, so no lookup needed. */
+			wait_on_page_writeback(ap->pages[i]);
 			ap->descs[i].length = PAGE_SIZE;
 		}
 		ap->num_pages = nr_pages;
 		fuse_send_readpages(ia, rac->file);
 	}
+
+	/*
+	 * The whole window is in flight, so prime the next one before the
+	 * readers get there.  Skipped on a congestion or allocation exit,
+	 * where there is already more queued than the connection wants.
+	 */
+	if (complete)
+		fuse_readahead_lookahead(rac->file, inode, target_end, unit);
 }
 
 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;
 	ssize_t res;
-	int lock_err = 0;
 
 	/*
 	 * In auto invalidate mode, always update attributes on read.
@@ -1169,65 +1464,35 @@ static ssize_t fuse_cache_read_iter(struct kiocb *iocb, struct iov_iter *to)
 	/* if we have dlm support acquire a read lock for the area
 	 * we are reading from. */
 	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);
 
 	/*
-	 * 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).
+	 * A NOTIFY invalidate racing this read drops the folios it
+	 * supersedes, so the read either misses and refetches or returns
+	 * data that was current when it was copied.  There is nothing to
+	 * fence: unlike a write, a read leaves nothing behind that could
+	 * reach the server under a grant it no longer holds.
 	 */
-	if (wb_sem) {
-		int tries = FUSE_DLM_READ_RETRIES;
+	if (fuse_inode_force_dio(inode)) {
+		size_t count = iov_iter_count(to);
 
-retry:
-		percpu_down_read(wb_sem);
-		if (fuse_inode_force_dio(inode)) {
-			percpu_up_read(wb_sem);
-			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.
+		 * A write that passed this same check just before the latch
+		 * took hold dirtied the page cache after the notify dropped
+		 * it, and a direct read does not look there.  Send it first.
 		 */
-		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;
+		if (count) {
+			res = filemap_write_and_wait_range(inode->i_mapping,
+					iocb->ki_pos, iocb->ki_pos + count - 1);
+			if (res)
+				return res;
 		}
+		return fuse_direct_read_iter(iocb, to);
 	}
 
 	res = generic_file_read_iter(iocb, to);
 
-	if (wb_sem)
-		percpu_up_read(wb_sem);
-
 	return res;
 }
 
@@ -1630,19 +1895,13 @@ 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.
  */
-static int fuse_cache_wr_dlm_lock(struct file *file, loff_t pos, size_t len,
-				  bool *unrecorded)
+static int fuse_cache_wr_dlm_lock(struct file *file, loff_t pos, size_t len)
 {
 	int err = fuse_get_dlm_lock(file, pos, len, FUSE_PAGE_LOCK_WRITE);
 
-	if (err < 0 && err != -ENOSYS)
-		return err;
-	*unrecorded = err > 0;
-	return 0;
+	return (err < 0 && err != -ENOSYS) ? err : 0;
 }
 
 /*
@@ -1736,11 +1995,7 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 	struct inode *inode = mapping->host;
 	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;
-	bool wb_guard = false;
 	bool exclusive = true;
-	bool dlm_unrecorded = false;
 	loff_t dlm_pos = 0;
 	size_t dlm_len = 0;
 
@@ -1799,17 +2054,18 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 		 * 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.
+		 * grant-to-use window this leaves open is closed on the way
+		 * out instead: a revoke keeps the range rather than forgetting
+		 * it, and writeback holds it again before sending anything.
+		 * 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 (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);
+			err = fuse_cache_wr_dlm_lock(file, dlm_pos, dlm_len);
 			if (err)
 				return err;
 
@@ -1827,17 +2083,12 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 		}
 
 		/*
-		 * Open-code generic_file_write_iter() so that the coherency
-		 * gate can be held for read across the page-cache dirtying: a
-		 * concurrent NOTIFY_INVAL_INODE -- which takes the write side
-		 * of that gate (blocking, with priority) around its invalidate
-		 * + latch set -- must not be able to strand the folios we are
-		 * about to dirty.  Re-check the latch under it (it may have been
-		 * set while we blocked on the inode lock) and re-route to the
-		 * direct path if it is now set; the DLM write lock taken above
-		 * is harmless there, as the direct path does its own server
-		 * coordination.  wb_sem is NULL on mounts where the gate is
-		 * inactive.
+		 * Open-code generic_file_write_iter() so the DLM lock can be
+		 * taken where the write really lands and the forced-direct-IO
+		 * latch re-checked after the inode lock, which may have been
+		 * set while we blocked on it.  A re-route then goes to the
+		 * direct path; the DLM write lock taken above is harmless
+		 * there, as the direct path does its own server coordination.
 		 */
 		if (exclusive)
 			inode_lock(inode);
@@ -1856,8 +2107,7 @@ 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);
+			err = fuse_cache_wr_dlm_lock(file, dlm_pos, dlm_len);
 			if (err)
 				goto wb_out;
 		}
@@ -1870,16 +2120,14 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 		 * The exclusive inode lock does not pin i_size for the append:
 		 * 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.
+		 * range.  Re-lock where the write really lands.
 		 */
 		if (fc->dlm && (iocb->ki_flags & IOCB_APPEND) &&
 		    iocb->ki_pos != dlm_pos) {
 			dlm_pos = iocb->ki_pos;
 			dlm_len = iov_iter_count(from);
 
-			err = fuse_cache_wr_dlm_lock(file, dlm_pos, dlm_len,
-						     &dlm_unrecorded);
+			err = fuse_cache_wr_dlm_lock(file, dlm_pos, dlm_len);
 			if (err) {
 				written = err;
 				goto wb_out;
@@ -1887,8 +2135,8 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 		}
 
 		/*
-		 * Kill suid/sgid and stamp the timestamps here, before the
-		 * gate, instead of leaving them to
+		 * Kill suid/sgid and stamp the timestamps here, ahead of the
+		 * write itself, instead of leaving them to
 		 * __generic_file_write_iter().  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
@@ -1896,11 +2144,10 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 		 * 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.  file_update_time() only marks the inode
-		 * dirty, but stays next to it to keep the VFS order.
+		 * inode from inside such a handler, and it must not find this
+		 * write holding anything it needs.  file_update_time() only
+		 * marks the inode dirty, but stays next to it to keep the VFS
+		 * order.
 		 */
 		err = file_remove_privs(file);
 		if (!err)
@@ -1910,33 +2157,36 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 			goto wb_out;
 		}
 
-		if (wb_sem) {
-			wb_guard = true;
-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 (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. */
-					written = err;
-					wb_guard = false;
-					goto wb_out;
-				}
-				goto retry;
-			}
+		if (fuse_inode_force_dio(inode)) {
+			/*
+			 * As on the read side, only worse: the direct write
+			 * would land under whatever a write racing the latch
+			 * left dirty, and the invalidate
+			 * fuse_direct_write_iter() does after it launders
+			 * rather than drops, putting that folio on the server
+			 * on top.  Send it first and the order is ordinary.
+			 */
+			count = iov_iter_count(from);
+			if (count)
+				err = filemap_write_and_wait_range(
+						inode->i_mapping, iocb->ki_pos,
+						iocb->ki_pos + count - 1);
+			fuse_cache_wr_unlock(inode, exclusive);
+			if (err)
+				return err;
+			return fuse_direct_write_iter(iocb, from);
 		}
+
+		/*
+		 * A NOTIFY invalidate can revoke the grant requested above
+		 * between here and the dirtying below, and nothing stops it:
+		 * the bytes are caught on the way out instead.
+		 * fuse_dlm_unlock_range() keeps a revoked range for as long as
+		 * there is page cache under it, and writeback holds the range
+		 * again before sending anything.  So a write racing a revoke
+		 * costs a round trip, not coverage.
+		 */
+
 		if (iocb->ki_flags & IOCB_DIRECT) {
 			written = generic_file_direct_write(iocb, from);
 			if (written < 0 || !iov_iter_count(from))
@@ -1957,8 +2207,6 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 				written = generic_perform_write(iocb, from);
 		}
 wb_out:
-		if (wb_guard)
-			percpu_up_read(wb_sem);
 		fuse_cache_wr_unlock(inode, exclusive);
 		if (written > 0)
 			written = generic_write_sync(iocb, written);
@@ -1973,12 +2221,10 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 		goto out;
 
 	/*
-	 * Kill suid/sgid and stamp the timestamps before entering the gate,
-	 * for the reason given in the writeback branch: file_remove_privs()
-	 * can issue a request, and a request must never be waited for under
-	 * the gate.  They run before the forced-DIO re-route below, so a
-	 * re-routed write repeats them; neither has anything left to do the
-	 * second time.
+	 * Kill suid/sgid and stamp the timestamps here, for the reason given
+	 * in the writeback branch.  They run before the forced-DIO re-route
+	 * below, so a re-routed write repeats them; neither has anything left
+	 * to do the second time.
 	 */
 	err = file_remove_privs(file);
 	if (err)
@@ -1990,23 +2236,19 @@ static ssize_t fuse_cache_write_iter(struct kiocb *iocb, struct iov_iter *from)
 
 	/*
 	 * The killpriv fallback lands here with the writeback cache still on,
-	 * so it populates the page cache too and needs the same guard as the
-	 * writeback branch above: hold the coherency gate for read across the
-	 * page-cache population and re-check the latch under it, so a
-	 * concurrent NOTIFY_INVAL_INODE cannot have the cache repopulated
-	 * behind the invalidate it just did.  Still taken before
-	 * task_io_account_write() so a re-route is not double-counted.
-	 * wb_sem is NULL on mounts where the gate is inactive, and such a
-	 * connection never latches either.
+	 * so it populates the page cache too and re-checks the latch like the
+	 * writeback branch above.  Still before task_io_account_write() so a
+	 * re-route is not double-counted.
 	 */
-	wb_guard = !!wb_sem;
-	if (wb_guard) {
-		percpu_down_read(wb_sem);
-		if (fuse_inode_force_dio(inode)) {
-			percpu_up_read(wb_sem);
-			inode_unlock(inode);
-			return fuse_direct_write_iter(iocb, from);
-		}
+	if (fuse_inode_force_dio(inode)) {
+		/* Flush before re-routing; see the writeback branch above. */
+		if (count)
+			err = filemap_write_and_wait_range(inode->i_mapping,
+					iocb->ki_pos, iocb->ki_pos + count - 1);
+		inode_unlock(inode);
+		if (err)
+			return err;
+		return fuse_direct_write_iter(iocb, from);
 	}
 
 	task_io_account_write(count);
@@ -2021,8 +2263,6 @@ 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);
 	inode_unlock(inode);
 	if (written > 0)
 		written = generic_write_sync(iocb, written);
@@ -2600,6 +2840,7 @@ static int fuse_writepage_locked(struct folio *folio)
 	struct address_space *mapping = folio->mapping;
 	struct inode *inode = mapping->host;
 	struct fuse_inode *fi = get_fuse_inode(inode);
+	struct fuse_conn *fc = get_fuse_conn(inode);
 	struct fuse_writepage_args *wpa;
 	struct fuse_args_pages *ap;
 	struct fuse_file *ff;
@@ -2609,6 +2850,22 @@ static int fuse_writepage_locked(struct folio *folio)
 	if (!ff)
 		goto err;
 
+	/*
+	 * Hold the range again before sending it; see fuse_writepages_fill().
+	 * fuse_launder_folio() cleared the folio, so put it back on failure
+	 * rather than lose the bytes; the invalidate that laundered it then
+	 * reports the folio busy, as it does for any folio it cannot free.
+	 */
+	if (fc->dlm && fc->writeback_cache) {
+		error = fuse_dlm_regrant_range(ff, inode, folio_pos(folio),
+					       folio_pos(folio) +
+					       folio_size(folio) - 1);
+		if (error < 0 && error != -ENOSYS) {
+			filemap_dirty_folio(mapping, folio);
+			goto err_writepage_args;
+		}
+	}
+
 	wpa = fuse_writepage_args_setup(folio, ff);
 	error = -ENOMEM;
 	if (!wpa)
@@ -2735,6 +2992,33 @@ static int fuse_writepages_fill(struct folio *folio,
 			goto out_unlock;
 	}
 
+	/*
+	 * Everything dirty here was written whole by this client: the
+	 * unaligned edges of a cached write go to the server directly and
+	 * the interior covers whole pages, so nothing partly written is
+	 * ever dirtied.  There is nothing to classify, only the grant to
+	 * make sure of: a revoke may have arrived since the write, and
+	 * these bytes must not go out from under one.
+	 *
+	 * fuse_dlm_regrant_range() takes the range back when it has gone,
+	 * and walks the record once under the lock held for read when it
+	 * has not.  On failure the folio goes back on the dirty list, so
+	 * the next writeback tries again: write_cache_pages() cleared it
+	 * before calling here and does not put it back itself, and these
+	 * bytes are not on the server.  A server with no DLM answers
+	 * -ENOSYS, which is not a failure.
+	 */
+	if (fc->dlm && fc->writeback_cache) {
+		err = fuse_dlm_regrant_range(data->ff, inode, folio_pos(folio),
+					     folio_pos(folio) +
+					     folio_size(folio) - 1);
+		if (err < 0 && err != -ENOSYS) {
+			folio_redirty_for_writepage(wbc, folio);
+			goto out_unlock;
+		}
+		err = 0;
+	}
+
 	if (wpa && fuse_writepage_need_send(fc, &folio->page, ap, data, wbc)) {
 		fuse_writepages_send(data);
 		data->wpa = NULL;
@@ -3068,9 +3352,9 @@ 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
-	 * 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
+	 * mode.  Revert without the inode 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
 	 * cache is sufficient -- writers re-check the latch and route to cached
 	 * IO once it is clear, and in-flight parallel dio drains itself.  Cached
@@ -3716,9 +4000,29 @@ static long fuse_file_fallocate(struct file *file, int mode, loff_t offset,
 			file_update_time(file);
 	}
 
-	if (mode & (FALLOC_FL_PUNCH_HOLE | FALLOC_FL_ZERO_RANGE))
+	if (mode & (FALLOC_FL_PUNCH_HOLE | FALLOC_FL_ZERO_RANGE)) {
 		truncate_pagecache_range(inode, offset, offset + length - 1);
 
+		/*
+		 * The whole pages inside the hole are gone, so a revoked
+		 * range over them has nothing left to make writeback take
+		 * the grant again for, and would sit in the tree unfreed.
+		 * The pages straddling the ends survive with their punched
+		 * part zeroed, so their record still names real (now zero)
+		 * bytes and stays.  Only when the drop really emptied the
+		 * span: a busy folio that survived keeps its record.
+		 */
+		if (fm->fc->dlm && fm->fc->writeback_cache) {
+			uint64_t first = PAGE_ALIGN(offset);
+			uint64_t last = (uint64_t)(offset + length) & PAGE_MASK;
+
+			if (first < last &&
+			    !filemap_range_has_page(inode->i_mapping, first,
+						    last - 1))
+				fuse_dlm_ranges_dropped(fi, first, last - 1);
+		}
+	}
+
 	fuse_invalidate_attr_mask(inode, FUSE_STATX_MODSIZE);
 
 out:
@@ -3822,9 +4126,24 @@ static ssize_t __fuse_copy_file_range(struct file *file_in, loff_t pos_in,
 	if (err)
 		goto out;
 
-	truncate_inode_pages_range(inode_out->i_mapping,
-				   ALIGN_DOWN(pos_out, PAGE_SIZE),
-				   ALIGN(pos_out + outarg.size, PAGE_SIZE) - 1);
+	{
+		loff_t lstart = ALIGN_DOWN(pos_out, PAGE_SIZE);
+		loff_t lend = ALIGN(pos_out + outarg.size, PAGE_SIZE) - 1;
+
+		truncate_inode_pages_range(inode_out->i_mapping, lstart, lend);
+
+		/*
+		 * The record over the dropped span has nothing left to
+		 * describe, and left DIRTY it would make the next partial
+		 * write there keep and flush folio bytes nobody wrote.
+		 * Only when the drop really emptied it: a folio a
+		 * concurrent fault put back keeps its record.
+		 */
+		if (fc->dlm && fc->writeback_cache &&
+		    !filemap_range_has_page(inode_out->i_mapping, lstart,
+					    lend))
+			fuse_dlm_ranges_dropped(fi_out, lstart, lend);
+	}
 
 	file_update_time(file_out);
 	fuse_write_update_attr(inode_out, pos_out + outarg.size, outarg.size);
@@ -3905,7 +4224,6 @@ static const struct address_space_operations fuse_file_aops  = {
 void fuse_init_file_inode(struct inode *inode, unsigned int flags)
 {
 	struct fuse_inode *fi = get_fuse_inode(inode);
-	struct fuse_conn *fc = get_fuse_conn(inode);
 
 	inode->i_fop = &fuse_file_operations;
 	inode->i_data.a_ops = &fuse_file_aops;
@@ -3917,23 +4235,7 @@ void fuse_init_file_inode(struct inode *inode, unsigned int flags)
 	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->ra_lookahead = 0;
 	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..5bdb2c5045b3f3 100644
--- a/fs/fuse/fuse_dlm_cache.c
+++ b/fs/fuse/fuse_dlm_cache.c
@@ -1,35 +1,113 @@
 // SPDX-License-Identifier: GPL-2.0-only
 /*
  * FUSE page lock cache implementation
+ *
+ * cache->ranges records the grants the server has given this client.  A
+ * grant still on the wire covers nothing and must not appear there, but
+ * a revoke has to be able to find it: otherwise a revoke processed
+ * before the grant is recorded removes nothing, and the grant recorded
+ * afterwards is never taken back.
+ *
+ * A range therefore carries enum fuse_dlm_range_state:
+ *
+ *  - REQUESTED, on cache->pending, while its FUSE_DLM_WB_LOCK is in
+ *    flight.
+ *
+ *  - REVOKED, in either place.  On cache->pending it is a request a
+ *    revoke overlapped while it was in flight, and
+ *    fuse_dlm_request_commit() drops that grant instead of recording it.
+ *    In cache->ranges it is a grant that was recorded and has since been
+ *    taken away, kept because the page cache under it is still
+ *    described.  It covers nothing either way, so the IO paths ask
+ *    again.
+ *
+ *  - READ or WRITE, in cache->ranges.  The only states
+ *    fuse_dlm_range_is_locked() reports as covered; the mode is not a
+ *    separate field, since a range is either not held or held in one
+ *    definite mode.
+ *
+ * In-flight requests are kept off the tree so the state is consulted
+ * only where a request is retired, not by every tree walker.
+ *
+ * The record says nothing about the page cache under a range.  What is
+ * cached there, and whether the server has seen it, is what the page
+ * cache itself answers.
  */
 #include "fuse_i.h"
 #include "fuse_dlm_cache.h"
 
 #include <linux/list.h>
+#include <linux/pagemap.h>
+#include <linux/sched/signal.h>
 #include <linux/types.h>
 #include <linux/slab.h>
 #include <linux/interval_tree_generic.h>
 
 
+/*
+ * How often to ask again for a grant a revoke killed while it was in
+ * flight, before giving up on the range.  Each pass is a round trip.
+ */
+#define FUSE_DLM_GRANT_RETRIES 16
+
+/* Lifecycle of a range; see the file comment above */
+enum fuse_dlm_range_state {
+	/* FUSE_DLM_WB_LOCK in flight, on cache->pending */
+	FUSE_DLM_RANGE_REQUESTED,
+	/*
+	 * On cache->pending, revoked in flight and the grant must not be
+	 * recorded.  In cache->ranges, granted once and taken away, kept to
+	 * describe the page cache under it.  Covers nothing either way.
+	 */
+	FUSE_DLM_RANGE_REVOKED,
+	/* Granted shared, in cache->ranges */
+	FUSE_DLM_RANGE_READ,
+	/* Granted exclusive, in cache->ranges */
+	FUSE_DLM_RANGE_WRITE,
+};
+
 /* A range of pages with a lock */
 struct fuse_dlm_range {
-	/* Interval tree node */
+	/* Interval tree node; only linked once granted */
 	struct rb_node rb;
-	/* Start page offset (inclusive) */
+	/*
+	 * The range, as byte offsets, both inclusive.  Grants arrive page
+	 * aligned, and a range is split only at the bounds of another, so
+	 * these are page aligned too.
+	 */
 	uint64_t start;
-	/* End page offset (inclusive) */
 	uint64_t end;
 	/* Subtree end value for interval tree */
 	uint64_t __subtree_end;
-	/* Lock mode */
-	enum fuse_page_lock_mode mode;
-	/* Temporary list entry for operations */
+	/* Lifecycle and, once granted, the mode; see the enum above */
+	enum fuse_dlm_range_state state;
+	/* Temporary list entry for operations, and the cache->pending link */
 	struct list_head list;
 };
 
-/* Lock modes for FUSE page cache */
-#define FUSE_PCACHE_LK_READ 1 /* Shared read lock */
-#define FUSE_PCACHE_LK_WRITE 2 /* Exclusive write lock */
+/* The state a grant in @mode is recorded under */
+static inline enum fuse_dlm_range_state
+fuse_dlm_granted_state(enum fuse_page_lock_mode mode)
+{
+	return mode == FUSE_PAGE_LOCK_READ ? FUSE_DLM_RANGE_READ :
+					     FUSE_DLM_RANGE_WRITE;
+}
+
+/**
+ * fuse_dlm_state_satisfies - is a range in @held usable for @want
+ * @held: the state of a range recorded in the tree
+ * @want: FUSE_DLM_RANGE_READ or FUSE_DLM_RANGE_WRITE
+ *
+ * A WRITE grant is exclusive and so covers a READ request; nothing else
+ * substitutes for anything.  The two pending states never appear in the
+ * tree and cover nothing.
+ */
+static inline bool fuse_dlm_state_satisfies(enum fuse_dlm_range_state held,
+					    enum fuse_dlm_range_state want)
+{
+	return held == want ||
+	       (held == FUSE_DLM_RANGE_WRITE && want == FUSE_DLM_RANGE_READ);
+}
 
 /* Interval tree definitions for page ranges */
 static inline uint64_t fuse_dlm_range_start(struct fuse_dlm_range *range)
@@ -46,6 +124,33 @@ INTERVAL_TREE_DEFINE(struct fuse_dlm_range, rb, uint64_t, __subtree_end,
 		    		fuse_dlm_range_start, fuse_dlm_range_last, static,
 		    		fuse_page_it);
 
+static void fuse_dlm_split_at(struct fuse_dlm_cache *cache, uint64_t off);
+
+/**
+ * fuse_dlm_kill_pending - mark in-flight requests overlapping [start, end]
+ * @cache: The page cache
+ * @start: Start byte offset of the revoked region
+ * @end: End byte offset of the revoked region
+ *
+ * A revoke overlapping a request still on the wire has nothing to remove
+ * from the tree, since that grant is not recorded yet.  Marking it makes
+ * fuse_dlm_request_commit() drop the grant instead of recording it.
+ *
+ * The nodes are owned by the threads waiting on their replies: mark
+ * only, never remove or free.
+ *
+ * Caller holds @cache->lock for write.
+ */
+static void fuse_dlm_kill_pending(struct fuse_dlm_cache *cache,
+				  uint64_t start, uint64_t end)
+{
+	struct fuse_dlm_range *req;
+
+	list_for_each_entry(req, &cache->pending, list)
+		if (req->start <= end && start <= req->end)
+			req->state = FUSE_DLM_RANGE_REVOKED;
+}
+
 /**
  * fuse_page_cache_init - Initialize a page cache lock manager
  * @cache: The cache to initialize
@@ -63,7 +168,7 @@ int fuse_dlm_cache_init(struct fuse_inode *inode)
 
 	init_rwsem(&cache->lock);
 	cache->ranges = RB_ROOT_CACHED;
-	cache->revoke_gen = 0;
+	INIT_LIST_HEAD(&cache->pending);
 
 	return 0;
 }
@@ -85,7 +190,11 @@ 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);
+	/*
+	 * Every grant goes, so every request in flight is revoked.  Mark
+	 * only; each node is owned by the thread waiting on its reply.
+	 */
+	fuse_dlm_kill_pending(cache, 0, U64_MAX);
 	while ((node = rb_first_cached(&cache->ranges)) != NULL) {
 		range = rb_entry(node, struct fuse_dlm_range, rb);
 		fuse_page_it_remove(range, &cache->ranges);
@@ -97,8 +206,8 @@ void fuse_dlm_cache_release_locks(struct fuse_inode *inode)
 /**
  * fuse_dlm_find_overlapping - Find a range that overlaps with [start, end]
  * @cache: The page cache
- * @start: Start page offset
- * @end: End page offset
+ * @start: Start byte offset
+ * @end: End byte offset
  *
  * Return: Pointer to the first overlapping range, or NULL if none found
  */
@@ -112,8 +221,8 @@ fuse_dlm_find_overlapping(struct fuse_dlm_cache *cache, uint64_t start,
 /**
  * fuse_page_try_merge - Try to merge ranges within a specific region
  * @cache: The page cache
- * @start: Start page offset
- * @end: End page offset
+ * @start: Start byte offset
+ * @end: End byte offset
  *
  * Attempt to merge ranges within and adjacent to the specified region
  * that have the same lock mode.
@@ -148,8 +257,8 @@ static void fuse_dlm_try_merge(struct fuse_dlm_cache *cache, uint64_t start,
 					struct fuse_dlm_range, rb);
 		}
 
-		/* Try to merge with next range if adjacent and same mode */
-		if (next && range->mode == next->mode &&
+		/* Merge neighbours the server has given us on the same terms */
+		if (next && range->state == next->state &&
 		    range->end + 1 == next->start) {
 			/* Merge ranges: re-insert so __subtree_end is updated */
 			fuse_page_it_remove(next, &cache->ranges);
@@ -168,13 +277,11 @@ static void fuse_dlm_try_merge(struct fuse_dlm_cache *cache, uint64_t start,
 }
 
 /**
- * __fuse_dlm_lock_range - Lock a range of pages
- * @cache: The page cache
- * @start: Start page offset
- * @end: End page offset
+ * fuse_dlm_lock_range_locked - Record a granted range of pages
+ * @inode: The fuse inode
+ * @start: Start byte offset
+ * @end: End byte 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.
@@ -183,15 +290,20 @@ static void fuse_dlm_try_merge(struct fuse_dlm_cache *cache, uint64_t start,
  * - READ locks are compatible with existing WRITE locks (downgrade not needed)
  * - WRITE locks need to upgrade existing READ locks
  *
+ * Everything inserted here is READ or WRITE: this runs only after the
+ * server has answered.
+ *
+ * Caller holds the cache lock for write.
+ *
  * 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)
+static int fuse_dlm_lock_range_locked(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;
-	int lock_mode;
+	enum fuse_dlm_range_state want;
 	bool covered_to_end = false;
 	int ret = 0;
 	LIST_HEAD(to_lock);
@@ -201,22 +313,21 @@ static int __fuse_dlm_lock_range(struct fuse_inode *inode, uint64_t start,
 	if (!cache || start > end)
 		return -EINVAL;
 
-	/* Convert to lock mode */
-	lock_mode = (mode == FUSE_PAGE_LOCK_READ) ? FUSE_PCACHE_LK_READ :
-						    FUSE_PCACHE_LK_WRITE;
-
-	down_write(&cache->lock);
+	/* The state this grant records */
+	want = fuse_dlm_granted_state(mode);
 
 	/*
-	 * 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.
+	 * Ranges are upgraded whole below, so split at the grant bounds
+	 * first: a range extending past the grant would otherwise have its
+	 * uncovered part upgraded with it, recording coverage the server
+	 * never gave.  A read range half-covered by a write grant would
+	 * report the other half held for write, and a revoked range
+	 * half-regranted would report its still-revoked bytes as held, so
+	 * writeback would send them without taking the range again.
 	 */
-	if (genp && cache->revoke_gen != *genp) {
-		up_write(&cache->lock);
-		return -EAGAIN;
-	}
+	fuse_dlm_split_at(cache, start);
+	if (end < U64_MAX)
+		fuse_dlm_split_at(cache, end + 1);
 
 	/* Find all ranges that overlap with [start, end] */
 	range = fuse_page_it_iter_first(&cache->ranges, start, end);
@@ -224,17 +335,19 @@ static int __fuse_dlm_lock_range(struct fuse_inode *inode, uint64_t start,
 		/* Get next overlapping range before we potentially modify the tree */
 		next = fuse_page_it_iter_next(range, start, end);
 
-		/* Check lock compatibility */
-		if (lock_mode == FUSE_PCACHE_LK_WRITE &&
-		    lock_mode != range->mode) {
-			/* we own the lock but have to update it. */
+		/*
+		 * A revoked range is covered again by this grant, and a read
+		 * range needs upgrading when a write is granted.
+		 */
+		if (range->state == FUSE_DLM_RANGE_REVOKED ||
+		    (want == FUSE_DLM_RANGE_WRITE &&
+		     range->state != FUSE_DLM_RANGE_WRITE))
 			list_add_tail(&range->list, &to_upgrade);
-		}
 		/* If WRITE lock already exists - nothing to do */
 
 		/* If there's a gap before this range, we need to add the missing range */
 		if (current_start < range->start) {
-			new_range = kmalloc(sizeof(*new_range), GFP_KERNEL);
+			new_range = kmalloc(sizeof(*new_range), GFP_NOFS);
 			if (!new_range) {
 				ret = -ENOMEM;
 				goto out_free;
@@ -242,7 +355,7 @@ static int __fuse_dlm_lock_range(struct fuse_inode *inode, uint64_t start,
 
 			new_range->start = current_start;
 			new_range->end = range->start - 1;
-			new_range->mode = lock_mode;
+			new_range->state = want;
 			INIT_LIST_HEAD(&new_range->list);
 
 			list_add_tail(&new_range->list, &to_lock);
@@ -260,7 +373,7 @@ static int __fuse_dlm_lock_range(struct fuse_inode *inode, uint64_t start,
 
 	/* If there's a gap after the last range to the end, extend the range */
 	if (!covered_to_end && current_start <= end) {
-		new_range = kmalloc(sizeof(*new_range), GFP_KERNEL);
+		new_range = kmalloc(sizeof(*new_range), GFP_NOFS);
 		if (!new_range) {
 			ret = -ENOMEM;
 			goto out_free;
@@ -268,17 +381,15 @@ static int __fuse_dlm_lock_range(struct fuse_inode *inode, uint64_t start,
 
 		new_range->start = current_start;
 		new_range->end = end;
-		new_range->mode = lock_mode;
+		new_range->state = want;
 		INIT_LIST_HEAD(&new_range->list);
 
 		list_add_tail(&new_range->list, &to_lock);
 	}
 
-	/* update locks, if any lock is in this list it has the wrong mode */
-	list_for_each_entry(range, &to_upgrade, list) {
-		/* Update the lock mode */
-		range->mode = lock_mode;
-	}
+	/* Everything on this list is now covered in @want */
+	list_for_each_entry(range, &to_upgrade, list)
+		range->state = want;
 
 	/* Add all new ranges to the tree */
 	list_for_each_entry(new_range, &to_lock, list) {
@@ -289,7 +400,6 @@ static int __fuse_dlm_lock_range(struct fuse_inode *inode, uint64_t start,
 	/* Try to merge adjacent ranges with the same mode */
 	fuse_dlm_try_merge(cache, start, end);
 
-	up_write(&cache->lock);
 	return 0;
 
 out_free:
@@ -301,125 +411,229 @@ static int __fuse_dlm_lock_range(struct fuse_inode *inode, uint64_t start,
 		kfree(new_range);
 	}
 
-	/* Restore original lock modes for any partially upgraded locks */
-	list_for_each_entry(range, &to_upgrade, list) {
-		if (lock_mode == FUSE_PCACHE_LK_WRITE) {
-			/* We upgraded this lock but failed later, downgrade it back */
-			range->mode = FUSE_PCACHE_LK_READ;
-		}
-	}
-
-	up_write(&cache->lock);
+	/*
+	 * Nothing to undo on @to_upgrade: every goto here is taken before
+	 * the loop above runs, so no state has been changed yet.
+	 */
 	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);
-}
+	struct fuse_dlm_cache *cache = &inode->dlm_locked_areas;
+	int ret;
 
-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);
+	down_write(&cache->lock);
+	ret = fuse_dlm_lock_range_locked(inode, start, end, mode);
+	up_write(&cache->lock);
+
+	return ret;
 }
 
 /**
- * fuse_dlm_revoke_gen - sample the revocation generation
+ * fuse_dlm_request_begin - publish a lock request before it is sent
  * @inode: the fuse inode
+ * @req: caller-owned storage for the request, live until commit or abort
+ * @start: start byte offset being requested (inclusive)
+ * @end: end byte offset being requested (inclusive)
+ *
+ * The mode is not recorded here: until the server answers the range is
+ * held in neither, and the mode that reaches the tree is the one passed
+ * to fuse_dlm_request_commit().
+ *
+ * A FUSE_DLM_WB_LOCK reply and a NOTIFY revoke are serviced on different
+ * threads, so a revoke can be processed before the grant the reply
+ * carries is recorded.  Publishing the request before it leaves gives
+ * that revoke a node to mark; without one it removes nothing, and the
+ * grant recorded afterwards is never taken back by any later NOTIFY.
  *
- * 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.
+ * The request covers nothing while in flight, so it is kept off the
+ * tree.  @req is reachable only through cache->pending, which both
+ * fuse_dlm_request_commit() and fuse_dlm_request_abort() unlink under
+ * the cache lock before the caller returns; stack storage is therefore
+ * fine and nothing is allocated here.
  */
-uint64_t fuse_dlm_revoke_gen(struct fuse_inode *inode)
+void fuse_dlm_request_begin(struct fuse_inode *inode,
+			    struct fuse_dlm_range *req, uint64_t start,
+			    uint64_t end)
 {
-	return READ_ONCE(inode->dlm_locked_areas.revoke_gen);
+	struct fuse_dlm_cache *cache = &inode->dlm_locked_areas;
+
+	RB_CLEAR_NODE(&req->rb);
+	req->start = start;
+	req->end = end;
+	req->state = FUSE_DLM_RANGE_REQUESTED;
+	/* Nothing reads this while the request is pending; publish it set */
+
+	down_write(&cache->lock);
+	list_add_tail(&req->list, &cache->pending);
+	up_write(&cache->lock);
 }
 
 /**
- * fuse_dlm_punch_hole - Punch a hole in a locked range
- * @cache: The page cache
- * @start: Start page offset of the hole
- * @end: End page offset of the hole
+ * fuse_dlm_request_commit - retire a request and record its grant
+ * @inode: the fuse inode
+ * @req: the request published by fuse_dlm_request_begin()
+ * @start: start byte offset the server granted (inclusive)
+ * @end: end byte offset the server granted (inclusive)
+ * @mode: the mode that was requested
  *
- * Create a hole in a locked range by splitting it into two ranges.
+ * Unlinking @req and recording the grant are one step under the cache
+ * lock, so a revoke lands either before it and is seen on @req, or after
+ * it and finds the grant in the tree.
  *
- * Return: 0 on success, negative error code on failure
+ * @req is retired in every case and may be reused.
+ *
+ * Return: -EAGAIN if a revoke overlapped @req while it was in flight,
+ * nothing recorded; otherwise the result of recording the grant.
  */
-static int fuse_dlm_punch_hole(struct fuse_dlm_cache *cache, uint64_t start,
-			       uint64_t end)
+int fuse_dlm_request_commit(struct fuse_inode *inode,
+			    struct fuse_dlm_range *req, uint64_t start,
+			    uint64_t end, enum fuse_page_lock_mode mode)
 {
-	struct fuse_dlm_range *range, *new_range;
+	struct fuse_dlm_cache *cache = &inode->dlm_locked_areas;
+	bool revoked;
 	int ret = 0;
 
-	if (!cache || start > end)
-		return -EINVAL;
+	down_write(&cache->lock);
+	list_del(&req->list);
+	revoked = req->state == FUSE_DLM_RANGE_REVOKED;
+	if (!revoked)
+		ret = fuse_dlm_lock_range_locked(inode, start, end, mode);
+	up_write(&cache->lock);
 
-	/* Find a range that contains [start, end] */
-	range = fuse_dlm_find_overlapping(cache, start, end);
-	if (!range) {
-		ret = -EINVAL;
-		goto out;
-	}
+	return revoked ? -EAGAIN : ret;
+}
 
-	/* If the hole is at the beginning of the range */
-	if (start == range->start) {
-		fuse_page_it_remove(range, &cache->ranges);
-		range->start = end + 1;
-		fuse_page_it_insert(range, &cache->ranges);
-		goto out;
-	}
+/**
+ * fuse_dlm_request_abort - retire a request that got no usable reply
+ * @inode: the fuse inode
+ * @req: the request published by fuse_dlm_request_begin()
+ *
+ * Nothing is recorded, so a mark left by a revoke does not matter.
+ */
+void fuse_dlm_request_abort(struct fuse_inode *inode,
+			    struct fuse_dlm_range *req)
+{
+	struct fuse_dlm_cache *cache = &inode->dlm_locked_areas;
 
-	/* If the hole is at the end of the range */
-	if (end == range->end) {
-		fuse_page_it_remove(range, &cache->ranges);
-		range->end = start - 1;
-		fuse_page_it_insert(range, &cache->ranges);
-		goto out;
-	}
+	down_write(&cache->lock);
+	list_del(&req->list);
+	up_write(&cache->lock);
+}
 
-	/* The hole is in the middle, need to split */
-	new_range = kmalloc(sizeof(*new_range), GFP_KERNEL);
-	if (!new_range) {
-		ret = -ENOMEM;
-		goto out;
-	}
+/**
+ * fuse_dlm_split_at - make @off start a range
+ * @cache: The page cache
+ * @off: byte offset to split at
+ *
+ * Splits the range containing @off in two, both halves keeping the state
+ * of the original, so a revoke can apply to one side only.  A no-op when
+ * @off already starts a range or falls in a gap.
+ *
+ * Caller holds @cache->lock for write.
+ *
+ * Cannot fail: the split decides which bytes a caller goes on to name,
+ * and both naming more than was written and lowering more than was sent
+ * lose data.  iomap allocates the state it keeps per folio the same way.
+ */
+static void fuse_dlm_split_at(struct fuse_dlm_cache *cache, uint64_t off)
+{
+	struct fuse_dlm_range *range, *tail;
 
-	/* Copy properties from original range */
-	*new_range = *range;
-	INIT_LIST_HEAD(&new_range->list);
+	if (!off)
+		return;
+
+	range = fuse_page_it_iter_first(&cache->ranges, off, off);
+	if (!range || range->start == off)
+		return;
 
-	/* Adjust ranges */
-	new_range->start = end + 1;
-	range->end = start - 1;
+	tail = kmalloc(sizeof(*tail), GFP_NOFS | __GFP_NOFAIL);
 
-	/* Update interval tree */
+	*tail = *range;
+	INIT_LIST_HEAD(&tail->list);
+	tail->start = off;
+
+	/*
+	 * Bounds are never edited in place: the interval tree caches a
+	 * subtree end that only insertion recomputes.
+	 */
 	fuse_page_it_remove(range, &cache->ranges);
+	range->end = off - 1;
 	fuse_page_it_insert(range, &cache->ranges);
-	fuse_page_it_insert(new_range, &cache->ranges);
+	fuse_page_it_insert(tail, &cache->ranges);
+}
 
-out:
-	return ret;
+/**
+ * fuse_dlm_ranges_dropped - the page cache under [start, end] is gone
+ * @inode: the fuse inode
+ * @start: start byte offset (inclusive)
+ * @end: end byte offset (inclusive)
+ *
+ * A revoked range exists only to describe page cache dirtied before the
+ * grant was taken away.  Once that cache is gone the range has nothing
+ * left to say and is freed; a range still held goes back to describing
+ * nothing.
+ *
+ * The caller must have established that the range really is empty, not
+ * merely asked for it to be dropped: a folio that survived an
+ * invalidate is still there, and claiming otherwise would let writeback
+ * send it with no record of where it came from.
+ */
+void fuse_dlm_ranges_dropped(struct fuse_inode *inode, uint64_t start,
+			     uint64_t end)
+{
+	struct fuse_dlm_cache *cache = &inode->dlm_locked_areas;
+	struct fuse_dlm_range *range, *next;
+
+	if (start > end)
+		return;
+
+	down_write(&cache->lock);
+
+	fuse_dlm_split_at(cache, start);
+	if (end < U64_MAX)
+		fuse_dlm_split_at(cache, end + 1);
+
+	range = fuse_page_it_iter_first(&cache->ranges, start, end);
+	while (range) {
+		next = fuse_page_it_iter_next(range, start, end);
+
+		if (range->state == FUSE_DLM_RANGE_REVOKED) {
+			fuse_page_it_remove(range, &cache->ranges);
+			kfree(range);
+		}
+
+		range = next;
+	}
+
+	fuse_dlm_try_merge(cache, start, end);
+
+	up_write(&cache->lock);
 }
 
 /**
- * fuse_dlm_unlock_range - Unlock a range of pages
- * @cache: The page cache
- * @start: Start page offset
- * @end: End page offset
+ * fuse_dlm_unlock_range - Revoke the grants over a range of pages
+ * @inode: The fuse inode
+ * @start: Start byte offset
+ * @end: End byte offset
+ *
+ * The server has taken [start, end] back.  A range that has nothing
+ * cached under it is removed; one that has is kept and marked
+ * FUSE_DLM_RANGE_REVOKED, so the page cache it covers stays described.
+ * Removing it instead would leave a gap, and a gap reads as "no record",
+ * which is what an untracked range looks like: writeback would then send
+ * folios dirtied under the grant that was just taken away.
+ *
+ * A revoked range covers nothing, so fuse_dlm_range_is_locked() reports
+ * it uncovered and the IO paths request again.
  *
- * Release locks on the specified range of pages.  An inverted range is
- * rejected rather than silently removing nothing: the callers revoke
- * coverage, and a revoke that quietly keeps the grant alive would let
- * the re-validating IO paths trust a lock the server has taken away.
- * To drop every grant use fuse_dlm_cache_release_locks() (there is no
- * in-band sentinel range for it).
+ * An inverted range is rejected rather than silently revoking nothing:
+ * the callers revoke coverage, and a revoke that quietly keeps the grant
+ * alive would let the re-validating IO paths trust a lock the server has
+ * taken away.  To drop every grant use fuse_dlm_cache_release_locks()
+ * (there is no in-band sentinel range for it).
  *
  * Return: 0 on success, negative error code on failure
  */
@@ -428,7 +642,6 @@ int fuse_dlm_unlock_range(struct fuse_inode *inode,
 {
 	struct fuse_dlm_cache *cache = &inode->dlm_locked_areas;
 	struct fuse_dlm_range *range, *next;
-	int ret = 0;
 
 	if (!cache || start > end)
 		return -EINVAL;
@@ -436,39 +649,37 @@ 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()).
+	 * Before touching the tree, and even when nothing in the tree
+	 * overlaps: a revoke racing an in-flight grant finds no overlap
+	 * because that grant is not recorded yet.
 	 */
-	WRITE_ONCE(cache->revoke_gen, cache->revoke_gen + 1);
+	fuse_dlm_kill_pending(cache, start, end);
+
+	/* Split so the revoked region has its own ranges */
+	fuse_dlm_split_at(cache, start);
+	if (end < U64_MAX)
+		fuse_dlm_split_at(cache, end + 1);
 
-	/* Find all ranges that overlap with [start, end] */
 	range = fuse_page_it_iter_first(&cache->ranges, start, end);
 	while (range) {
-		/* Get next overlapping range before we potentially modify the tree */
+		/* Get next overlapping range before we modify the tree */
 		next = fuse_page_it_iter_next(range, start, end);
 
-		/* Check if we need to punch a hole */
-		if (start > range->start && end < range->end) {
-			/* Punch a hole in the middle */
-			ret = fuse_dlm_punch_hole(cache, start, end);
-			if (ret)
-				goto out;
-			/* After punching a hole, we're done */
-			break;
-		} else if (start > range->start) {
-			/* Adjust the end of the range */
-			fuse_page_it_remove(range, &cache->ranges);
-			range->end = start - 1;
-			fuse_page_it_insert(range, &cache->ranges);
-		} else if (end < range->end) {
-			/* Adjust the start of the range */
-			fuse_page_it_remove(range, &cache->ranges);
-			range->start = end + 1;
-			fuse_page_it_insert(range, &cache->ranges);
+		/*
+		 * A revoked range is kept only to say that the page cache
+		 * under it was dirtied under a grant that has gone, so that
+		 * writeback takes the range again before sending it.  With
+		 * nothing cached there it has nothing to say.
+		 *
+		 * A grant with no end is recorded to U64_MAX; the page cache
+		 * is indexed by a signed offset, so ask it about as much of
+		 * that as it can name.
+		 */
+		if (filemap_range_has_page(inode->inode.i_mapping, range->start,
+					   min_t(uint64_t, range->end,
+						 LLONG_MAX))) {
+			range->state = FUSE_DLM_RANGE_REVOKED;
 		} else {
-			/* Complete overlap, remove the range */
 			fuse_page_it_remove(range, &cache->ranges);
 			kfree(range);
 		}
@@ -476,20 +687,18 @@ int fuse_dlm_unlock_range(struct fuse_inode *inode,
 		range = next;
 	}
 
-out:
+	fuse_dlm_try_merge(cache, start, end);
+
 	up_write(&cache->lock);
-	return ret;
+	return 0;
 }
 
 /**
- * fuse_dlm_range_is_locked - Check if a page range is already locked
- * @cache: The page cache
- * @start: Start page offset
- * @end: End page offset
- * @mode: Lock mode to check for (or NULL to check for any lock)
- *
- * Check if the specified range of pages is already locked.
- * The entire range must be locked for this to return true.
+ * fuse_dlm_range_is_locked - Check if a byte range is already locked
+ * @inode: The fuse inode
+ * @start: Start byte offset
+ * @end: End byte offset
+ * @mode: Lock mode to check for
  *
  * Return: true if the entire range is locked, false otherwise
  */
@@ -498,17 +707,14 @@ bool fuse_dlm_range_is_locked(struct fuse_inode *inode, uint64_t start,
 {
 	struct fuse_dlm_cache *cache = &inode->dlm_locked_areas;
 	struct fuse_dlm_range *range;
-	int lock_mode = 0;
+	enum fuse_dlm_range_state want;
 	uint64_t current_start = start;
 
 	if (!cache || start > end)
 		return false;
 
-	/* Convert to lock mode if specified */
-	if (mode == FUSE_PAGE_LOCK_READ)
-		lock_mode = FUSE_PCACHE_LK_READ;
-	else if (mode == FUSE_PAGE_LOCK_WRITE)
-		lock_mode = FUSE_PCACHE_LK_WRITE;
+	/* The state a range has to be in to cover this request */
+	want = fuse_dlm_granted_state(mode);
 
 	down_read(&cache->lock);
 
@@ -525,7 +731,7 @@ bool fuse_dlm_range_is_locked(struct fuse_inode *inode, uint64_t start,
 		 * re-requesting a READ lock for a range we already hold
 		 * a WRITE lock on (e.g. read-after-write).
 		 */
-		if (lock_mode && range->mode < lock_mode) {
+		if (!fuse_dlm_state_satisfies(range->state, want)) {
 			/* Held lock is weaker than requested */
 			up_read(&cache->lock);
 			return false;
@@ -572,6 +778,11 @@ bool fuse_dlm_range_is_locked(struct fuse_inode *inode, uint64_t start,
  * server has not seen, so its mtime and ctime run ahead of anything the
  * server can report.
  *
+ * A revoked range is not counted.  The state does not keep the mode the
+ * range was granted in, so a revoked one cannot be told from a read that
+ * was taken away, and the dirty page cache such a range describes is
+ * what the caller checks the mapping for instead.
+ *
  * Return: true if at least one recorded range is held for write
  */
 bool fuse_dlm_write_grant_exists(struct fuse_inode *fi)
@@ -583,7 +794,7 @@ bool fuse_dlm_write_grant_exists(struct fuse_inode *fi)
 	down_read(&cache->lock);
 	for (range = fuse_dlm_find_overlapping(cache, 0, U64_MAX); range;
 	     range = fuse_page_it_iter_next(range, 0, U64_MAX)) {
-		if (range->mode == FUSE_PCACHE_LK_WRITE) {
+		if (range->state == FUSE_DLM_RANGE_WRITE) {
 			held = true;
 			break;
 		}
@@ -634,11 +845,10 @@ bool fuse_dlm_lock_is_held(struct fuse_inode *fi, loff_t offset,
  * re-validating the grant must not re-request on a nonzero return or
  * they would spin.
  */
-int fuse_get_dlm_lock(struct file *file, loff_t offset,
-		      size_t length, enum fuse_page_lock_mode mode)
+static int __fuse_get_dlm_lock(struct fuse_file *ff, struct inode *inode,
+			       loff_t offset, size_t length,
+			       enum fuse_page_lock_mode mode)
 {
-	struct fuse_file *ff = file->private_data;
-	struct inode *inode = file_inode(file);
 	struct fuse_conn *fc = get_fuse_conn(inode);
 	struct fuse_inode *fi = get_fuse_inode(inode);
 	struct fuse_mount *fm = ff->fm;
@@ -646,13 +856,23 @@ int fuse_get_dlm_lock(struct file *file, loff_t offset,
 	FUSE_ARGS(args);
 	struct fuse_dlm_lock_in inarg;
 	struct fuse_dlm_lock_out outarg;
-	uint64_t gen;
+	struct fuse_dlm_range req;
+	uint64_t pg_start, pg_end;
+	int tries = FUSE_DLM_GRANT_RETRIES;
 	int err;
 
 	/* An empty range needs no lock. */
 	if (!length)
 		return 0;
 
+	/*
+	 * note that the offset and length don't have to be page aligned
+	 * here but since we only get here on writeback caching we will
+	 * send out page aligned requests
+	 */
+	pg_start = (uint64_t)offset & PAGE_MASK;
+	pg_end = ((uint64_t)offset + length - 1) | (PAGE_SIZE - 1);
+
 restart:
 	/* note that this can be run from different processes
 	 * at the same time. It is intentionally not protected
@@ -661,27 +881,19 @@ 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))
-		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);
+	if (fuse_dlm_lock_is_held(fi, offset, length, mode)) {
+		/*
+		 * Already covered, and the record says nothing beyond that,
+		 * so this is one shared acquisition end to end.
+		 */
+		return 0;
+	}
 
 	memset(&inarg, 0, sizeof(inarg));
 	inarg.fh = ff->fh;
 
-	/* note that the offset and length don't have to be page aligned
-	 * here but since we only get here on writeback caching we will
-	 * send out page aligned requests */
-	inarg.start = offset & PAGE_MASK;
-	inarg.end = (offset + length - 1) | (PAGE_SIZE - 1);
+	inarg.start = pg_start;
+	inarg.end = pg_end;
 	inarg.type = (mode == FUSE_PAGE_LOCK_WRITE) ?
 		FUSE_DLM_LOCK_WRITE : FUSE_DLM_LOCK_READ;
 
@@ -693,18 +905,23 @@ int fuse_get_dlm_lock(struct file *file, loff_t offset,
 	args.out_numargs = 1;
 	args.out_args[0].size = sizeof(outarg);
 	args.out_args[0].value = &outarg;
+
+	/* Publish before sending; see fuse_dlm_request_begin() */
+	fuse_dlm_request_begin(fi, &req, inarg.start, inarg.end);
+
 	err = fuse_simple_request(fm, &args);
-	if (err == -ENOSYS) {
-		/* fuse server does not support dlm, save the info */
-		fc->dlm = 0;
+	if (err) {
+		fuse_dlm_request_abort(fi, &req);
+		if (err == -ENOSYS) {
+			/* fuse server does not support dlm, save the info */
+			fc->dlm = 0;
+		}
 		return err;
 	}
 
-	if (err)
-		return err;
-
 	if (inarg.start < outarg.start || inarg.end > outarg.end) {
 		/* fuse server is seriously broken */
+		fuse_dlm_request_abort(fi, &req);
 		pr_warn("fuse: dlm lock request for %llu:%llu returned %llu:%llu bytes\n",
 			inarg.start, inarg.end, outarg.start, outarg.end);
 		fuse_abort_conn(fc);
@@ -712,21 +929,33 @@ 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.
+	 * Retire the request and record the grant.  A server may grant more
+	 * than was asked for, and recording all of it is what lets later IO
+	 * over the same neighbourhood skip the round trip.
+	 *
+	 * fuse_dlm_kill_pending() matches a request in flight on the bounds
+	 * published to it, so a revoke aimed only at the part beyond them
+	 * marks nothing here.  It would have to be a revoke of a sub-range
+	 * the server is granting in the same breath.
 	 */
-	err = fuse_dlm_lock_range_gen(fi, outarg.start, outarg.end, mode, gen);
+	err = fuse_dlm_request_commit(fi, &req, outarg.start, outarg.end, mode);
 	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.
+		 * A revoke overlapping this range was processed while the
+		 * request was in flight, so the grant is dead.  Retry
+		 * rather than fail: no one else holds the range, and the
+		 * write path turns an error into a failed write.
+		 *
+		 * Not forever, though.  Every pass is a whole round trip,
+		 * which throttles the loop but does not end it, and
+		 * writeback asks for a grant with a folio locked, so a node
+		 * revoking as fast as the grants arrive would hold that
+		 * folio and this task for as long as it kept going.
 		 */
+		if (fatal_signal_pending(current))
+			return -EINTR;
+		if (!tries--)
+			return -EIO;
 		goto restart;
 	}
 
@@ -743,3 +972,32 @@ int fuse_get_dlm_lock(struct file *file, loff_t offset,
 
 	return 0;
 }
+
+int fuse_get_dlm_lock(struct file *file, loff_t offset,
+		      size_t length, enum fuse_page_lock_mode mode)
+{
+	return __fuse_get_dlm_lock(file->private_data, file_inode(file),
+				   offset, length, mode);
+}
+
+/**
+ * fuse_dlm_regrant_range - hold [start, end] again for writeback
+ * @ff: a fuse file open for writing on @inode
+ * @inode: the inode
+ * @start: start byte offset (inclusive)
+ * @end: end byte offset (inclusive)
+ *
+ * Writeback holds the range again before sending a folio, since a revoke
+ * may have arrived between the write and the send.  Whatever the other
+ * holder wrote in between is overwritten, which for two writers that
+ * never synchronised is a legitimate order.
+ *
+ * A range still held is the ordinary case: the grant is found recorded
+ * and nothing is sent to the server.
+ */
+int fuse_dlm_regrant_range(struct fuse_file *ff, struct inode *inode,
+			   uint64_t start, uint64_t end)
+{
+	return __fuse_get_dlm_lock(ff, inode, start, end - start + 1,
+				   FUSE_PAGE_LOCK_WRITE);
+}
diff --git a/fs/fuse/fuse_dlm_cache.h b/fs/fuse/fuse_dlm_cache.h
index 30fdbb26bd3daf..d1078cd84546d5 100644
--- a/fs/fuse/fuse_dlm_cache.h
+++ b/fs/fuse/fuse_dlm_cache.h
@@ -13,6 +13,8 @@
 
 
 struct fuse_inode;
+struct fuse_dlm_range;
+struct fuse_file;
 
 /* Lock modes for page ranges */
 enum fuse_page_lock_mode { FUSE_PAGE_LOCK_READ, FUSE_PAGE_LOCK_WRITE };
@@ -26,19 +28,26 @@ enum fuse_page_lock_mode { FUSE_PAGE_LOCK_READ, FUSE_PAGE_LOCK_WRITE };
  */
 #define FUSE_DLM_GRANT_UNRECORDED 1
 
-/* Page cache lock manager */
+/*
+ * Page cache lock manager.
+ *
+ * @ranges holds the grants the client has been given, and the ones it
+ * has had taken back that still describe page cache
+ * (FUSE_DLM_RANGE_REVOKED).  A request still on the wire covers nothing
+ * and lives on @pending instead, so tree walkers never filter on state.
+ * See enum fuse_dlm_range_state in fuse_dlm_cache.c.
+ */
 struct fuse_dlm_cache {
-	/* Lock protecting the tree */
+	/* Lock protecting the tree and the pending list */
 	struct rw_semaphore lock;
-	/* Interval tree of locked ranges */
+	/* Interval tree of recorded ranges, granted or revoked */
 	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.
+	 * FUSE_DLM_WB_LOCK requests in flight (REQUESTED, or REVOKED once
+	 * a revoke has overlapped one).  Owned by the queueing thread; the
+	 * revoke paths mark them only.
 	 */
-	uint64_t revoke_gen;
+	struct list_head pending;
 };
 
 /* Initialize a page cache lock manager */
@@ -51,13 +60,29 @@ 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);
+/*
+ * Publish a FUSE_DLM_WB_LOCK for [start, end] before it is sent, so a
+ * revoke processed while the reply is on the wire can mark it.  @req is
+ * caller-owned storage, live until the matching commit or abort.  The
+ * mode is not recorded until the grant is, so only the commit takes it.
+ */
+void fuse_dlm_request_begin(struct fuse_inode *inode,
+			    struct fuse_dlm_range *req, uint64_t start,
+			    uint64_t end);
 
-/* Sample the revocation generation (see fuse_dlm_lock_range_gen()) */
-uint64_t fuse_dlm_revoke_gen(struct fuse_inode *inode);
+/*
+ * Retire @req and record the grant [start, end] as one step under the
+ * cache lock.  -EAGAIN means a revoke overlapped @req in flight and
+ * nothing was recorded; the caller must request again.  @req is retired
+ * either way.
+ */
+int fuse_dlm_request_commit(struct fuse_inode *inode,
+			    struct fuse_dlm_range *req, uint64_t start,
+			    uint64_t end, enum fuse_page_lock_mode mode);
+
+/* Retire @req without recording anything */
+void fuse_dlm_request_abort(struct fuse_inode *inode,
+			    struct fuse_dlm_range *req);
 
 /* Unlock a range of pages */
 int fuse_dlm_unlock_range(struct fuse_inode *inode, uint64_t start,
@@ -74,6 +99,17 @@ 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);
 
+/*
+ * The page cache under [start, end] is gone: free the revoked ranges over
+ * it.  The caller must have established the range really is empty.
+ */
+void fuse_dlm_ranges_dropped(struct fuse_inode *inode, uint64_t start,
+			     uint64_t end);
+
+/* Hold [start, end] again so writeback can send what it found revoked */
+int fuse_dlm_regrant_range(struct fuse_file *ff, struct inode *inode,
+			   uint64_t start, uint64_t end);
+
 /* This is the interface to the filesystem */
 int fuse_get_dlm_lock(struct file *file, loff_t offset,
 		      size_t length, enum fuse_page_lock_mode mode);
diff --git a/fs/fuse/fuse_i.h b/fs/fuse/fuse_i.h
index 8dc81ff1f27452..1cf4974ddda3ea 100644
--- a/fs/fuse/fuse_i.h
+++ b/fs/fuse/fuse_i.h
@@ -189,31 +189,6 @@ struct fuse_inode {
 			/* dlm locked areas we have sent lock requests for */
 			struct fuse_dlm_cache dlm_locked_areas;
 
-			/*
-			 * Per-inode read/write coherency gate for the
-			 * forced-direct-IO feature.  Cache-serving buffered reads
-			 * and buffered writes hold it for read; being a
-			 * percpu_rw_semaphore the read side is per-CPU cheap and
-			 * scales on a shared file.  The NOTIFY invalidate
-			 * (fuse_reverse_inval_inode()) holds it for write, which
-			 * BLOCKS so the coherency notify has priority: it fences
-			 * cache-serving reads (and buffered writes) out for the
-			 * whole invalidate, so no folio a remote modify has
-			 * superseded is ever handed back.
-			 *
-			 * The write side may run on the server thread delivering
-			 * the notify, so a blocking writer is safe only under a
-			 * server that services request replies on threads other
-			 * than the one delivering the notify (see the NOTIFY site).
-			 *
-			 * Allocated out of line only for writeback+dlm regular
-			 * files (it shares storage with the readdir-cache union
-			 * arm); NULL on other mounts and on allocation failure,
-			 * where the gate is inactive and the invalidate falls back
-			 * to best-effort.
-			 */
-			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
@@ -226,6 +201,15 @@ struct fuse_inode {
 			 */
 			unsigned long notify_stamp;
 			unsigned int notify_interval_ewma;
+
+			/*
+			 * Index most recently claimed by the inode wide read
+			 * lookahead, so several readers walking one file do
+			 * not each populate the same range.  Protected by
+			 * fi->lock; regular files only (shares the
+			 * readdir-cache union arm).
+			 */
+			pgoff_t ra_lookahead;
 		};
 
 		/* readdir cache (directory only) */
@@ -1262,6 +1246,9 @@ struct fuse_file *fuse_file_alloc(struct fuse_mount *fm, bool release);
 void fuse_file_free(struct fuse_file *ff);
 int fuse_finish_open(struct inode *inode, struct file *file);
 
+/* Drop the page cache an open must not keep; true if it came out empty */
+bool fuse_open_drop_cache(struct inode *inode);
+
 void fuse_sync_release(struct fuse_inode *fi, struct fuse_file *ff,
 		       unsigned int flags);
 
diff --git a/fs/fuse/inode.c b/fs/fuse/inode.c
index 69694ed4ee9105..bb869e68619a83 100644
--- a/fs/fuse/inode.c
+++ b/fs/fuse/inode.c
@@ -208,23 +208,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)
@@ -546,16 +529,36 @@ static u32 fuse_attr_cache_mask(struct inode *inode, struct fuse_attr *attr,
 	    !S_ISREG(inode->i_mode))
 		return cache_mask;
 
-	if (!fuse_dlm_write_grant_exists(fi))
+	/*
+	 * A dirty mapping keeps the local attributes authoritative even
+	 * when no grant is recorded: a fault dirties pages under a
+	 * page-mkwrite lock that is never recorded, and a truncate revokes
+	 * the tail grants itself while cached writes above the new size
+	 * are still waiting for writeback.
+	 */
+	if (!fuse_dlm_write_grant_exists(fi) &&
+	    !mapping_tagged(inode->i_mapping, PAGECACHE_TAG_DIRTY) &&
+	    !mapping_tagged(inode->i_mapping, PAGECACHE_TAG_WRITEBACK))
 		return cache_mask;
 
 	if (mapping_tagged(inode->i_mapping, PAGECACHE_TAG_DIRTY) ||
 	    mapping_tagged(inode->i_mapping, PAGECACHE_TAG_WRITEBACK))
 		cache_mask |= STATX_MTIME | STATX_CTIME;
 
+	/*
+	 * The local size stays authoritative while the extension is
+	 * covered by a write grant, and also while anything in
+	 * [attr->size, size) is dirty or under writeback: those bytes
+	 * exist only here, and taking the server's smaller size would
+	 * truncate them away before they are ever sent.  The grant check
+	 * alone misses them, because a page-mkwrite grant is never
+	 * recorded and a local truncate revokes its own tail grants.
+	 */
 	if (have_size && size > (loff_t) attr->size &&
-	    fuse_dlm_lock_is_held(fi, attr->size, size - attr->size,
-				  FUSE_PAGE_LOCK_WRITE))
+	    (fuse_dlm_lock_is_held(fi, attr->size, size - attr->size,
+				   FUSE_PAGE_LOCK_WRITE) ||
+	     filemap_range_needs_writeback(inode->i_mapping, attr->size,
+					   size - 1)))
 		cache_mask |= STATX_SIZE;
 
 	return cache_mask;
@@ -643,6 +646,27 @@ static void fuse_change_attributes_i(struct inode *inode, struct fuse_attr *attr
 
 		if (inval)
 			invalidate_inode_pages2(inode->i_mapping);
+
+		/*
+		 * The DLM record has to follow the cache out, as on every
+		 * other path that drops it.  A revoked range exists only to
+		 * make writeback take the grant again before sending the
+		 * folios under it, so once they are gone it describes
+		 * nothing and would sit in the tree unfreed.  The pages above
+		 * the new size are gone unconditionally; the rest only when
+		 * the invalidate really emptied the mapping, so a folio that
+		 * survived (or was faulted back) keeps its record.
+		 */
+		if (fc->dlm && fc->writeback_cache) {
+			if (have_size && oldsize != attr->size)
+				fuse_dlm_ranges_dropped(fi,
+							PAGE_ALIGN(attr->size),
+							U64_MAX);
+			if (inval &&
+			    !filemap_range_has_page(inode->i_mapping, 0,
+						    LLONG_MAX))
+				fuse_dlm_ranges_dropped(fi, 0, U64_MAX);
+		}
 	}
 
 	if (IS_ENABLED(CONFIG_FUSE_DAX))
@@ -890,22 +914,25 @@ static void fuse_dlm_revoke_inval_range(struct fuse_inode *fi, loff_t offset,
  * Drop a page-cache range on behalf of a NOTIFY invalidate.
  *
  * invalidate_inode_pages2_range() waits out folios under writeback and
- * launders dirty ones, both of which need a FUSE_WRITE reply.  While
- * writepages are frozen (fuse_set_nowrite(): truncate, O_TRUNC open, fsync,
- * pre-SETATTR flush) no reply can arrive, because fuse_flush_writepages()
- * parks the request on fi->queued_writes until fuse_release_nowrite().  A
- * server that revokes from inside the handler it is revoking for then
- * deadlocks against its own reply.  fuse_do_setattr() states the same rule
- * for its own invalidate.
+ * launders dirty ones, both of which need a FUSE_WRITE reply.  It is only
+ * needed when the range can hold data the server has not seen.
+ *
+ * @may_be_dirty false says it cannot.  invalidate_mapping_pages() then
+ * drops the same folios without ever waiting for the server.
  *
- * So while frozen use invalidate_mapping_pages(), which skips dirty and
- * under-writeback folios and never blocks.  The stale clean folios still
- * go, and the freezes that span a request drop the cache themselves once
- * they complete: fuse_do_setattr() invalidates the mapping after releasing
- * the freeze, the O_TRUNC open path calls truncate_pagecache().
+ * The same substitution is forced while writepages are frozen
+ * (fuse_set_nowrite(): truncate, O_TRUNC open, fsync, pre-SETATTR flush),
+ * where no reply can arrive because fuse_flush_writepages() parks the
+ * request on fi->queued_writes until fuse_release_nowrite().  A server that
+ * revokes from inside the handler it is revoking for would otherwise
+ * deadlock against its own reply.  fuse_do_setattr() states the same rule
+ * for its own invalidate.  There the dirty folios are left behind, and the
+ * freezes that span a request drop the cache themselves once they complete:
+ * fuse_do_setattr() invalidates the mapping after releasing the freeze, the
+ * O_TRUNC open path calls truncate_pagecache().
  */
 static void fuse_notify_invalidate_range(struct inode *inode, pgoff_t start,
-					 pgoff_t end)
+					 pgoff_t end, bool may_be_dirty)
 {
 	struct fuse_inode *fi = get_fuse_inode(inode);
 	bool frozen;
@@ -914,7 +941,7 @@ static void fuse_notify_invalidate_range(struct inode *inode, pgoff_t start,
 	frozen = fi->writectr < 0;
 	spin_unlock(&fi->lock);
 
-	if (frozen)
+	if (frozen || !may_be_dirty)
 		invalidate_mapping_pages(inode->i_mapping, start, end);
 	else
 		invalidate_inode_pages2_range(inode->i_mapping, start, end);
@@ -923,11 +950,14 @@ 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_inode *fi;
 	struct inode *inode;
+	uint64_t pg_first;
+	uint64_t pg_last;
+	loff_t end_byte;
 	pgoff_t pg_start;
 	pgoff_t pg_end;
+	bool tracked;
 
 	inode = fuse_ilookup(fc, nodeid, NULL);
 	if (!inode)
@@ -960,24 +990,30 @@ int fuse_reverse_inval_inode(struct fuse_conn *fc, u64 nodeid,
 		else
 			pg_end = (offset + len - 1) >> PAGE_SHIFT;
 
+		/*
+		 * Byte bounds of the same region, and the page aligned form
+		 * the DLM record is told about.  A grant is recorded page
+		 * aligned, so the range handed to it has to cover whole
+		 * pages or the record would keep a range the page cache no
+		 * longer backs.
+		 */
+		end_byte = len <= 0 ? LLONG_MAX : offset + len - 1;
+		pg_first = (uint64_t)offset & PAGE_MASK;
+		pg_last = len <= 0 ? U64_MAX :
+			  (((uint64_t)offset + len - 1) | (PAGE_SIZE - 1));
+
 		/*
 		 * 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.
+		 *    read returns a folio the remote modify has superseded.
+		 *    Nothing is fenced out for it.  A read racing the drop
+		 *    either misses and refetches or returns data that was
+		 *    current when it was copied.  A write racing it is caught
+		 *    on the way out instead: this revoke marks the range rather
+		 *    than forgetting it, and writeback holds the range again
+		 *    before sending anything it finds marked that way.
 		 *
 		 * 2. Latch.  Keep a moving average (fuse_notify_inval_hot(), under
 		 *    fi->lock, updated for every data invalidation) of how fast
@@ -995,22 +1031,19 @@ 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 the latch exist only for writeback+dlm
+		 * regular files; elsewhere there is no record to consult and
+		 * the range is dropped as it always was.  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;
+		tracked = S_ISREG(inode->i_mode) && fc->writeback_cache &&
+			  fc->dlm && !FUSE_IS_DAX(inode) &&
+			  !fuse_inode_backing(fi);
 
-		if (wb_sem) {
+		if (tracked) {
 			bool hot, has_writer, latched = false;
+			bool may_be_dirty, has_pages;
 
 			spin_lock(&fi->lock);
 			hot = fuse_notify_inval_hot(fi);
@@ -1018,22 +1051,42 @@ int fuse_reverse_inval_inode(struct fuse_conn *fc, u64 nodeid,
 			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.
+			 * What this notify has to do.  Nothing cached in the
+			 * range means the drop is a no-op and the revoke is
+			 * the whole job.  Otherwise the page cache says
+			 * whether the drop has to launder, which is what
+			 * makes it wait for a FUSE_WRITE reply.
 			 */
-			percpu_down_write(wb_sem);
+			has_pages = filemap_range_has_page(inode->i_mapping,
+							   offset, end_byte);
+			may_be_dirty = filemap_range_needs_writeback(
+					inode->i_mapping, offset, end_byte);
 
 			/*
-			 * 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.
+			 * Put unwritten data on the server while the grant
+			 * still covers it, rather than leaving it to the drop
+			 * below.  After the revoke writeback would have to
+			 * take the range again to send those bytes: a DLM
+			 * round trip from inside the handler the server is
+			 * waiting on.  do_writepages() runs in this context,
+			 * so the grant is asked for before the revoke.
+			 *
+			 * Waited out here rather than left to the drop, which
+			 * launders when the range may be dirty and so waits
+			 * for these same replies.  One explicit wait, before
+			 * the revoke, and the drop then finds nothing under
+			 * writeback to block on.  Either way a server that
+			 * revokes from a thread it also needs to answer
+			 * FUSE_WRITE on deadlocks here, the same contract
+			 * fuse_notify_invalidate_range() states for a frozen
+			 * inode.  The error is left to the mapping, where
+			 * fsync collects it.
 			 */
-			if (fc->dlm && fc->writeback_cache)
-				fuse_dlm_revoke_inval_range(fi, offset, len);
+			if (has_pages && may_be_dirty)
+				filemap_write_and_wait_range(inode->i_mapping,
+							     offset, end_byte);
+
+			fuse_dlm_revoke_inval_range(fi, offset, len);
 
 			if (enable_notify_dio && hot && has_writer &&
 			    !mapping_mapped(inode->i_mapping) &&
@@ -1049,27 +1102,43 @@ int fuse_reverse_inval_inode(struct fuse_conn *fc, u64 nodeid,
 			/*
 			 * Latched: drop the whole mapping (dirty folios
 			 * outside the notified range would be invisible to
-			 * the forced direct reads).  Otherwise just the
-			 * notified range.
+			 * the forced direct reads), and nothing was asked
+			 * about the rest of the file, so launder.  Otherwise
+			 * just the notified range, and only if anything is
+			 * cached there.
 			 */
 			if (fuse_inode_force_dio(inode))
-				fuse_notify_invalidate_range(inode, 0, -1);
-			else
+				fuse_notify_invalidate_range(inode, 0, -1, true);
+			else if (has_pages)
 				fuse_notify_invalidate_range(inode, pg_start,
-							     pg_end);
+							     pg_end,
+							     may_be_dirty);
 
-			percpu_up_write(wb_sem);
+			/*
+			 * A revoked range exists to describe page cache
+			 * dirtied before the grant went; with that cache gone
+			 * it has nothing left to say.  Only when it really
+			 * went: an invalidate can leave a busy folio behind,
+			 * and that folio still needs its record.
+			 */
+			if (has_pages &&
+			    !filemap_range_has_page(inode->i_mapping, offset,
+						    end_byte))
+				fuse_dlm_ranges_dropped(fi, pg_first, pg_last);
 
 			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. */
+			/*
+			 * No record on this inode (DAX, backing, non-regular,
+			 * or no DLM), so assume the range can hold unwritten
+			 * data and drop it 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);
+			fuse_notify_invalidate_range(inode, pg_start, pg_end,
+						     true);
 		}
 	}
 	iput(inode);
@@ -1931,6 +2000,16 @@ static void process_init_reply(struct fuse_mount *fm, struct fuse_args *args,
 		else
 			fm->sb->s_bdi->ra_pages =
 				min(fm->sb->s_bdi->ra_pages, ra_pages);
+		/*
+		 * ra_pages caps speculative readahead; io_pages is what the
+		 * device can carry in one go, and mm uses it to let a read
+		 * larger than the readahead window through at the size it
+		 * was actually asked for instead of clamping it to ra_pages.
+		 * For fuse that size is a request, so say so -- otherwise a
+		 * large explicit read or a FADV_WILLNEED gets chopped up by a
+		 * window that was only ever meant to bound guessing.
+		 */
+		fm->sb->s_bdi->io_pages = fc->max_pages;
 		fc->minor = arg->minor;
 		fc->max_write = arg->minor < 5 ? 4096 : arg->max_write;
 		fc->max_write = max_t(unsigned, 4096, fc->max_write);