nfs41: refactor and fix read delegation handling

kofemann · kofemann · commit 532f71ee3e4c · 2025-05-09T17:52:54.000+02:00
Motivation:
current delegation code has following issues:
- open squashing happens before recall of conflicting delegation
- client can conflict with himself
- invalidate delegations on close
- uses Stream#peak is an anti-pattern to recall delegations
- delegation stateids not disposed on return

Modification:
- Move delegation conflict check ahead of open processing
- ignore conflicts from the same client
- rename canDelegate to canDelegateRead (just in case we are write
  delegations too)
- don't dispose delegation on close (but on delegation return)
- replace peak+count with reduce

Result:
better spec compliance, clean code.

Acked-by: Marina Sahakyan
Target: master
diff --git a/core/src/main/java/org/dcache/nfs/v4/FileTracker.java b/core/src/main/java/org/dcache/nfs/v4/FileTracker.java
@@ -42,12 +42,16 @@
 import org.dcache.nfs.v4.xdr.stateid4;
 import org.dcache.nfs.vfs.Inode;
 import org.dcache.nfs.util.Opaque;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * A class which tracks open files.
  */
 public class FileTracker {
 
+    public static final Logger LOG = LoggerFactory.getLogger(FileTracker.class);
+
     /*
      * we use {@link Striped} locks here to split synchronized block on open files
      * into multiple partitions to increase concurrency, while guaranteeing atomicity
@@ -135,7 +139,7 @@ public NFS4Client getClient() {
      * @param stateid
      * @param delegationType
      */
-    record DelegationState(NFS4Client client, stateid4 openStateId, stateid4 stateid, int delegationType) {
+    record DelegationState(NFS4Client client, NFS4State delegationStateid, int delegationType) {
 
     }
 
@@ -187,6 +191,43 @@ public OpenRecord addOpen(NFS4Client client, StateOwner owner, Inode inode, int
                     throw new ShareDeniedException("Conflicting share");
             }
 
+            /*
+             * REVISIT: currently only read-delegations are supported
+             */
+            var existingDelegations = delegations.get(fileId);
+
+            /*
+             * delegation is possible if:
+             * - client has a callback channel
+             * - client does not have a delegation for this file
+             * - no other open has write access
+             */
+            boolean canDelegateRead = client.getCB() != null &&
+                    (existingDelegations == null || existingDelegations.stream().noneMatch(d -> d.client().getId() == client.getId())) &&
+                    opens.stream().noneMatch(os -> (os.shareAccess & nfs4_prot.OPEN4_SHARE_ACCESS_WRITE) != 0);
+
+            // recall any read delegations if write
+            if ((existingDelegations != null) && (shareAccess & nfs4_prot.OPEN4_SHARE_ACCESS_WRITE) != 0) {
+                var fh = new nfs_fh4(inode.toNfsHandle());
+                int recalledDelegations = existingDelegations.stream()
+                        .filter(d -> d.client().isLeaseValid())
+                        .filter(d -> !d.client().getId().equals(client.getId()))
+                        .reduce(0, (c, d) -> {
+                            try {
+                                d.client().getCB().cbDelegationRecall(fh, d.delegationStateid().stateid(), false);
+                                return c + 1;
+                            } catch (IOException e) {
+                                LOG.warn("Failed to recall delegation from {} : {}", d.client(), e.toString());
+                                d.delegationStateid().disposeIgnoreFailures();
+                                return c;
+                            }
+                        }, Integer::sum);
+
+                if (recalledDelegations > 0) {
+                    throw new DelayException("Recalling read delegations");
+                }
+            }
+
             // if there is another open from the same client we must merge
             // access mode and return the same stateid as required by rfc5661#18.16.3
 
@@ -210,41 +251,6 @@ public OpenRecord addOpen(NFS4Client client, StateOwner owner, Inode inode, int
                 }
             }
 
-            /*
-             * REVISIT: currently only read-delegations are supported
-             */
-            var existingDelegations = delegations.get(fileId);
-
-            /*
-             * delegation is possible if:
-             * - client has a callback channel
-             * - client does not have a delegation for this file
-             * - no other open has write access
-             */
-            boolean canDelegate = client.getCB() != null &&
-                  (existingDelegations == null || existingDelegations.stream().noneMatch(d -> d.client().getId() == client.getId())) &&
-                  opens.stream().noneMatch(os -> (os.shareAccess & nfs4_prot.OPEN4_SHARE_ACCESS_WRITE) != 0);
-
-            // recall any read delegations if write
-            if ((existingDelegations != null) && (shareAccess & nfs4_prot.OPEN4_SHARE_ACCESS_WRITE) != 0) {
-
-                // REVISIT: usage of Stream#peek is an anti-pattern
-                boolean haveRecalled = existingDelegations.stream()
-                      .filter(d -> client.isLeaseValid())
-                      .peek(d -> {
-                          try {
-                              d.client().getCB()
-                                    .cbDelegationRecall(new nfs_fh4(inode.toNfsHandle()), d.stateid(), false);
-                          } catch (IOException e) {
-                              // ignore
-                          }
-                      }).count() > 0;
-
-                if (haveRecalled) {
-                    throw new DelayException("Recalling read delegations");
-                }
-            }
-
             NFS4State state = client.createState(owner);
             stateid = state.stateid();
             OpenState openState = new OpenState(client, owner, stateid, shareAccess, shareDeny);
@@ -256,12 +262,12 @@ public OpenRecord addOpen(NFS4Client client, StateOwner owner, Inode inode, int
             var openStateid = new stateid4(stateid.other, stateid.seqid);
 
             // REVISIT: currently only read-delegations are supported
-            if (wantReadDelegation && canDelegate) {
+            if (wantReadDelegation && canDelegateRead) {
                 // REVISIT: currently only read-delegations are supported
-                stateid4 delegationStateid = client.createState(state.getStateOwner(), state).stateid();
+                var delegationStateid = client.createState(state.getStateOwner(), state);
                 delegations.computeIfAbsent(fileId, x -> new ArrayList<>(1))
-                        .add(new DelegationState(client, openStateid, delegationStateid, open_delegation_type4.OPEN_DELEGATE_READ));
-                return new OpenRecord(openStateid, delegationStateid, true);
+                        .add(new DelegationState(client, delegationStateid, open_delegation_type4.OPEN_DELEGATE_READ));
+                return new OpenRecord(openStateid, delegationStateid.stateid(), true);
             } else {
                 //we need to return copy to avoid modification by concurrent opens
                 return new OpenRecord(openStateid, null, false);
@@ -332,7 +338,7 @@ public stateid4 downgradeOpen(NFS4Client client, stateid4 stateid, Inode inode,
      * @param inode the inode of the delegated file.
      */
     public void delegationReturn(NFS4Client client, stateid4 stateid, Inode inode)
-          throws StaleException {
+            throws ChimeraNFSException {
 
         Opaque fileId = new Opaque(inode.getFileId());
         Lock lock = filesLock.get(fileId);
@@ -345,11 +351,12 @@ public void delegationReturn(NFS4Client client, stateid4 stateid, Inode inode)
             }
 
             DelegationState delegation = fileDelegations.stream()
-                    .filter(d -> d.client().getId() == client.getId())
-                    .filter(d -> d.stateid().equals(stateid))
+                    .filter(d -> d.client().getId().equals(client.getId()))
+                    .filter(d -> d.delegationStateid().stateid().equals(stateid))
                     .findFirst()
                     .orElseThrow(StaleException::new);
 
+            delegation.delegationStateid().tryDispose();
             fileDelegations.remove(delegation);
             if (fileDelegations.isEmpty()) {
                 delegations.remove(fileId);
@@ -392,22 +399,6 @@ void removeOpen(Inode inode, stateid4 stateid) {
                 }
             }
 
-            var existingDelegations = delegations.get(fileId);
-            if (existingDelegations != null) {
-                Iterator<DelegationState> dsi = existingDelegations.iterator();
-                while (dsi.hasNext()) {
-                    stateid4 os = dsi.next().openStateId();
-                    if (os.equals(stateid)) {
-                        dsi.remove();
-                        break;
-                    }
-                }
-
-                if (existingDelegations.isEmpty()) {
-                    delegations.remove(fileId);
-                }
-            }
-
         } finally {
             lock.unlock();
         }
diff --git a/core/src/test/java/org/dcache/nfs/v4/FileTrackerTest.java b/core/src/test/java/org/dcache/nfs/v4/FileTrackerTest.java
@@ -238,24 +238,25 @@ public void shouldNotReadDelegation() throws Exception {
     @Test
     public void shouldReCallReadDelegationOnConflict() throws Exception {
 
-        NFS4Client client = createClient(sh);
+        NFS4Client client1 = createClient(sh);
+        NFS4Client client2 = createClient(sh);
 
-        StateOwner stateOwner1 = client.getOrCreateOwner("client1".getBytes(StandardCharsets.UTF_8), new seqid4(0));
-        StateOwner stateOwner2 = client.getOrCreateOwner("client2".getBytes(StandardCharsets.UTF_8), new seqid4(0));
+        StateOwner stateOwner1 = client1.getOrCreateOwner("client1".getBytes(StandardCharsets.UTF_8), new seqid4(0));
+        StateOwner stateOwner2 = client2.getOrCreateOwner("client2".getBytes(StandardCharsets.UTF_8), new seqid4(0));
 
         nfs_fh4 fh = generateFileHandle();
         Inode inode = Inode.forFile(fh.value);
 
-        var openRecord1 = tracker.addOpen(client, stateOwner1, inode, OPEN4_SHARE_ACCESS_READ | nfs4_prot.OPEN4_SHARE_ACCESS_WANT_READ_DELEG, 0);
+        var openRecord1 = tracker.addOpen(client1, stateOwner1, inode, OPEN4_SHARE_ACCESS_READ | nfs4_prot.OPEN4_SHARE_ACCESS_WANT_READ_DELEG, 0);
         try {
-            var openRecord2 = tracker.addOpen(client, stateOwner2, inode, OPEN4_SHARE_ACCESS_WRITE,
+            var openRecord2 = tracker.addOpen(client2, stateOwner2, inode, OPEN4_SHARE_ACCESS_WRITE,
                   0);
             fail("Delay exception expected");
         } catch (DelayException e) {
             // expected
         }
 
-        verify(client.getCB()).cbDelegationRecall(any(), any(), anyBoolean());
+        verify(client1.getCB()).cbDelegationRecall(any(), any(), anyBoolean());
     }
 
     @Test
