pnfs: layoutget, layoutreturn rpcs no longer operate on shared data

LAYOUTGET xdr now supports decoding of multiple layout segments, which are returned in a list with pnfs_layoutget_res_ok
LAYOUTRETURN no longer operates on an existing pnfs_file_layout.  it now takes a copy of the layout stateid, and returns the new stateid with pnfs_layoutreturn_res

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

daemon/nfs41_ops.c

@@ -1606,7 +1606,7 @@ enum nfsstat4 pnfs_rpc_layoutget(
     IN enum pnfs_iomode iomode,
     IN uint64_t offset,
     IN uint64_t length,
-    OUT pnfs_file_layout *layout)
+    OUT pnfs_layoutget_res_ok *layoutget_res_ok)
 {
     enum nfsstat4 status;
     nfs41_compound compound;
@@ -1619,6 +1619,7 @@ enum nfsstat4 pnfs_rpc_layoutget(
     pnfs_layoutget_args layoutget_args;
     pnfs_layoutget_res layoutget_res;
     uint32_t i;
+    struct list_entry *entry;
 
     compound_init(&compound, argops, resops, "layoutget");
 
@@ -1639,7 +1640,8 @@ enum nfsstat4 pnfs_rpc_layoutget(
     layoutget_args.minlength = layoutget_args.length = length;
     layoutget_args.stateid = stateid;
     layoutget_args.maxcount = session->fore_chan_attrs.ca_maxresponsesize - READ_OVERHEAD;
-    layoutget_res.u.res_ok.layout = layout;
+    ZeroMemory(&layoutget_res, sizeof(layoutget_res));
+    layoutget_res.u.res_ok = layoutget_res_ok;
 
     status = compound_encode_send_decode(session, &compound, FALSE);
     if (status)
@@ -1649,8 +1651,14 @@ enum nfsstat4 pnfs_rpc_layoutget(
         goto out;
 
     /* point each file handle to the meta server's superblock */
+    list_for_each(entry, &layoutget_res_ok->layouts) {
+        pnfs_layout *base = list_container(entry, pnfs_layout, entry);
+        if (base->type == PNFS_LAYOUTTYPE_FILE) {
+            pnfs_file_layout *layout = (pnfs_file_layout*)base;
             for (i = 0; i < layout->filehandles.count; i++)
                 layout->filehandles.arr[i].fh.superblock = file->fh.superblock;
+        }
+    }
 out:
     return status;
 }
@@ -1723,7 +1731,12 @@ out:
 enum nfsstat4 pnfs_rpc_layoutreturn(
     IN nfs41_session *session,
     IN nfs41_path_fh *file,
-    IN pnfs_file_layout *layout)
+    IN enum pnfs_layout_type type,
+    IN enum pnfs_iomode iomode,
+    IN uint64_t offset,
+    IN uint64_t length,
+    IN stateid4 *stateid,
+    OUT pnfs_layoutreturn_res *layoutreturn_res)
 {
     enum nfsstat4 status;
     nfs41_compound compound;
@@ -1734,7 +1747,6 @@ enum nfsstat4 pnfs_rpc_layoutreturn(
     nfs41_putfh_args putfh_args;
     nfs41_putfh_res putfh_res;
     pnfs_layoutreturn_args layoutreturn_args;
-    pnfs_layoutreturn_res layoutreturn_res;
 
     compound_init(&compound, argops, resops, "layoutreturn");
 
@@ -1747,30 +1759,20 @@ enum nfsstat4 pnfs_rpc_layoutreturn(
     putfh_args.file = file;
     putfh_args.in_recovery = 0;
 
-    compound_add_op(&compound, OP_LAYOUTRETURN, &layoutreturn_args, &layoutreturn_res);
+    compound_add_op(&compound, OP_LAYOUTRETURN, &layoutreturn_args, layoutreturn_res);
     layoutreturn_args.reclaim = 0;
-    layoutreturn_args.type = layout->layout.type;
+    layoutreturn_args.type = type;
-    layoutreturn_args.iomode = layout->layout.iomode;
+    layoutreturn_args.iomode = iomode;
     layoutreturn_args.return_type = PNFS_RETURN_FILE;
-    layoutreturn_args.offset = layout->layout.offset;
+    layoutreturn_args.offset = offset;
-    layoutreturn_args.length = layout->layout.length;
+    layoutreturn_args.length = length;
-    layoutreturn_args.stateid = &layout->layout.state;
+    layoutreturn_args.stateid = stateid;
 
     status = compound_encode_send_decode(session, &compound, FALSE);
     if (status)
         goto out;
 
-    if (compound_error(status = compound.res.status))
+    compound_error(status = compound.res.status);
-        goto out;
-
-    if (layoutreturn_res.stateid_present) {
-        /* update the layout seqid */
-        layout->layout.state.seqid = layoutreturn_res.stateid.seqid;
-    } else {
-        /* 12.5.3. Layout Stateid: Once a client has no more layouts on a file,
-         * the layout stateid is no longer valid and MUST NOT be used. */
-        ZeroMemory(&layout->layout.state, sizeof(stateid4));
-    }
 out:
     return status;
 }

daemon/nfs41_ops.h

@@ -796,14 +796,17 @@ typedef struct __pnfs_layoutget_args {
 } pnfs_layoutget_args;
 
 typedef struct __pnfs_layoutget_res_ok {
-    pnfs_file_layout        *layout;
+    bool_t                  return_on_close;
+    stateid4                stateid;
+    uint32_t                count;
+    struct list_entry       layouts; /* list of pnfs_layouts */
 } pnfs_layoutget_res_ok;
 
 typedef struct __pnfs_layoutget_res {
     enum nfsstat4           status;
     union {
     /* case NFS4_OK: */
-        pnfs_layoutget_res_ok res_ok;
+        pnfs_layoutget_res_ok *res_ok;
     /* case NFS4ERR_LAYOUTTRYLATER: */
         bool_t              will_signal_layout_avail;
     /* default: void; */
@@ -1086,7 +1089,7 @@ enum nfsstat4 pnfs_rpc_layoutget(
     IN enum pnfs_iomode iomode,
     IN uint64_t offset,
     IN uint64_t length,
-    OUT pnfs_file_layout *layout);
+    OUT pnfs_layoutget_res_ok *layoutget_res);
 
 enum nfsstat4 pnfs_rpc_layoutcommit(
     IN nfs41_session *session,
@@ -1100,7 +1103,12 @@ enum nfsstat4 pnfs_rpc_layoutcommit(
 enum nfsstat4 pnfs_rpc_layoutreturn(
     IN nfs41_session *session,
     IN nfs41_path_fh *file,
-    IN pnfs_file_layout *layout);
+    IN enum pnfs_layout_type type,
+    IN enum pnfs_iomode iomode,
+    IN uint64_t offset,
+    IN uint64_t length,
+    IN stateid4 *stateid,
+    OUT pnfs_layoutreturn_res *layoutreturn_res);
 
 enum nfsstat4 pnfs_rpc_getdeviceinfo(
     IN nfs41_session *session,

daemon/nfs41_xdr.c

@@ -2894,39 +2894,6 @@ static bool_t encode_op_layoutget(
     return xdr_u_int32_t(xdr, &args->maxcount);
 }
 
-static bool_t decode_layout(
-    XDR *xdr,
-    pnfs_layout *layout)
-{
-    uint32_t count;
-
-    if (!xdr_bool(xdr, &layout->return_on_close))
-        return FALSE;
-
-    if (!xdr_stateid4(xdr, &layout->state))
-        return FALSE;
-
-    if (!xdr_u_int32_t(xdr, &count))
-        return FALSE;
-
-    if (count != 1) {
-        eprintf("decode_layout: server returned a list of "
-            "%d layouts! only single layouts are supported!\n", count);
-        return FALSE;
-    }
-
-    if (!xdr_u_hyper(xdr, &layout->offset))
-        return FALSE;
-
-    if (!xdr_u_hyper(xdr, &layout->length))
-        return FALSE;
-
-    if (!xdr_enum(xdr, (enum_t *)&layout->iomode))
-        return FALSE;
-
-    return xdr_enum(xdr, (enum_t *)&layout->type);
-}
-
 static bool_t decode_file_layout_handles(
     XDR *xdr,
     pnfs_file_layout_handles *handles)
@@ -2955,35 +2922,96 @@ static bool_t decode_file_layout_handles(
 
 static bool_t decode_file_layout(
     XDR *xdr,
-    pnfs_file_layout *layout)
+    struct list_entry *list,
+    pnfs_layout *base)
 {
+    pnfs_file_layout *layout;
     u_int32_t len_ignored;
 
-    if (!decode_layout(xdr, &layout->layout))
-        return FALSE;
-
-    if (layout->layout.type != PNFS_LAYOUTTYPE_FILE) {
-        eprintf("%s: received non-FILE layout type, %d\n",
-            "decode_file_layout", layout->layout.type);
-        return FALSE;
-    }
-
     if (!xdr_u_int32_t(xdr, &len_ignored))
         return FALSE;
 
-    if (!xdr_opaque(xdr, (char *)layout->deviceid, PNFS_DEVICEID_SIZE))
+    layout = calloc(1, sizeof(pnfs_file_layout));
+    if (layout == NULL)
         return FALSE;
 
+    layout->layout.offset = base->offset;
+    layout->layout.length = base->length;
+    layout->layout.iomode = base->iomode;
+    layout->layout.type = base->type;
+    list_init(&layout->layout.entry);
+
+    if (!xdr_opaque(xdr, (char *)layout->deviceid, PNFS_DEVICEID_SIZE))
+        goto out_error;
+
     if (!xdr_u_int32_t(xdr, &layout->util))
-        return FALSE;
+        goto out_error;
 
     if (!xdr_u_int32_t(xdr, &layout->first_index))
-        return FALSE;
+        goto out_error;
 
     if (!xdr_u_hyper(xdr, &layout->pattern_offset))
+        goto out_error;
+
+    if (!decode_file_layout_handles(xdr, &layout->filehandles))
+        goto out_error;
+
+    list_add_tail(list, &layout->layout.entry);
+    return TRUE;
+
+out_error:
+    free(layout);
+    return FALSE;
+}
+
+static bool_t decode_layout(
+    XDR *xdr,
+    struct list_entry *list)
+{
+    pnfs_layout layout;
+
+    if (!xdr_u_hyper(xdr, &layout.offset))
         return FALSE;
 
-    return decode_file_layout_handles(xdr, &layout->filehandles);
+    if (!xdr_u_hyper(xdr, &layout.length))
+        return FALSE;
+
+    if (!xdr_enum(xdr, (enum_t *)&layout.iomode))
+        return FALSE;
+
+    if (!xdr_enum(xdr, (enum_t *)&layout.type))
+        return FALSE;
+
+    switch (layout.type) {
+    case PNFS_LAYOUTTYPE_FILE:
+        return decode_file_layout(xdr, list, &layout);
+
+    default:
+        eprintf("%s: received non-FILE layout type, %d\n",
+            "decode_file_layout", layout.type);
+    }
+    return FALSE;
+}
+
+static bool_t decode_layout_res_ok(
+    XDR *xdr,
+    pnfs_layoutget_res_ok *res)
+{
+    uint32_t i;
+
+    if (!xdr_bool(xdr, &res->return_on_close))
+        return FALSE;
+
+    if (!xdr_stateid4(xdr, &res->stateid))
+        return FALSE;
+
+    if (!xdr_u_int32_t(xdr, &res->count))
+        return FALSE;
+
+    for (i = 0; i < res->count; i++)
+        if (!decode_layout(xdr, &res->layouts))
+            return FALSE;
+    return TRUE;
 }
 
 static bool_t decode_op_layoutget(
@@ -3000,14 +3028,9 @@ static bool_t decode_op_layoutget(
 
     switch (res->status) {
     case NFS4_OK:
-        return decode_file_layout(xdr, res->u.res_ok.layout);
+        return decode_layout_res_ok(xdr, res->u.res_ok);
-        break;
     case NFS4ERR_LAYOUTTRYLATER:
-        {
+        return xdr_bool(xdr, &res->u.will_signal_layout_avail);
-            bool_t ignored;
-            return xdr_bool(xdr, &ignored);
-        }
-        break;
     }
     return TRUE;
 }