20
20
package org .dcache .nfs .v4 ;
21
21
22
22
import com .google .common .util .concurrent .Striped ;
23
+
24
+ import java .io .IOException ;
23
25
import java .util .ArrayList ;
24
26
import java .util .Collection ;
25
27
import java .util .Iterator ;
30
32
import java .util .stream .Collectors ;
31
33
import org .dcache .nfs .ChimeraNFSException ;
32
34
import org .dcache .nfs .status .BadStateidException ;
35
+ import org .dcache .nfs .status .DelayException ;
33
36
import org .dcache .nfs .status .InvalException ;
34
37
import org .dcache .nfs .status .ShareDeniedException ;
38
+ import org .dcache .nfs .status .StaleException ;
35
39
import org .dcache .nfs .v4 .xdr .nfs4_prot ;
40
+ import org .dcache .nfs .v4 .xdr .nfs_fh4 ;
41
+ import org .dcache .nfs .v4 .xdr .open_delegation_type4 ;
36
42
import org .dcache .nfs .v4 .xdr .stateid4 ;
37
43
import org .dcache .nfs .vfs .Inode ;
38
44
import org .dcache .nfs .util .Opaque ;
@@ -55,6 +61,11 @@ public class FileTracker {
55
61
private final Striped <Lock > filesLock = Striped .lock (Runtime .getRuntime ().availableProcessors ()*4 );
56
62
private final Map <Opaque , List <OpenState >> files = new ConcurrentHashMap <>();
57
63
64
+ /**
65
+ * Delegation records associated with open files.
66
+ */
67
+ private final Map <Opaque , List <DelegationState >> delegations = new ConcurrentHashMap <>();
68
+
58
69
private static class OpenState {
59
70
60
71
private final NFS4Client client ;
@@ -118,6 +129,27 @@ public NFS4Client getClient() {
118
129
}
119
130
}
120
131
132
+ /**
133
+ * Record associated with open-delegation.
134
+ * @param client
135
+ * @param stateid
136
+ * @param delegationType
137
+ */
138
+ record DelegationState (NFS4Client client , stateid4 openStateId , stateid4 stateid , int delegationType ) {
139
+
140
+ }
141
+
142
+ /**
143
+ * Record associated with an open file.
144
+ *
145
+ * @param openStateId
146
+ * @param delegationStateId
147
+ * @param hasDelegation
148
+ */
149
+ public record OpenRecord (stateid4 openStateId , stateid4 delegationStateId , boolean hasDelegation ) {
150
+
151
+ }
152
+
121
153
/**
122
154
* Add a new open to the list of open files. If provided {@code shareAccess}
123
155
* and {@code shareDeny} conflicts with existing opens, @{link ShareDeniedException}
@@ -127,11 +159,14 @@ public NFS4Client getClient() {
127
159
* @param inode of opened file.
128
160
* @param shareAccess type of access required.
129
161
* @param shareDeny type of access to deny others.
130
- * @return a snapshot of the stateid associated with open.
162
+ * @return a snapshot of an OpenRecord associated with open.
131
163
* @throws ShareDeniedException if share reservation conflicts with an existing open.
132
164
* @throws ChimeraNFSException
133
165
*/
134
- public stateid4 addOpen (NFS4Client client , StateOwner owner , Inode inode , int shareAccess , int shareDeny ) throws ChimeraNFSException {
166
+ public OpenRecord addOpen (NFS4Client client , StateOwner owner , Inode inode , int shareAccess , int shareDeny ) throws ChimeraNFSException {
167
+
168
+ boolean wantReadDelegation = (shareAccess & nfs4_prot .OPEN4_SHARE_ACCESS_WANT_READ_DELEG ) != 0 ;
169
+ boolean wantWriteDelegation = (shareAccess & nfs4_prot .OPEN4_SHARE_ACCESS_WANT_WRITE_DELEG ) != 0 ;
135
170
136
171
Opaque fileId = new Opaque (inode .getFileId ());
137
172
Lock lock = filesLock .get (fileId );
@@ -170,7 +205,43 @@ public stateid4 addOpen(NFS4Client client, StateOwner owner, Inode inode, int sh
170
205
171
206
os .stateid .seqid ++;
172
207
//we need to return copy to avoid modification by concurrent opens
173
- return new stateid4 (os .stateid .other , os .stateid .seqid );
208
+ var openStateid = new stateid4 (os .stateid .other , os .stateid .seqid );
209
+ return new OpenRecord (openStateid , null , false );
210
+ }
211
+ }
212
+
213
+ /*
214
+ * REVISIT: currently only read-delegations are supported
215
+ */
216
+ var existingDelegations = delegations .get (fileId );
217
+
218
+ /*
219
+ * delegation is possible if:
220
+ * - client has a callback channel
221
+ * - client does not have a delegation for this file
222
+ * - no other open has write access
223
+ */
224
+ boolean canDelegate = client .getCB () != null &&
225
+ (existingDelegations == null || existingDelegations .stream ().noneMatch (d -> d .client ().getId () == client .getId ())) &&
226
+ opens .stream ().noneMatch (os -> (os .shareAccess & nfs4_prot .OPEN4_SHARE_ACCESS_WRITE ) != 0 );
227
+
228
+ // recall any read delegations if write
229
+ if ((existingDelegations != null ) && (shareAccess & nfs4_prot .OPEN4_SHARE_ACCESS_WRITE ) != 0 ) {
230
+
231
+ // REVISIT: usage of Stream#peek is an anti-pattern
232
+ boolean haveRecalled = existingDelegations .stream ()
233
+ .filter (d -> client .isLeaseValid ())
234
+ .peek (d -> {
235
+ try {
236
+ d .client ().getCB ()
237
+ .cbDelegationRecall (new nfs_fh4 (inode .toNfsHandle ()), d .stateid (), false );
238
+ } catch (IOException e ) {
239
+ // ignore
240
+ }
241
+ }).count () > 0 ;
242
+
243
+ if (haveRecalled ) {
244
+ throw new DelayException ("Recalling read delegations" );
174
245
}
175
246
}
176
247
@@ -180,8 +251,21 @@ public stateid4 addOpen(NFS4Client client, StateOwner owner, Inode inode, int sh
180
251
opens .add (openState );
181
252
state .addDisposeListener (s -> removeOpen (inode , stateid ));
182
253
stateid .seqid ++;
254
+
183
255
//we need to return copy to avoid modification by concurrent opens
184
- return new stateid4 (stateid .other , stateid .seqid );
256
+ var openStateid = new stateid4 (stateid .other , stateid .seqid );
257
+
258
+ // REVISIT: currently only read-delegations are supported
259
+ if (wantReadDelegation && canDelegate ) {
260
+ // REVISIT: currently only read-delegations are supported
261
+ stateid4 delegationStateid = client .createState (state .getStateOwner (), state ).stateid ();
262
+ delegations .computeIfAbsent (fileId , x -> new ArrayList <>(1 ))
263
+ .add (new DelegationState (client , openStateid , delegationStateid , open_delegation_type4 .OPEN_DELEGATE_READ ));
264
+ return new OpenRecord (openStateid , delegationStateid , true );
265
+ } else {
266
+ //we need to return copy to avoid modification by concurrent opens
267
+ return new OpenRecord (openStateid , null , false );
268
+ }
185
269
} finally {
186
270
lock .unlock ();
187
271
}
@@ -241,6 +325,42 @@ public stateid4 downgradeOpen(NFS4Client client, stateid4 stateid, Inode inode,
241
325
}
242
326
}
243
327
328
+ /**
329
+ * Return delegation for the given file
330
+ * @param client nfs client who returns the delegation.
331
+ * @param stateid delegation stateid
332
+ * @param inode the inode of the delegated file.
333
+ */
334
+ public void delegationReturn (NFS4Client client , stateid4 stateid , Inode inode )
335
+ throws StaleException {
336
+
337
+ Opaque fileId = new Opaque (inode .getFileId ());
338
+ Lock lock = filesLock .get (fileId );
339
+ lock .lock ();
340
+ try {
341
+
342
+ var fileDelegations = delegations .get (fileId );
343
+ if (fileDelegations == null ) {
344
+ throw new StaleException ("no delegation found" );
345
+ }
346
+
347
+ DelegationState delegation = fileDelegations .stream ()
348
+ .filter (d -> d .client ().getId () == client .getId ())
349
+ .filter (d -> d .stateid ().equals (stateid ))
350
+ .findFirst ()
351
+ .orElseThrow (StaleException ::new );
352
+
353
+ fileDelegations .remove (delegation );
354
+ if (fileDelegations .isEmpty ()) {
355
+ delegations .remove (fileId );
356
+ }
357
+
358
+ } finally {
359
+ lock .unlock ();
360
+ }
361
+ }
362
+
363
+
244
364
/**
245
365
* Remove an open from the list.
246
366
* @param inode of the opened file
@@ -271,6 +391,23 @@ void removeOpen(Inode inode, stateid4 stateid) {
271
391
files .remove (fileId );
272
392
}
273
393
}
394
+
395
+ var existingDelegations = delegations .get (fileId );
396
+ if (existingDelegations != null ) {
397
+ Iterator <DelegationState > dsi = existingDelegations .iterator ();
398
+ while (dsi .hasNext ()) {
399
+ stateid4 os = dsi .next ().openStateId ();
400
+ if (os .equals (stateid )) {
401
+ dsi .remove ();
402
+ break ;
403
+ }
404
+ }
405
+
406
+ if (existingDelegations .isEmpty ()) {
407
+ delegations .remove (fileId );
408
+ }
409
+ }
410
+
274
411
} finally {
275
412
lock .unlock ();
276
413
}
0 commit comments