pnfs: track last_offset for layoutcommit

iozone tests were failing intermittently against emc-2 due to out-of-order LAYOUTCOMMITs that both specified new_last_offset

nfs41_open_state now maintains a cached value of the last_offset, and avoids sending it with LAYOUTCOMMIT unless the new last_offset is greater than the cached offset.  this cached value is initialized on open/delegation, and updated on setattr for size

Signed-off-by: Casey Bodley <cbodley@citi.umich.edu>

daemon/nfs41.h

@@ -135,6 +135,7 @@ typedef struct __nfs41_open_state {
     LONG ref_count;
     uint32_t share_access;
     uint32_t share_deny;
+    uint64_t pnfs_last_offset; /* for layoutcommit */
 
     struct {
         nfs41_delegation_state *state;

daemon/open.c

@@ -239,6 +239,8 @@ static int open_or_delegate(
     if (status)
         status = do_open(state, create, createhow, mode, try_recovery, info);
 
+    state->pnfs_last_offset = info->size ? info->size - 1 : 0;
+
     /* register the client's open state on success */
     if (status == NFS4_OK)
         client_state_add(state);

daemon/pnfs_io.c

@@ -475,6 +475,44 @@ out:
     return status;
 }
 
+static enum pnfs_status layout_commit(
+    IN nfs41_open_state *state,
+    IN pnfs_layout_state *layout,
+    IN uint64_t offset,
+    IN uint64_t length)
+{
+    stateid4 layout_stateid;
+    uint64_t last_offset = offset + length - 1;
+    uint64_t *new_last_offset = NULL;
+    enum nfsstat4 nfsstat;
+    enum pnfs_status status = PNFS_SUCCESS;
+
+    AcquireSRWLockExclusive(&state->lock);
+    /* if this is past the current eof, update the open state's
+     * last offset, and pass a pointer to LAYOUTCOMMIT */
+    if (state->pnfs_last_offset < last_offset ||
+        (state->pnfs_last_offset == 0 && last_offset == 0)) {
+        state->pnfs_last_offset = last_offset;
+        new_last_offset = &last_offset;
+    }
+    ReleaseSRWLockExclusive(&state->lock);
+
+    AcquireSRWLockShared(&layout->lock);
+    memcpy(&layout_stateid, &layout->stateid, sizeof(layout_stateid));
+    ReleaseSRWLockShared(&layout->lock);
+
+    dprintf(1, "LAYOUTCOMMIT for offset=%lld len=%lld new_last_offset=%u\n",
+        offset, length, new_last_offset ? 1 : 0);
+    nfsstat = pnfs_rpc_layoutcommit(state->session, &state->file,
+        &layout_stateid, offset, length, new_last_offset, NULL);
+    if (nfsstat) {
+        dprintf(IOLVL, "pnfs_rpc_layoutcommit() failed with %s\n",
+            nfs_error_string(nfsstat));
+        status = PNFSERR_IO;
+    }
+    return status;
+}
+
 enum pnfs_status pnfs_write(
     IN nfs41_root *root,
     IN nfs41_open_state *state,
@@ -523,20 +561,7 @@ enum pnfs_status pnfs_write(
         }
     } else if (stable == DATA_SYNC4) {
         /* send LAYOUTCOMMIT to sync the metadata */
-        stateid4 layout_stateid;
+        status = layout_commit(state, layout, offset, *len_out);
-        uint64_t new_last_offset = offset + *len_out - 1;
-
-        AcquireSRWLockShared(&layout->lock);
-        memcpy(&layout_stateid, &layout->stateid, sizeof(layout_stateid));
-        ReleaseSRWLockShared(&layout->lock);
-
-        nfsstat = pnfs_rpc_layoutcommit(state->session, &state->file,
-            &layout_stateid, offset, *len_out, &new_last_offset, NULL);
-        if (nfsstat) {
-            dprintf(IOLVL, "pnfs_rpc_layoutcommit() failed with %s\n",
-                nfs_error_string(nfsstat));
-            status = PNFSERR_IO;
-        }
     }
 out_free_pattern:
     pattern_free(&pattern);

daemon/setattr.c

@@ -353,10 +353,17 @@ static int handle_nfs41_set_size(setattr_upcall_args *args)
 
     dprintf(2, "calling setattr() with size=%lld\n", info.size);
     status = nfs41_setattr(state->session, &state->file, &stateid, &info);
-    if (status)
+    if (status) {
         dprintf(1, "nfs41_setattr() failed with error %s.\n",
             nfs_error_string(status));
+        goto out;
+    }
 
+    /* update the last offset for LAYOUTCOMMIT */
+    AcquireSRWLockExclusive(&state->lock);
+    state->pnfs_last_offset = info.size ? info.size - 1 : 0;
+    ReleaseSRWLockExclusive(&state->lock);
+out:
     return status = nfs_to_windows_error(status, ERROR_NOT_SUPPORTED);
 }