check write verifiers on COMMIT

we were previously only verifying that the server didn't reboot between WRITEs.  COMMIT returns a verifier that needs to be checked as well

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

daemon/nfs41_ops.c

@@ -840,7 +840,8 @@ int nfs41_commit(
     IN nfs41_path_fh *file,
     IN uint64_t offset,
     IN uint32_t count,
-    IN bool_t do_getattr)
+    IN bool_t do_getattr,
+    OUT nfs41_write_verf *verf)
 {
     int status;
     nfs41_compound compound;
@@ -872,6 +873,7 @@ int nfs41_commit(
     compound_add_op(&compound, OP_COMMIT, &commit_args, &commit_res);
     commit_args.offset = offset;
     commit_args.count = count;
+    commit_res.verf = verf;
 
     /* send a GETATTR request to update the attribute cache,
      * but not if we're talking to a data server! */

daemon/nfs41_ops.h

@@ -335,7 +335,7 @@ typedef struct __nfs41_commit_args {
 
 typedef struct __nfs41_commit_res {
     uint32_t                status;
-    unsigned char           writeverf[NFS4_VERIFIER_SIZE];
+    nfs41_write_verf        *verf;
 } nfs41_commit_res;
 
 
@@ -825,12 +825,6 @@ enum stable_how4 {
     FILE_SYNC4      = 2
 };
 
-typedef struct __nfs41_write_verf {
-    unsigned char           verf[NFS4_VERIFIER_SIZE];
-    unsigned char           expected[NFS4_VERIFIER_SIZE];
-    enum stable_how4        committed;
-} nfs41_write_verf;
-
 typedef struct __nfs41_write_args {
     stateid_arg             *stateid; /* -> nfs41_op_open_res_ok.stateid */
     uint64_t                offset;
@@ -1079,7 +1073,8 @@ int nfs41_commit(
     IN nfs41_path_fh *file,
     IN uint64_t offset,
     IN uint32_t count,
-    IN bool_t do_getattr);
+    IN bool_t do_getattr,
+    OUT nfs41_write_verf *verf);
 
 int nfs41_lock(
     IN nfs41_session *session,

daemon/nfs41_types.h

@@ -77,6 +77,12 @@ typedef struct __nfs41_readdir_cookie {
     unsigned char   verf[NFS4_VERIFIER_SIZE];
 } nfs41_readdir_cookie;
 
+typedef struct __nfs41_write_verf {
+    unsigned char   verf[NFS4_VERIFIER_SIZE];
+    unsigned char   expected[NFS4_VERIFIER_SIZE];
+    enum stable_how4 committed;
+} nfs41_write_verf;
+
 typedef struct __netaddr4 {
     char            netid[NFS41_NETWORK_ID_LEN+1];
     char            uaddr[NFS41_UNIVERSAL_ADDR_LEN+1];

daemon/nfs41_xdr.c

@@ -1302,7 +1302,7 @@ static bool_t decode_op_commit(
         return FALSE;
 
     if (res->status == NFS4_OK)
-        return xdr_opaque(xdr, (char *)res->writeverf, NFS4_VERIFIER_SIZE);
+        return xdr_opaque(xdr, (char *)res->verf->verf, NFS4_VERIFIER_SIZE);
 
     return TRUE;
 }

daemon/pnfs_io.c

@@ -421,14 +421,20 @@ retry_write:
     dprintf(1, "sending COMMIT to data server for offset=%lld len=%lld\n",
         commit_min, commit_max - commit_min);
     nfsstat = nfs41_commit(client->session, commit_file,
-        commit_min, (uint32_t)(commit_max - commit_min), 0);
+        commit_min, (uint32_t)(commit_max - commit_min), 0, &verf);
 
-    /* on successful commit, leave pnfs_status unchanged; if the layout
-     * was recalled, we still want to return the error */
-    if (nfsstat == NFS4_OK)
-        thread->stable = DATA_SYNC4;
-    else
+    if (nfsstat)
         status = map_ds_error(nfsstat, pattern->state);
+    else if (!verify_commit(&verf)) {
+        /* resend the writes unless the layout was recalled */
+        if (status != PNFSERR_LAYOUT_RECALLED)
+            goto retry_write;
+        status = PNFSERR_IO;
+    } else {
+        /* on successful commit, leave pnfs_status unchanged; if the
+         * layout was recalled, we still want to return the error */
+        thread->stable = DATA_SYNC4;
+    }
 out:
     dprintf(IOLVL, "<-- file_layout_write_thread(%u) returning %s\n",
         thread->id, pnfs_error_string(status));
@@ -550,10 +556,13 @@ enum pnfs_status pnfs_write(
         goto out_free_pattern;
 
     if (stable == UNSTABLE4) {
+        nfs41_write_verf ignored;
+
         /* not all data was committed, so commit to metadata server */
         dprintf(1, "sending COMMIT to meta server for offset=%lld len=%lld\n",
             offset, *len_out);
-        nfsstat = nfs41_commit(state->session, &state->file, offset, *len_out, 1);
+        nfsstat = nfs41_commit(state->session, &state->file,
+            offset, *len_out, 1, &ignored);
         if (nfsstat) {
             dprintf(IOLVL, "nfs41_commit() failed with %s\n",
                 nfs_error_string(nfsstat));

daemon/readwrite.c

@@ -217,7 +217,12 @@ retry_write:
     }
     if (committed == UNSTABLE4) {
         dprintf(1, "sending COMMIT for offset=%d and len=%d\n", args->offset, len);
-        status = nfs41_commit(session, file, args->offset, len, 1);
+        status = nfs41_commit(session, file, args->offset, len, 1, &verf);
+        if (status)
+            goto out;
+
+        if (!verify_commit(&verf))
+            goto retry_write;
     }
 out:
     args->out_len = len;

daemon/util.c

@@ -127,6 +127,17 @@ bool_t verify_write(
     return 0;
 }
 
+bool_t verify_commit(
+    IN nfs41_write_verf *verf)
+{
+    if (memcmp(verf->expected, verf->verf, NFS4_VERIFIER_SIZE) == 0) {
+        dprintf(3, "verify_commit: verifier matches expected\n");
+        return 1;
+    }
+    dprintf(2, "verify_commit: verifier changed; writes have been lost!\n");
+    return 0;
+}
+
 ULONG nfs_file_info_to_attributes(
     IN const nfs41_file_info *info)
 {

daemon/util.h

@@ -50,8 +50,10 @@ uint32_t max_write_size(
     IN const nfs41_fh *fh);
 
 bool_t verify_write(
-    IN struct __nfs41_write_verf *verf,
+    IN nfs41_write_verf *verf,
     IN OUT enum stable_how4 *stable);
+bool_t verify_commit(
+    IN nfs41_write_verf *verf);
 
 /* bitmap4 */
 static __inline bool_t bitmap_isset(