[PATCHSET 00/13] SQUASHME pnfs-obj: Lots of changes addressing comments by Trond and Benny

[PATCHSET 00/13] SQUASHME pnfs-obj: Lots of changes addressing comments by Trond and Benny

[Boaz Harrosh]

Submitted are SQUASHME patches on-top of *pnfs-exofs-all* that demonstrate all the
changes I did addressing comments by Trond and Benny. So it is easy to see the changes
and not need to review the complete patchset.

Thanks guys for the excellent review!

Next I will send a new *pnfs-submit* patchset so they can be reviewed and put in next.
(Benny you need to come back to me if it is OK to submit the full read/write raid-engine
 as one patch, instead of three)

Here is the list of SQUASHME patches:

[PATCH 01/13] SQUASHME: re-reorder the functions so it compares better with raids base

  Benny decided to change functions order. This completely through off all the changes
  I had pending in my trees. This patch does nothing but moves code back to its original
  place. See as reply to this email the diff of the nice changes Benny did after the last
  tree at pnfs/pnfs-all-2.6.38

[PATCH 02/13] SQUASHME: pnfs-obj: objio_segment only needs the pnfs_osd_layout->olo_comps
[PATCH 03/13] SQUASHME: pnfs-obj: Rename ios->objio_seg => ios->layout
[PATCH 04/13] SQUASHME: pnfs-obj: Fix groups code when num_devices is not divisible by group_width

  These three are bug fixes and last minute code style changes

[PATCH 05/13] SQUASHME: pnfs-obj: Convert layout and deviceinfo decoding to new XDR

  This one was requested by Trond and rightly so. It is now supported a bigger then
  4K layout, using all the new(old) exr_stream API. Thanks Trond!

[PATCH 06/13] SQUASHME: pnfs-obj: Change API of objlayout_io_set_result
[PATCH 07/13] SQUASHME: pnfs-obj: Avoid double allocation logic in objlayout_alloc_lseg

  Another nice cleanup and performance boost.

[PATCH 08/13] SQUASHME: pnfs_osd_xdr: Remove Server API declarations

  Remove the dependency on the exp_xdr header.

[PATCH 09/13] SQUASHME: pnfs_osd_xdr: Avoid using xdr_rewind_stream

  Also requested by Trond

[PATCH 10/13] [RFC] Bugs in new pnfs write path

  This is a resend of the BUGs patch to pnfs-core (So I can test the tree)

Below is some exofs changes needed to compile and test.
[PATCH 11/13] SQUASHME: pnfs_osd_xdr: Add Server API for encoding/decoding osd XDRs
[PATCH 12/13] SQUASHME: XDR API changes to pnfs_osd_xdr_decode_ioerr()
[PATCH 13/13] SQUASHME: dbg Print the full device_id returned

  Benny could you submit these to the pnfs-exofs-all branch?

So next I'll send these already squashed into a new submit patchset

Thanks
Boaz

[PATCH 01/13] SQUASHME: re-reorder the functions so it compares better with raids base

[Boaz Harrosh]

This patch is based on pnfs-exofs. It will not patch on pnfs-submit
because of a one-liner patch  that comes after pnfs-submit. (I have
one for there as well)

It is here so I can compare with previouse code .i.e pnfs-all-2.6.38
and so it patches with all the work I have in my trees.

I'm reverting to the old functions order. All this code
shuffling was out of place, and gave me lots of grief. If at all
it must be done as a very last-last step. After there is no
more development to do. (And will still break all out of tree
work. Then why?)

Signed-off-by: Boaz Harrosh <bharrosh@panasas.com>
---
 fs/nfs/objlayout/objio_osd.c        |   52 ++++----
 fs/nfs/objlayout/objlayout.c        |  246 ++++++++++++++++++-----------------
 fs/nfs/objlayout/objlayout.h        |   24 ++--
 fs/nfs/objlayout/pnfs_osd_xdr_cli.c |   46 ++++----
 4 files changed, 186 insertions(+), 182 deletions(-)

diff --git a/fs/nfs/objlayout/objio_osd.c b/fs/nfs/objlayout/objio_osd.c
index e7a0fcb..a2cb9a3 100644
--- a/fs/nfs/objlayout/objio_osd.c
+++ b/fs/nfs/objlayout/objio_osd.c
@@ -142,6 +142,31 @@ struct objio_segment {
 	struct osd_dev	*ods[1];
 };
 
+struct objio_state;
+typedef ssize_t (*objio_done_fn)(struct objio_state *ios);
+
+struct objio_state {
+	/* Generic layer */
+	struct objlayout_io_state ol_state;
+
+	struct objio_segment *objio_seg;
+
+	struct kref kref;
+	objio_done_fn done;
+	void *private;
+
+	unsigned long length;
+	unsigned numdevs; /* Actually used devs in this IO */
+	/* A per-device variable array of size numdevs */
+	struct _objio_per_comp {
+		struct bio *bio;
+		struct osd_request *or;
+		unsigned long length;
+		u64 offset;
+		unsigned dev;
+	} per_dev[];
+};
+
 /* Send and wait for a get_device_info of devices in the layout,
    then look them up with the osd_initiator library */
 static struct osd_dev *_device_lookup(struct pnfs_layout_hdr *pnfslay,
@@ -225,31 +250,6 @@ out:
 	return err;
 }
 
-struct objio_state;
-typedef ssize_t (*objio_done_fn)(struct objio_state *ios);
-
-struct objio_state {
-	/* Generic layer */
-	struct objlayout_io_state ol_state;
-
-	struct objio_segment *objio_seg;
-
-	struct kref kref;
-	objio_done_fn done;
-	void *private;
-
-	unsigned long length;
-	unsigned numdevs; /* Actually used devs in this IO */
-	/* A per-device variable array of size numdevs */
-	struct _objio_per_comp {
-		struct bio *bio;
-		struct osd_request *or;
-		unsigned long length;
-		u64 offset;
-		unsigned dev;
-	} per_dev[];
-};
-
 static int _verify_data_map(struct pnfs_osd_layout *layout)
 {
 	struct pnfs_osd_data_map *data_map = &layout->olo_map;
@@ -976,8 +976,8 @@ static struct pnfs_layoutdriver_type objlayout_type = {
 	.read_pagelist           = objlayout_read_pagelist,
 	.write_pagelist          = objlayout_write_pagelist,
 
+	.encode_layoutcommit	 = objlayout_encode_layoutcommit,
 	.encode_layoutreturn     = objlayout_encode_layoutreturn,
-	.encode_layoutcommit     = objlayout_encode_layoutcommit,
 };
 
 void *objio_init_mt(void)
diff --git a/fs/nfs/objlayout/objlayout.c b/fs/nfs/objlayout/objlayout.c
index a477492..a04e232 100644
--- a/fs/nfs/objlayout/objlayout.c
+++ b/fs/nfs/objlayout/objlayout.c
@@ -152,107 +152,6 @@ objlayout_free_lseg(struct pnfs_layout_segment *lseg)
 	kfree(objlseg);
 }
 
-struct objlayout_deviceinfo {
-	struct page *page;
-	struct pnfs_osd_deviceaddr da; /* This must be last */
-};
-
-/* Initialize and call nfs_getdeviceinfo, then decode and return a
- * "struct pnfs_osd_deviceaddr *" Eventually objlayout_put_deviceinfo()
- * should be called.
- */
-int objlayout_get_deviceinfo(struct pnfs_layout_hdr *pnfslay,
-	struct nfs4_deviceid *d_id, struct pnfs_osd_deviceaddr **deviceaddr)
-{
-	struct objlayout_deviceinfo *odi;
-	struct pnfs_device pd;
-	struct super_block *sb;
-	struct page *page, **pages;
-	size_t sz;
-	u32 *p;
-	int err;
-
-	page = alloc_page(GFP_KERNEL);
-	if (!page)
-		return -ENOMEM;
-
-	pages = &page;
-	pd.pages = pages;
-
-	memcpy(&pd.dev_id, d_id, sizeof(*d_id));
-	pd.layout_type = LAYOUT_OSD2_OBJECTS;
-	pd.pages = &page;
-	pd.pgbase = 0;
-	pd.pglen = PAGE_SIZE;
-	pd.mincount = 0;
-
-	sb = pnfslay->plh_inode->i_sb;
-	err = nfs4_proc_getdeviceinfo(NFS_SERVER(pnfslay->plh_inode), &pd);
-	dprintk("%s nfs_getdeviceinfo returned %d\n", __func__, err);
-	if (err)
-		goto err_out;
-
-	p = page_address(page);
-	sz = pnfs_osd_xdr_deviceaddr_incore_sz(p);
-	odi = kzalloc(sz + (sizeof(*odi) - sizeof(odi->da)), GFP_KERNEL);
-	if (!odi) {
-		err = -ENOMEM;
-		goto err_out;
-	}
-	pnfs_osd_xdr_decode_deviceaddr(&odi->da, p);
-	odi->page = page;
-	*deviceaddr = &odi->da;
-	return 0;
-
-err_out:
-	__free_page(page);
-	return err;
-}
-
-void objlayout_put_deviceinfo(struct pnfs_osd_deviceaddr *deviceaddr)
-{
-	struct objlayout_deviceinfo *odi = container_of(deviceaddr,
-						struct objlayout_deviceinfo,
-						da);
-
-	__free_page(odi->page);
-	kfree(odi);
-}
-
-/*
- * Perform the objio specific init_mt method.
- * Set the layout driver private data pointer for later use.
- */
-int
-objlayout_set_layoutdriver(struct nfs_server *server,
-			   const struct nfs_fh *mntfh)
-{
-	void *data;
-
-	data = objio_init_mt();
-	if (IS_ERR(data)) {
-		printk(KERN_INFO "%s: objlayout lib not ready err=%ld\n",
-		       __func__, PTR_ERR(data));
-		return PTR_ERR(data);
-	}
-	server->pnfs_ld_data = data;
-
-	dprintk("%s: Return data=%p\n", __func__, data);
-	return 0;
-}
-
-/*
- * Perform the objio specific fini_mt method to release the
- * layoutdriver private data.
- */
-int
-objlayout_unset_layoutdriver(struct nfs_server *server)
-{
-	dprintk("%s: Begin %p\n", __func__, server->pnfs_ld_data);
-	objio_fini_mt(server->pnfs_ld_data);
-	return 0;
-}
-
 /*
  * I/O Operations
  */
@@ -553,6 +452,35 @@ objlayout_write_pagelist(struct nfs_write_data *wdata,
 	return PNFS_ATTEMPTED;
 }
 
+void
+objlayout_encode_layoutcommit(struct pnfs_layout_hdr *pnfslay,
+			      struct xdr_stream *xdr,
+			      const struct nfs4_layoutcommit_args *args)
+{
+	struct objlayout *objlay = OBJLAYOUT(pnfslay);
+	struct pnfs_osd_layoutupdate lou;
+	__be32 *start;
+
+	dprintk("%s: Begin\n", __func__);
+
+	spin_lock(&objlay->lock);
+	lou.dsu_valid = (objlay->delta_space_valid == OBJ_DSU_VALID);
+	lou.dsu_delta = objlay->delta_space_used;
+	objlay->delta_space_used = 0;
+	objlay->delta_space_valid = OBJ_DSU_INIT;
+	lou.olu_ioerr_flag = !list_empty(&objlay->err_list);
+	spin_unlock(&objlay->lock);
+
+	start = xdr_reserve_space(xdr, 4);
+
+	BUG_ON(pnfs_osd_xdr_encode_layoutupdate(xdr, &lou));
+
+	*start = cpu_to_be32((xdr->p - start - 1) * 4);
+
+	dprintk("%s: Return delta_space_used %lld err %d\n", __func__,
+		lou.dsu_delta, lou.olu_ioerr_flag);
+}
+
 static int
 err_prio(u32 oer_errno)
 {
@@ -726,31 +654,107 @@ loop_done:
 	dprintk("%s: Return\n", __func__);
 }
 
-void
-objlayout_encode_layoutcommit(struct pnfs_layout_hdr *pnfslay,
-			      struct xdr_stream *xdr,
-			      const struct nfs4_layoutcommit_args *args)
+
+/*
+ * Get Device Info API for io engines
+ */
+struct objlayout_deviceinfo {
+	struct page *page;
+	struct pnfs_osd_deviceaddr da; /* This must be last */
+};
+
+/* Initialize and call nfs_getdeviceinfo, then decode and return a
+ * "struct pnfs_osd_deviceaddr *" Eventually objlayout_put_deviceinfo()
+ * should be called.
+ */
+int objlayout_get_deviceinfo(struct pnfs_layout_hdr *pnfslay,
+	struct nfs4_deviceid *d_id, struct pnfs_osd_deviceaddr **deviceaddr)
 {
-	struct objlayout *objlay = OBJLAYOUT(pnfslay);
-	struct pnfs_osd_layoutupdate lou;
-	__be32 *start;
+	struct objlayout_deviceinfo *odi;
+	struct pnfs_device pd;
+	struct super_block *sb;
+	struct page *page, **pages;
+	size_t sz;
+	u32 *p;
+	int err;
 
-	dprintk("%s: Begin\n", __func__);
+	page = alloc_page(GFP_KERNEL);
+	if (!page)
+		return -ENOMEM;
 
-	spin_lock(&objlay->lock);
-	lou.dsu_valid = (objlay->delta_space_valid == OBJ_DSU_VALID);
-	lou.dsu_delta = objlay->delta_space_used;
-	objlay->delta_space_used = 0;
-	objlay->delta_space_valid = OBJ_DSU_INIT;
-	lou.olu_ioerr_flag = !list_empty(&objlay->err_list);
-	spin_unlock(&objlay->lock);
+	pages = &page;
+	pd.pages = pages;
 
-	start = xdr_reserve_space(xdr, 4);
+	memcpy(&pd.dev_id, d_id, sizeof(*d_id));
+	pd.layout_type = LAYOUT_OSD2_OBJECTS;
+	pd.pages = &page;
+	pd.pgbase = 0;
+	pd.pglen = PAGE_SIZE;
+	pd.mincount = 0;
 
-	BUG_ON(pnfs_osd_xdr_encode_layoutupdate(xdr, &lou));
+	sb = pnfslay->plh_inode->i_sb;
+	err = nfs4_proc_getdeviceinfo(NFS_SERVER(pnfslay->plh_inode), &pd);
+	dprintk("%s nfs_getdeviceinfo returned %d\n", __func__, err);
+	if (err)
+		goto err_out;
 
-	*start = cpu_to_be32((xdr->p - start - 1) * 4);
+	p = page_address(page);
+	sz = pnfs_osd_xdr_deviceaddr_incore_sz(p);
+	odi = kzalloc(sz + (sizeof(*odi) - sizeof(odi->da)), GFP_KERNEL);
+	if (!odi) {
+		err = -ENOMEM;
+		goto err_out;
+	}
+	pnfs_osd_xdr_decode_deviceaddr(&odi->da, p);
+	odi->page = page;
+	*deviceaddr = &odi->da;
+	return 0;
 
-	dprintk("%s: Return delta_space_used %lld err %d\n", __func__,
-		lou.dsu_delta, lou.olu_ioerr_flag);
+err_out:
+	__free_page(page);
+	return err;
+}
+
+void objlayout_put_deviceinfo(struct pnfs_osd_deviceaddr *deviceaddr)
+{
+	struct objlayout_deviceinfo *odi = container_of(deviceaddr,
+						struct objlayout_deviceinfo,
+						da);
+
+	__free_page(odi->page);
+	kfree(odi);
+}
+
+/*
+ * Perform the objio specific init_mt method.
+ * Set the layout driver private data pointer for later use.
+ */
+int
+objlayout_set_layoutdriver(struct nfs_server *server,
+			   const struct nfs_fh *mntfh)
+{
+	void *data;
+
+	data = objio_init_mt();
+	if (IS_ERR(data)) {
+		printk(KERN_INFO "%s: objlayout lib not ready err=%ld\n",
+		       __func__, PTR_ERR(data));
+		return PTR_ERR(data);
+	}
+	server->pnfs_ld_data = data;
+
+	dprintk("%s: Return data=%p\n", __func__, data);
+	return 0;
+}
+
+/*
+ * Perform the objio specific fini_mt method to release the
+ * layoutdriver private data.
+ */
+int
+objlayout_unset_layoutdriver(struct nfs_server *server)
+{
+	dprintk("%s: Begin %p\n", __func__, server->pnfs_ld_data);
+	objio_fini_mt(server->pnfs_ld_data);
+	return 0;
 }
diff --git a/fs/nfs/objlayout/objlayout.h b/fs/nfs/objlayout/objlayout.h
index 3157071a..38abb01 100644
--- a/fs/nfs/objlayout/objlayout.h
+++ b/fs/nfs/objlayout/objlayout.h
@@ -59,10 +59,6 @@ struct objlayout_segment {
  */
 struct objlayout {
 	struct pnfs_layout_hdr pnfs_layout;
-	spinlock_t lock;
-
-	 /* for layout_return */
-	struct list_head err_list;
 
 	 /* for layout_commit */
 	enum osd_delta_space_valid_enum {
@@ -71,6 +67,10 @@ struct objlayout {
 		OBJ_DSU_INVALID,
 	} delta_space_valid;
 	s64 delta_space_used;  /* consumed by write ops */
+
+	 /* for layout_return */
+	spinlock_t lock;
+	struct list_head err_list;
 };
 
 static inline struct objlayout *
@@ -157,6 +157,10 @@ extern void objlayout_read_done(struct objlayout_io_state *state,
 extern void objlayout_write_done(struct objlayout_io_state *state,
 				 ssize_t status, bool sync);
 
+extern int objlayout_get_deviceinfo(struct pnfs_layout_hdr *pnfslay,
+	struct nfs4_deviceid *d_id, struct pnfs_osd_deviceaddr **deviceaddr);
+extern void objlayout_put_deviceinfo(struct pnfs_osd_deviceaddr *deviceaddr);
+
 /*
  * exported generic objects function vectors
  */
@@ -173,10 +177,6 @@ extern struct pnfs_layout_segment *objlayout_alloc_lseg(
 	struct nfs4_layoutget_res *);
 extern void objlayout_free_lseg(struct pnfs_layout_segment *);
 
-extern int objlayout_get_deviceinfo(struct pnfs_layout_hdr *pnfslay,
-	struct nfs4_deviceid *d_id, struct pnfs_osd_deviceaddr **deviceaddr);
-extern void objlayout_put_deviceinfo(struct pnfs_osd_deviceaddr *deviceaddr);
-
 extern enum pnfs_try_status objlayout_read_pagelist(
 	struct nfs_read_data *);
 
@@ -184,14 +184,14 @@ extern enum pnfs_try_status objlayout_write_pagelist(
 	struct nfs_write_data *,
 	int how);
 
-extern void objlayout_encode_layoutreturn(
+extern void objlayout_encode_layoutcommit(
 	struct pnfs_layout_hdr *,
 	struct xdr_stream *,
-	const struct nfs4_layoutreturn_args *);
+	const struct nfs4_layoutcommit_args *);
 
-extern void objlayout_encode_layoutcommit(
+extern void objlayout_encode_layoutreturn(
 	struct pnfs_layout_hdr *,
 	struct xdr_stream *,
-	const struct nfs4_layoutcommit_args *);
+	const struct nfs4_layoutreturn_args *);
 
 #endif /* _OBJLAYOUT_H */
diff --git a/fs/nfs/objlayout/pnfs_osd_xdr_cli.c b/fs/nfs/objlayout/pnfs_osd_xdr_cli.c
index 4ff2e3e..4dea458 100644
--- a/fs/nfs/objlayout/pnfs_osd_xdr_cli.c
+++ b/fs/nfs/objlayout/pnfs_osd_xdr_cli.c
@@ -353,6 +353,29 @@ void pnfs_osd_xdr_decode_deviceaddr(
 }
 
 /*
+ * struct pnfs_osd_layoutupdate {
+ * 	u32	dsu_valid;
+ * 	s64	dsu_delta;
+ * 	u32	olu_ioerr_flag;
+ * };
+ */
+int
+pnfs_osd_xdr_encode_layoutupdate(struct xdr_stream *xdr,
+				 struct pnfs_osd_layoutupdate *lou)
+{
+	__be32 *p = xdr_reserve_space(xdr, 16);
+
+	if (!p)
+		return -E2BIG;
+
+	*p++ = cpu_to_be32(lou->dsu_valid);
+	if (lou->dsu_valid)
+		p = xdr_encode_hyper(p, lou->dsu_delta);
+	*p++ = cpu_to_be32(lou->olu_ioerr_flag);
+	return 0;
+}
+
+/*
  * struct pnfs_osd_objid {
  * 	struct pnfs_deviceid	oid_device_id;
  * 	u64			oid_partition_id;
@@ -405,26 +428,3 @@ int pnfs_osd_xdr_encode_ioerr(struct xdr_stream *xdr,
 
 	return 0;
 }
-
-/*
- * struct pnfs_osd_layoutupdate {
- * 	u32	dsu_valid;
- * 	s64	dsu_delta;
- * 	u32	olu_ioerr_flag;
- * };
- */
-int
-pnfs_osd_xdr_encode_layoutupdate(struct xdr_stream *xdr,
-				 struct pnfs_osd_layoutupdate *lou)
-{
-	__be32 *p = xdr_reserve_space(xdr, 16);
-
-	if (!p)
-		return -E2BIG;
-
-	*p++ = cpu_to_be32(lou->dsu_valid);
-	if (lou->dsu_valid)
-		p = xdr_encode_hyper(p, lou->dsu_delta);
-	*p++ = cpu_to_be32(lou->olu_ioerr_flag);
-	return 0;
-}
-- 
1.7.2.3

[PATCH 02/13] SQUASHME: pnfs-obj: objio_segment only needs the pnfs_osd_layout->olo_comps

[Boaz Harrosh]

All of struct pnfs_osd_layout is duplicated / enhanced in
objio_segment the only information we used from the generic
pnfs_osd_layout was the pnfs_osd_object_cred *olo_comps
array and the olo_comps_index. Embed these into objio_segment
directly.

This patch is a part of a patchset that will convert the
pnfs-obj code to use the exofs io_engine instead of duplicating
it here.

Signed-off-by: Boaz Harrosh <bharrosh@panasas.com>
---
 fs/nfs/objlayout/objio_osd.c |   24 ++++++++++++------------
 1 files changed, 12 insertions(+), 12 deletions(-)

diff --git a/fs/nfs/objlayout/objio_osd.c b/fs/nfs/objlayout/objio_osd.c
index a2cb9a3..9524d93 100644
--- a/fs/nfs/objlayout/objio_osd.c
+++ b/fs/nfs/objlayout/objio_osd.c
@@ -129,7 +129,7 @@ out:
 }
 
 struct objio_segment {
-	struct pnfs_osd_layout *layout;
+	struct pnfs_osd_object_cred *comps;
 
 	unsigned mirrors_p1;
 	unsigned stripe_unit;
@@ -137,6 +137,7 @@ struct objio_segment {
 	u64 group_depth;
 	unsigned group_count;
 
+	unsigned comps_index;
 	unsigned num_comps;
 	/* variable length */
 	struct osd_dev	*ods[1];
@@ -172,7 +173,6 @@ struct objio_state {
 static struct osd_dev *_device_lookup(struct pnfs_layout_hdr *pnfslay,
 			       struct objio_segment *objio_seg, unsigned comp)
 {
-	struct pnfs_osd_layout *layout = objio_seg->layout;
 	struct pnfs_osd_deviceaddr *deviceaddr;
 	struct nfs4_deviceid *d_id;
 	struct osd_dev *od;
@@ -180,7 +180,7 @@ static struct osd_dev *_device_lookup(struct pnfs_layout_hdr *pnfslay,
 	struct objio_mount_type *omt = NFS_SERVER(pnfslay->plh_inode)->pnfs_ld_data;
 	int err;
 
-	d_id = &layout->olo_comps[comp].oc_object_id.oid_device_id;
+	d_id = &objio_seg->comps[comp].oc_object_id.oid_device_id;
 
 	od = _dev_list_find(omt, d_id);
 	if (od)
@@ -227,12 +227,11 @@ out:
 static int objio_devices_lookup(struct pnfs_layout_hdr *pnfslay,
 	struct objio_segment *objio_seg)
 {
-	struct pnfs_osd_layout *layout = objio_seg->layout;
-	unsigned i, num_comps = layout->olo_num_comps;
+	unsigned i;
 	int err;
 
 	/* lookup all devices */
-	for (i = 0; i < num_comps; i++) {
+	for (i = 0; i < objio_seg->num_comps; i++) {
 		struct osd_dev *od;
 
 		od = _device_lookup(pnfslay, objio_seg, i);
@@ -242,7 +241,6 @@ static int objio_devices_lookup(struct pnfs_layout_hdr *pnfslay,
 		}
 		objio_seg->ods[i] = od;
 	}
-	objio_seg->num_comps = num_comps;
 	err = 0;
 
 out:
@@ -308,7 +306,9 @@ int objio_alloc_lseg(void **outp,
 	if (!objio_seg)
 		return -ENOMEM;
 
-	objio_seg->layout = layout;
+	objio_seg->comps_index = layout->olo_comps_index;
+	objio_seg->num_comps = layout->olo_num_comps;
+	objio_seg->comps = layout->olo_comps;
 	err = objio_devices_lookup(pnfslay, objio_seg);
 	if (err)
 		goto free_seg;
@@ -488,7 +488,7 @@ static void _io_free(struct objio_state *ios)
 
 struct osd_dev * _io_od(struct objio_state *ios, unsigned dev)
 {
-	unsigned min_dev = ios->objio_seg->layout->olo_comps_index;
+	unsigned min_dev = ios->objio_seg->comps_index;
 	unsigned max_dev = min_dev + ios->ol_state.num_comps;
 
 	BUG_ON(dev < min_dev || max_dev <= dev);
@@ -653,7 +653,7 @@ static int _io_rw_pagelist(struct objio_state *ios)
 	unsigned devs_in_group = ios->objio_seg->group_width *
 				 ios->objio_seg->mirrors_p1;
 	unsigned first_comp = 0;
-	unsigned num_comps = ios->objio_seg->layout->olo_map.odm_num_comps;
+	unsigned num_comps = devs_in_group * ios->objio_seg->group_count;
 	unsigned last_pg = 0;
 	int ret = 0;
 
@@ -771,7 +771,7 @@ static int _read_mirrors(struct objio_state *ios, unsigned cur_comp)
 	struct _objio_per_comp *per_dev = &ios->per_dev[cur_comp];
 	unsigned dev = per_dev->dev;
 	struct pnfs_osd_object_cred *cred =
-			&ios->objio_seg->layout->olo_comps[dev];
+			&ios->objio_seg->comps[dev];
 	struct osd_obj_id obj = {
 		.partition = cred->oc_object_id.oid_partition_id,
 		.id = cred->oc_object_id.oid_object_id,
@@ -869,7 +869,7 @@ static int _write_mirrors(struct objio_state *ios, unsigned cur_comp)
 	for (; cur_comp < last_comp; ++cur_comp, ++dev) {
 		struct osd_request *or = NULL;
 		struct pnfs_osd_object_cred *cred =
-					&ios->objio_seg->layout->olo_comps[dev];
+					&ios->objio_seg->comps[dev];
 		struct osd_obj_id obj = {
 			.partition = cred->oc_object_id.oid_partition_id,
 			.id = cred->oc_object_id.oid_object_id,
-- 
1.7.2.3

[PATCH 03/13] SQUASHME: pnfs-obj: Rename ios->objio_seg => ios->layout

[Boaz Harrosh]

This is what the exofs raid engine calls it. As we transition
to use exofs_io_state directly and drop objio_io_start.

Signed-off-by: Boaz Harrosh <bharrosh@panasas.com>
---
 fs/nfs/objlayout/objio_osd.c |   48 +++++++++++++++++++++---------------------
 1 files changed, 24 insertions(+), 24 deletions(-)

diff --git a/fs/nfs/objlayout/objio_osd.c b/fs/nfs/objlayout/objio_osd.c
index 9524d93..1289dff 100644
--- a/fs/nfs/objlayout/objio_osd.c
+++ b/fs/nfs/objlayout/objio_osd.c
@@ -150,7 +150,7 @@ struct objio_state {
 	/* Generic layer */
 	struct objlayout_io_state ol_state;
 
-	struct objio_segment *objio_seg;
+	struct objio_segment *layout;
 
 	struct kref kref;
 	objio_done_fn done;
@@ -359,7 +359,7 @@ int objio_alloc_io_state(void *seg, struct objlayout_io_state **outp)
 	if (unlikely(!ios))
 		return -ENOMEM;
 
-	ios->objio_seg = objio_seg;
+	ios->layout = objio_seg;
 	ios->ol_state.ioerrs = ((void *)ios) + first_size;
 	ios->ol_state.num_comps = objio_seg->num_comps;
 
@@ -488,11 +488,11 @@ static void _io_free(struct objio_state *ios)
 
 struct osd_dev * _io_od(struct objio_state *ios, unsigned dev)
 {
-	unsigned min_dev = ios->objio_seg->comps_index;
+	unsigned min_dev = ios->layout->comps_index;
 	unsigned max_dev = min_dev + ios->ol_state.num_comps;
 
 	BUG_ON(dev < min_dev || max_dev <= dev);
-	return ios->objio_seg->ods[dev - min_dev];
+	return ios->layout->ods[dev - min_dev];
 }
 
 struct _striping_info {
@@ -507,13 +507,13 @@ struct _striping_info {
 static void _calc_stripe_info(struct objio_state *ios, u64 file_offset,
 			      struct _striping_info *si)
 {
-	u32	stripe_unit = ios->objio_seg->stripe_unit;
-	u32	group_width = ios->objio_seg->group_width;
-	u64	group_depth = ios->objio_seg->group_depth;
+	u32	stripe_unit = ios->layout->stripe_unit;
+	u32	group_width = ios->layout->group_width;
+	u64	group_depth = ios->layout->group_depth;
 	u32	U = stripe_unit * group_width;
 
 	u64	T = U * group_depth;
-	u64	S = T * ios->objio_seg->group_count;
+	u64	S = T * ios->layout->group_count;
 	u64	M = div64_u64(file_offset, S);
 
 	/*
@@ -532,7 +532,7 @@ static void _calc_stripe_info(struct objio_state *ios, u64 file_offset,
 
 	/* "H - (N * U)" is just "H % U" so it's bound to u32 */
 	si->dev = (u32)(H - (N * U)) / stripe_unit + G * group_width;
-	si->dev *= ios->objio_seg->mirrors_p1;
+	si->dev *= ios->layout->mirrors_p1;
 
 	si->group_length = T - H;
 	si->total_group_length = T;
@@ -550,9 +550,9 @@ static int _add_stripe_unit(struct objio_state *ios,  unsigned *cur_pg,
 
 	if (per_dev->bio == NULL) {
 		unsigned stripes = ios->ol_state.num_comps /
-						     ios->objio_seg->mirrors_p1;
+						     ios->layout->mirrors_p1;
 		unsigned pages_in_stripe = stripes *
-				      (ios->objio_seg->stripe_unit / PAGE_SIZE);
+				      (ios->layout->stripe_unit / PAGE_SIZE);
 		unsigned bio_size = (ios->ol_state.nr_pages + pages_in_stripe) /
 				    stripes;
 
@@ -587,9 +587,9 @@ static int _prepare_one_group(struct objio_state *ios, u64 length,
 			      struct _striping_info *si, unsigned first_comp,
 			      unsigned *last_pg)
 {
-	unsigned stripe_unit = ios->objio_seg->stripe_unit;
-	unsigned mirrors_p1 = ios->objio_seg->mirrors_p1;
-	unsigned devs_in_group = ios->objio_seg->group_width * mirrors_p1;
+	unsigned stripe_unit = ios->layout->stripe_unit;
+	unsigned mirrors_p1 = ios->layout->mirrors_p1;
+	unsigned devs_in_group = ios->layout->group_width * mirrors_p1;
 	unsigned dev = si->dev;
 	unsigned first_dev = dev - (dev % devs_in_group);
 	unsigned comp = first_comp + (dev - first_dev);
@@ -650,10 +650,10 @@ static int _io_rw_pagelist(struct objio_state *ios)
 {
 	u64 length = ios->ol_state.count;
 	struct _striping_info si;
-	unsigned devs_in_group = ios->objio_seg->group_width *
-				 ios->objio_seg->mirrors_p1;
+	unsigned devs_in_group = ios->layout->group_width *
+				 ios->layout->mirrors_p1;
 	unsigned first_comp = 0;
-	unsigned num_comps = devs_in_group * ios->objio_seg->group_count;
+	unsigned num_comps = devs_in_group * ios->layout->group_count;
 	unsigned last_pg = 0;
 	int ret = 0;
 
@@ -672,8 +672,8 @@ static int _io_rw_pagelist(struct objio_state *ios)
 		si.group_length = si.total_group_length;
 		si.unit_off = 0;
 		++si.Major;
-		si.obj_offset = si.Major * ios->objio_seg->stripe_unit *
-						ios->objio_seg->group_depth;
+		si.obj_offset = si.Major * ios->layout->stripe_unit *
+						ios->layout->group_depth;
 
 		si.dev = (si.dev - (si.dev % devs_in_group)) + devs_in_group;
 		si.dev %= num_comps;
@@ -771,7 +771,7 @@ static int _read_mirrors(struct objio_state *ios, unsigned cur_comp)
 	struct _objio_per_comp *per_dev = &ios->per_dev[cur_comp];
 	unsigned dev = per_dev->dev;
 	struct pnfs_osd_object_cred *cred =
-			&ios->objio_seg->comps[dev];
+			&ios->layout->comps[dev];
 	struct osd_obj_id obj = {
 		.partition = cred->oc_object_id.oid_partition_id,
 		.id = cred->oc_object_id.oid_object_id,
@@ -807,7 +807,7 @@ static ssize_t _read_exec(struct objio_state *ios)
 	unsigned i;
 	int ret;
 
-	for (i = 0; i < ios->numdevs; i += ios->objio_seg->mirrors_p1) {
+	for (i = 0; i < ios->numdevs; i += ios->layout->mirrors_p1) {
 		if (!ios->per_dev[i].length)
 			continue;
 		ret = _read_mirrors(ios, i);
@@ -863,13 +863,13 @@ static int _write_mirrors(struct objio_state *ios, unsigned cur_comp)
 {
 	struct _objio_per_comp *master_dev = &ios->per_dev[cur_comp];
 	unsigned dev = ios->per_dev[cur_comp].dev;
-	unsigned last_comp = cur_comp + ios->objio_seg->mirrors_p1;
+	unsigned last_comp = cur_comp + ios->layout->mirrors_p1;
 	int ret;
 
 	for (; cur_comp < last_comp; ++cur_comp, ++dev) {
 		struct osd_request *or = NULL;
 		struct pnfs_osd_object_cred *cred =
-					&ios->objio_seg->comps[dev];
+					&ios->layout->comps[dev];
 		struct osd_obj_id obj = {
 			.partition = cred->oc_object_id.oid_partition_id,
 			.id = cred->oc_object_id.oid_object_id,
@@ -929,7 +929,7 @@ static ssize_t _write_exec(struct objio_state *ios)
 	unsigned i;
 	int ret;
 
-	for (i = 0; i < ios->numdevs; i += ios->objio_seg->mirrors_p1) {
+	for (i = 0; i < ios->numdevs; i += ios->layout->mirrors_p1) {
 		if (!ios->per_dev[i].length)
 			continue;
 		ret = _write_mirrors(ios, i);
-- 
1.7.2.3

[PATCH 05/13] SQUASHME: pnfs-obj: Convert layout and deviceinfo decoding to new XDR

[Boaz Harrosh]

Change layout_decoding API into two parts, avoid double pass
and allocations.

It is better for users to first inspect the map information
before attempting to decoding the comps array. This is so
the proper dynamic allocations can be made.

So break up pnfs_osd_xdr_decode_layout() into:
	1. pnfs_osd_xdr_decode_layout_map()
	2. while (pnfs_osd_xdr_decode_layout_comp()) ...

With new API it is easy to convert every thing to a single
pass and use the new XDR API.

get deviceinfo is not using an xdr_stream but is also converted
to more uptodate coding style.

Signed-off-by: Boaz Harrosh <bharrosh@panasas.com>
---
 fs/nfs/objlayout/objlayout.c        |   60 +++++--
 fs/nfs/objlayout/objlayout.h        |    2 +-
 fs/nfs/objlayout/pnfs_osd_xdr_cli.c |  349 +++++++++++++++++------------------
 include/linux/pnfs_osd_xdr.h        |  181 ++++++-------------
 4 files changed, 271 insertions(+), 321 deletions(-)

diff --git a/fs/nfs/objlayout/objlayout.c b/fs/nfs/objlayout/objlayout.c
index a04e232..1ab9bdc 100644
--- a/fs/nfs/objlayout/objlayout.c
+++ b/fs/nfs/objlayout/objlayout.c
@@ -75,6 +75,28 @@ objlayout_free_layout_hdr(struct pnfs_layout_hdr *lo)
 	kfree(objlay);
 }
 
+struct caps_buffers {
+	u8 caps_key[OSD_CRYPTO_KEYID_SIZE];
+	u8 creds[OSD_CAP_LEN];
+};
+
+static void copy_single_comp(struct pnfs_osd_object_cred *cur_comp,
+			     struct pnfs_osd_object_cred *src_comp,
+			     struct caps_buffers *caps_p)
+{
+	WARN_ON(src_comp->oc_cap_key.cred_len > sizeof(caps_p->caps_key) >> 2);
+	WARN_ON(src_comp->oc_cap.cred_len > sizeof(caps_p->creds) >> 2);
+	*cur_comp = *src_comp;
+
+	memcpy(caps_p->caps_key, src_comp->oc_cap_key.cred,
+	       sizeof(caps_p->caps_key));
+	cur_comp->oc_cap_key.cred = caps_p->caps_key;
+
+	memcpy(caps_p->creds, src_comp->oc_cap.cred,
+	       sizeof(caps_p->creds));
+	cur_comp->oc_cap.cred = caps_p->creds;
+}
+
 /*
  * Unmarshall layout and store it in pnfslay.
  */
@@ -91,9 +113,11 @@ objlayout_alloc_lseg(struct pnfs_layout_hdr *pnfslay,
 		.len = lgr->layoutp->len,
 	};
 	struct page *scratch;
-	__be32 *p;
 	struct objlayout_segment *objlseg = NULL;
-	struct pnfs_osd_layout *pnfs_osd_layout;
+	struct pnfs_osd_layout *layout;
+	struct pnfs_osd_object_cred *cur_comp, src_comp;
+	struct caps_buffers *caps_p;
+	struct pnfs_osd_xdr_decode_layout_iter iter;
 
 	dprintk("%s: Begin pnfslay %p\n", __func__, pnfslay);
 
@@ -104,21 +128,32 @@ objlayout_alloc_lseg(struct pnfs_layout_hdr *pnfslay,
 	xdr_init_decode(&stream, &buf, NULL);
 	xdr_set_scratch_buffer(&stream, page_address(scratch), PAGE_SIZE);
 
-	p = xdr_inline_decode(&stream, pnfs_osd_data_map_xdr_sz() << 2);
-	if (unlikely(!p))
+	layout = &objlseg->layout;
+	status = pnfs_osd_xdr_decode_layout_map(layout, &iter, &stream);
+	if (unlikely(status))
 		goto err;
 
 	objlseg = kzalloc(sizeof(*objlseg) +
-			  pnfs_osd_layout_incore_sz(p), GFP_KERNEL);
-	if (!objlseg)
+			  sizeof(*layout->olo_comps) * layout->olo_num_comps +
+			  sizeof(struct caps_buffers) * layout->olo_num_comps,
+			  GFP_KERNEL);
+	if (unlikely(!objlseg)) {
+		status = -ENOMEM;
 		goto err;
+	}
 
-	pnfs_osd_layout = (struct pnfs_osd_layout *)objlseg->pnfs_osd_layout;
-	pnfs_osd_xdr_decode_layout(pnfs_osd_layout, p);
+	cur_comp = layout->olo_comps = (void *)(objlseg + 1);
+	caps_p = (void *)(cur_comp + layout->olo_num_comps);
+	while(pnfs_osd_xdr_decode_layout_comp(&src_comp, &iter, &stream,
+					      &status)) {
+		copy_single_comp(cur_comp++, &src_comp, caps_p++);
+	}
+	if (unlikely(status))
+		goto err;
 
 	objlseg->lseg.pls_range = lgr->range;
 	status = objio_alloc_lseg(&objlseg->internal, pnfslay, &objlseg->lseg,
-				  pnfs_osd_layout);
+				  layout);
 	if (status)
 		goto err;
 
@@ -263,8 +298,7 @@ objlayout_io_set_result(struct objlayout_io_state *state, unsigned index,
 
 	BUG_ON(index >= state->num_comps);
 	if (osd_error) {
-		struct pnfs_osd_layout *layout =
-			(typeof(layout))state->objlseg->pnfs_osd_layout;
+		struct pnfs_osd_layout *layout = &state->objlseg->layout;
 
 		ioerr->oer_component = layout->olo_comps[index].oc_object_id;
 		ioerr->oer_comp_offset = offset;
@@ -674,7 +708,6 @@ int objlayout_get_deviceinfo(struct pnfs_layout_hdr *pnfslay,
 	struct pnfs_device pd;
 	struct super_block *sb;
 	struct page *page, **pages;
-	size_t sz;
 	u32 *p;
 	int err;
 
@@ -699,8 +732,7 @@ int objlayout_get_deviceinfo(struct pnfs_layout_hdr *pnfslay,
 		goto err_out;
 
 	p = page_address(page);
-	sz = pnfs_osd_xdr_deviceaddr_incore_sz(p);
-	odi = kzalloc(sz + (sizeof(*odi) - sizeof(odi->da)), GFP_KERNEL);
+	odi = kzalloc(sizeof(*odi), GFP_KERNEL);
 	if (!odi) {
 		err = -ENOMEM;
 		goto err_out;
diff --git a/fs/nfs/objlayout/objlayout.h b/fs/nfs/objlayout/objlayout.h
index 38abb01..6ea54f6 100644
--- a/fs/nfs/objlayout/objlayout.h
+++ b/fs/nfs/objlayout/objlayout.h
@@ -51,7 +51,7 @@
 struct objlayout_segment {
 	struct pnfs_layout_segment lseg;
 	void *internal;    /* for provider internal use */
-	u8 pnfs_osd_layout[];
+	struct pnfs_osd_layout layout;
 };
 
 /*
diff --git a/fs/nfs/objlayout/pnfs_osd_xdr_cli.c b/fs/nfs/objlayout/pnfs_osd_xdr_cli.c
index 4dea458..b55131c 100644
--- a/fs/nfs/objlayout/pnfs_osd_xdr_cli.c
+++ b/fs/nfs/objlayout/pnfs_osd_xdr_cli.c
@@ -5,6 +5,7 @@
  *  All rights reserved.
  *
  *  Benny Halevy <bhalevy@panasas.com>
+ *  Boaz Harrosh <bharrosh@panasas.com>
  *
  *  This program is free software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License version 2
@@ -49,24 +50,42 @@
  * 	struct pnfs_deviceid	oid_device_id;
  * 	u64			oid_partition_id;
  * 	u64			oid_object_id;
- * };
+ * }; // xdr size 32 bytes
  */
-static inline __be32 *
-pnfs_osd_xdr_decode_objid(__be32 *p, struct pnfs_osd_objid *objid)
+static __be32 *
+_osd_xdr_decode_objid(__be32 *p, struct pnfs_osd_objid *objid)
 {
-	COPYMEM(objid->oid_device_id.data, sizeof(objid->oid_device_id.data));
-	READ64(objid->oid_partition_id);
-	READ64(objid->oid_object_id);
+	p = xdr_decode_opaque_fixed(p, objid->oid_device_id.data,
+				    sizeof(objid->oid_device_id.data));
+
+	p = xdr_decode_hyper(p, &objid->oid_partition_id);
+	p = xdr_decode_hyper(p, &objid->oid_object_id);
 	return p;
 }
-
-static inline __be32 *
-pnfs_osd_xdr_decode_opaque_cred(__be32 *p,
-				struct pnfs_osd_opaque_cred *opaque_cred)
+/*
+ * struct pnfs_osd_opaque_cred {
+ * 	u32 cred_len;
+ * 	void *cred;
+ * }; // xdr size [variable]
+ * The return pointers are from the xdr buffer
+ */
+static int
+_osd_xdr_decode_opaque_cred(struct pnfs_osd_opaque_cred *opaque_cred,
+			    struct xdr_stream *xdr)
 {
-	READ32(opaque_cred->cred_len);
-	COPYMEM(opaque_cred->cred, opaque_cred->cred_len);
-	return p;
+	__be32 *p = xdr_inline_decode(xdr, 1);
+
+	if (!p)
+		return -EINVAL;
+
+	opaque_cred->cred_len = be32_to_cpu(*p++);
+
+	p = xdr_inline_decode(xdr, opaque_cred->cred_len);
+	if (!p)
+		return -EINVAL;
+
+	opaque_cred->cred = p;
+	return 0;
 }
 
 /*
@@ -76,28 +95,28 @@ pnfs_osd_xdr_decode_opaque_cred(__be32 *p,
  * 	u32				oc_cap_key_sec;
  * 	struct pnfs_osd_opaque_cred	oc_cap_key
  * 	struct pnfs_osd_opaque_cred	oc_cap;
- * };
+ * }; // xdr size 32 + 4 + 4 + [variable] + [variable]
  */
-static inline __be32 *
-pnfs_osd_xdr_decode_object_cred(__be32 *p, struct pnfs_osd_object_cred *comp,
-				u8 **credp)
+static int
+_osd_xdr_decode_object_cred(struct pnfs_osd_object_cred *comp,
+			    struct xdr_stream *xdr)
 {
-	u8 *cred;
+	__be32 *p = xdr_inline_decode(xdr, 32 + 4 + 4);
+	int ret;
+
+	if (!p)
+		return -EIO;
 
-	p = pnfs_osd_xdr_decode_objid(p, &comp->oc_object_id);
-	READ32(comp->oc_osd_version);
-	READ32(comp->oc_cap_key_sec);
+	p = _osd_xdr_decode_objid(p, &comp->oc_object_id);
+	comp->oc_osd_version = be32_to_cpup(p++);
+	comp->oc_cap_key_sec = be32_to_cpup(p);
 
-	cred = *credp;
-	comp->oc_cap_key.cred = cred;
-	p = pnfs_osd_xdr_decode_opaque_cred(p, &comp->oc_cap_key);
-	cred = (u8 *)((u32 *)cred + XDR_QUADLEN(comp->oc_cap_key.cred_len));
-	comp->oc_cap.cred = cred;
-	p = pnfs_osd_xdr_decode_opaque_cred(p, &comp->oc_cap);
-	cred = (u8 *)((u32 *)cred + XDR_QUADLEN(comp->oc_cap.cred_len));
-	*credp = cred;
+	ret = _osd_xdr_decode_opaque_cred(&comp->oc_cap_key, xdr);
+	if (unlikely(ret))
+		return ret;
 
-	return p;
+	ret = _osd_xdr_decode_opaque_cred(&comp->oc_cap, xdr);
+	return ret;
 }
 
 /*
@@ -108,17 +127,23 @@ pnfs_osd_xdr_decode_object_cred(__be32 *p, struct pnfs_osd_object_cred *comp,
  * 	u32	odm_group_depth;
  * 	u32	odm_mirror_cnt;
  * 	u32	odm_raid_algorithm;
- * };
+ * }; // xdr size 4 + 8 + 4 + 4 + 4 + 4
  */
-static inline u32 *
-pnfs_osd_xdr_decode_data_map(__be32 *p, struct pnfs_osd_data_map *data_map)
+static inline int
+_osd_data_map_xdr_sz(void)
 {
-	READ32(data_map->odm_num_comps);
-	READ64(data_map->odm_stripe_unit);
-	READ32(data_map->odm_group_width);
-	READ32(data_map->odm_group_depth);
-	READ32(data_map->odm_mirror_cnt);
-	READ32(data_map->odm_raid_algorithm);
+	return 4 + 8 + 4 + 4 + 4 + 4;
+}
+
+static __be32 *
+_osd_xdr_decode_data_map(__be32 *p, struct pnfs_osd_data_map *data_map)
+{
+	data_map->odm_num_comps = be32_to_cpup(p++);
+	p = xdr_decode_hyper(p, &data_map->odm_stripe_unit);
+	data_map->odm_group_width = be32_to_cpup(p++);
+	data_map->odm_group_depth = be32_to_cpup(p++);
+	data_map->odm_mirror_cnt = be32_to_cpup(p++);
+	data_map->odm_raid_algorithm = be32_to_cpup(p++);
 	dprintk("%s: odm_num_comps=%u odm_stripe_unit=%llu odm_group_width=%u "
 		"odm_group_depth=%u odm_mirror_cnt=%u odm_raid_algorithm=%u\n",
 		__func__,
@@ -131,92 +156,74 @@ pnfs_osd_xdr_decode_data_map(__be32 *p, struct pnfs_osd_data_map *data_map)
 	return p;
 }
 
-struct pnfs_osd_layout *
-pnfs_osd_xdr_decode_layout(struct pnfs_osd_layout *layout, __be32 *p)
+int pnfs_osd_xdr_decode_layout_map(struct pnfs_osd_layout *layout,
+	struct pnfs_osd_xdr_decode_layout_iter *iter, struct xdr_stream *xdr)
+{
+	__be32 *p;
+
+	memset(iter, 0, sizeof(*iter));
+
+	p = xdr_inline_decode(xdr, _osd_data_map_xdr_sz() + 4 + 4);
+	if (unlikely(!p))
+		return -EINVAL;
+
+	p = _osd_xdr_decode_data_map(p, &layout->olo_map);
+	layout->olo_comps_index = be32_to_cpup(p++);
+	layout->olo_num_comps = be32_to_cpup(p++);
+	iter->total_comps = layout->olo_num_comps;
+	return 0;
+}
+
+bool pnfs_osd_xdr_decode_layout_comp(struct pnfs_osd_object_cred *comp,
+	struct pnfs_osd_xdr_decode_layout_iter *iter, struct xdr_stream *xdr,
+	int *err)
 {
-	int i;
-	__be32 *start = p;
-	struct pnfs_osd_object_cred *comp;
-	u8 *cred;
-
-	p = pnfs_osd_xdr_decode_data_map(p, &layout->olo_map);
-	READ32(layout->olo_comps_index);
-	READ32(layout->olo_num_comps);
-	layout->olo_comps = (struct pnfs_osd_object_cred *)(layout + 1);
-	comp = layout->olo_comps;
-	cred = (u8 *)(comp + layout->olo_num_comps);
-	dprintk("%s: comps_index=%u num_comps=%u\n",
-		__func__, layout->olo_comps_index, layout->olo_num_comps);
-	for (i = 0; i < layout->olo_num_comps; i++) {
-		p = pnfs_osd_xdr_decode_object_cred(p, comp, &cred);
-		dprintk("%s: comp[%d]=dev(%llx:%llx) par=0x%llx obj=0x%llx "
-			"key_len=%u cap_len=%u\n",
-			__func__, i,
-			_DEVID_LO(&comp->oc_object_id.oid_device_id),
-			_DEVID_HI(&comp->oc_object_id.oid_device_id),
-			comp->oc_object_id.oid_partition_id,
-			comp->oc_object_id.oid_object_id,
-			comp->oc_cap_key.cred_len, comp->oc_cap.cred_len);
-		comp++;
+	BUG_ON(iter->decoded_comps > iter->total_comps);
+	if (iter->decoded_comps == iter->total_comps)
+		return false;
+
+	*err = _osd_xdr_decode_object_cred(comp, xdr);
+	if (unlikely(*err)) {
+		dprintk("%s: _osd_xdr_decode_object_cred=>%d decoded_comps=%d "
+			"total_comps=%d\n",__func__, *err,
+			iter->decoded_comps, iter->total_comps);
+		return false; /* stop the loop */
 	}
-	dprintk("%s: xdr_size=%Zd end=%p in_core_size=%Zd\n", __func__,
-	       (char *)p - (char *)start, cred, (char *)cred - (char *)layout);
-	return layout;
+	dprintk("%s: dev(%llx:%llx) par=0x%llx obj=0x%llx "
+		"key_len=%u cap_len=%u\n",
+		__func__,
+		_DEVID_LO(&comp->oc_object_id.oid_device_id),
+		_DEVID_HI(&comp->oc_object_id.oid_device_id),
+		comp->oc_object_id.oid_partition_id,
+		comp->oc_object_id.oid_object_id,
+		comp->oc_cap_key.cred_len, comp->oc_cap.cred_len);
+
+	iter->decoded_comps++;
+	return true;
 }
 
 /*
  * Get Device Information Decoding
  *
- * Note: since Device Information is currently done synchronously, most
- *       of the actual fields are left inside the rpc buffer and are only
+ * Note: since Device Information is currently done synchronously, all
+ *       variable strings fields are left inside the rpc buffer and are only
  *       pointed to by the pnfs_osd_deviceaddr members. So the read buffer
  *       should not be freed while the returned information is in use.
  */
-
-__be32 *__xdr_read_calc_nfs4_string(
-	__be32 *p, struct nfs4_string *str, u8 **freespace)
-{
-	u32 len;
-	char *data;
-	bool need_copy;
-
-	READ32(len);
-	data = (char *)p;
-
-	if (data[len]) { /* Not null terminated we'll need extra space */
-		data = *freespace;
-		*freespace += len + 1;
-		need_copy = true;
-	} else {
-		need_copy = false;
-	}
-
-	if (str) {
-		str->len = len;
-		str->data = data;
-		if (need_copy) {
-			memcpy(data, p, len);
-			data[len] = 0;
-		}
-	}
-
-	p += XDR_QUADLEN(len);
-	return p;
-}
-
-__be32 *__xdr_read_calc_u8_opaque(
-	__be32 *p, struct nfs4_string *str)
+/*
+ *struct nfs4_string {
+ *	unsigned int len;
+ *	char *data;
+ *}; // size [variable]
+ * NOTE: Returned string points to inside the XDR buffer
+ */
+static __be32 *
+__read_u8_opaque(__be32 *p, struct nfs4_string *str)
 {
-	u32 len;
-
-	READ32(len);
+	str->len = be32_to_cpup(p++);
+	str->data = (char *)p;
 
-	if (str) {
-		str->len = len;
-		str->data = (char *)p;
-	}
-
-	p += XDR_QUADLEN(len);
+	p += XDR_QUADLEN(str->len);
 	return p;
 }
 
@@ -224,22 +231,20 @@ __be32 *__xdr_read_calc_u8_opaque(
  * struct pnfs_osd_targetid {
  * 	u32			oti_type;
  * 	struct nfs4_string	oti_scsi_device_id;
- * };
+ * };// size 4 + [variable]
  */
-__be32 *__xdr_read_calc_targetid(
-	__be32 *p, struct pnfs_osd_targetid* targetid, u8 **freespace)
+static __be32 *
+__read_targetid(__be32 *p, struct pnfs_osd_targetid* targetid)
 {
 	u32 oti_type;
 
-	READ32(oti_type);
-	if (targetid)
-		targetid->oti_type = oti_type;
+	oti_type = be32_to_cpup(p++);
+	targetid->oti_type = oti_type;
 
 	switch (oti_type) {
 	case OBJ_TARGET_SCSI_NAME:
 	case OBJ_TARGET_SCSI_DEVICE_ID:
-		p = __xdr_read_calc_u8_opaque(p,
-			targetid ? &targetid->oti_scsi_device_id : NULL);
+		p = __read_u8_opaque(p, &targetid->oti_scsi_device_id);
 	}
 
 	return p;
@@ -251,17 +256,11 @@ __be32 *__xdr_read_calc_targetid(
  * 	struct nfs4_string	r_addr;
  * };
  */
-__be32 *__xdr_read_calc_net_addr(
-	__be32 *p, struct pnfs_osd_net_addr* netaddr, u8 **freespace)
+static __be32 *
+__read_net_addr(__be32 *p, struct pnfs_osd_net_addr* netaddr)
 {
-
-	p = __xdr_read_calc_nfs4_string(p,
-			netaddr ? &netaddr->r_netid : NULL,
-			freespace);
-
-	p = __xdr_read_calc_nfs4_string(p,
-			netaddr ? &netaddr->r_addr : NULL,
-			freespace);
+	p = __read_u8_opaque(p, &netaddr->r_netid);
+	p = __read_u8_opaque(p, &netaddr->r_addr);
 
 	return p;
 }
@@ -272,19 +271,16 @@ __be32 *__xdr_read_calc_net_addr(
  * 	struct pnfs_osd_net_addr	ota_netaddr;
  * };
  */
-__be32 *__xdr_read_calc_targetaddr(
-	__be32 *p, struct pnfs_osd_targetaddr *targetaddr, u8 **freespace)
+static __be32 *
+__read_targetaddr(__be32 *p, struct pnfs_osd_targetaddr *targetaddr)
 {
 	u32 ota_available;
 
-	READ32(ota_available);
-	if (targetaddr)
-		targetaddr->ota_available = ota_available;
+	ota_available = be32_to_cpup(p++);
+	targetaddr->ota_available = ota_available;
 
 	if (ota_available) {
-		p = __xdr_read_calc_net_addr(p,
-				targetaddr ? &targetaddr->ota_netaddr : NULL,
-				freespace);
+		p = __read_net_addr(p, &targetaddr->ota_netaddr);
 	}
 
 	return p;
@@ -300,56 +296,49 @@ __be32 *__xdr_read_calc_targetaddr(
  * 	struct nfs4_string		oda_osdname;
  * };
  */
-__be32 *__xdr_read_calc_deviceaddr(
-	__be32 *p, struct pnfs_osd_deviceaddr *deviceaddr, u8 **freespace)
-{
-	p = __xdr_read_calc_targetid(p,
-			deviceaddr ? &deviceaddr->oda_targetid : NULL,
-			freespace);
-
-	p = __xdr_read_calc_targetaddr(p,
-			deviceaddr ? &deviceaddr->oda_targetaddr : NULL,
-			freespace);
-
-	if (deviceaddr)
-		COPYMEM(deviceaddr->oda_lun, sizeof(deviceaddr->oda_lun));
-	else
-		p += XDR_QUADLEN(sizeof(deviceaddr->oda_lun));
-
-	p = __xdr_read_calc_u8_opaque(p,
-			deviceaddr ? &deviceaddr->oda_systemid : NULL);
-
-	if (deviceaddr) {
-		p = pnfs_osd_xdr_decode_object_cred(p,
-				&deviceaddr->oda_root_obj_cred, freespace);
-	} else {
-		*freespace += pnfs_osd_object_cred_incore_sz(p);
-		p += pnfs_osd_object_cred_xdr_sz(p);
-	}
-
-	p = __xdr_read_calc_u8_opaque(p,
-			deviceaddr ? &deviceaddr->oda_osdname : NULL);
 
-	return p;
+/* We need this version for the pnfs_osd_xdr_decode_deviceaddr which does
+ * not have an xdr_stream
+ */
+static __be32 *
+__read_opaque_cred(__be32 *p,
+			      struct pnfs_osd_opaque_cred *opaque_cred)
+{
+	opaque_cred->cred_len = be32_to_cpu(*p++);
+	opaque_cred->cred = p;
+	return p + XDR_QUADLEN(opaque_cred->cred_len);
 }
 
-size_t pnfs_osd_xdr_deviceaddr_incore_sz(__be32 *p)
+static __be32 *
+__read_object_cred(__be32 *p, struct pnfs_osd_object_cred *comp)
 {
-	u8 *null_freespace = NULL;
-	size_t sz;
-
-	__xdr_read_calc_deviceaddr(p, NULL, &null_freespace);
-	sz = sizeof(struct pnfs_osd_deviceaddr) + (size_t)null_freespace;
+	p = _osd_xdr_decode_objid(p, &comp->oc_object_id);
+	comp->oc_osd_version = be32_to_cpup(p++);
+	comp->oc_cap_key_sec = be32_to_cpup(p++);
 
-	return sz;
+	p = __read_opaque_cred(p, &comp->oc_cap_key);
+	p = __read_opaque_cred(p, &comp->oc_cap);
+	return p;
 }
 
 void pnfs_osd_xdr_decode_deviceaddr(
 	struct pnfs_osd_deviceaddr *deviceaddr, __be32 *p)
 {
-	u8 *freespace = (u8 *)(deviceaddr + 1);
+	p = __read_targetid(p, &deviceaddr->oda_targetid);
+
+	p = __read_targetaddr(p, &deviceaddr->oda_targetaddr);
+
+	p = xdr_decode_opaque_fixed(p, deviceaddr->oda_lun,
+				    sizeof(deviceaddr->oda_lun));
+
+	p = __read_u8_opaque(p, &deviceaddr->oda_systemid);
 
-	__xdr_read_calc_deviceaddr(p, deviceaddr, &freespace);
+	p = __read_object_cred(p, &deviceaddr->oda_root_obj_cred);
+
+	p = __read_u8_opaque(p, &deviceaddr->oda_osdname);
+
+	/* libosd likes this terminated in dbg. It's last, so no problems */
+	deviceaddr->oda_osdname.data[deviceaddr->oda_osdname.len] = 0;
 }
 
 /*
@@ -357,13 +346,13 @@ void pnfs_osd_xdr_decode_deviceaddr(
  * 	u32	dsu_valid;
  * 	s64	dsu_delta;
  * 	u32	olu_ioerr_flag;
- * };
+ * }; xdr size 4 + 8 + 4
  */
 int
 pnfs_osd_xdr_encode_layoutupdate(struct xdr_stream *xdr,
 				 struct pnfs_osd_layoutupdate *lou)
 {
-	__be32 *p = xdr_reserve_space(xdr, 16);
+	__be32 *p = xdr_reserve_space(xdr,  4 + 8 + 4);
 
 	if (!p)
 		return -E2BIG;
diff --git a/include/linux/pnfs_osd_xdr.h b/include/linux/pnfs_osd_xdr.h
index aed693f..e4c2b25 100644
--- a/include/linux/pnfs_osd_xdr.h
+++ b/include/linux/pnfs_osd_xdr.h
@@ -5,6 +5,7 @@
  *  All rights reserved.
  *
  *  Benny Halevy <bhalevy@panasas.com>
+ *  Boaz Harrosh <bharrosh@panasas.com>
  *
  *  This program is free software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License version 2
@@ -46,26 +47,6 @@
 #define PNFS_OSD_OSDNAME_MAXSIZE 256
 
 /*
- * START OF "GENERIC" DECODE ROUTINES.
- *   These may look a little ugly since they are imported from a "generic"
- * set of XDR encode/decode routines which are intended to be shared by
- * all of our NFSv4 implementations (OpenBSD, MacOS X...).
- *
- * If the pain of reading these is too great, it should be a straightforward
- * task to translate them into Linux-specific versions which are more
- * consistent with the style used in NFSv2/v3...
- */
-#define READ32(x)         (x) = ntohl(*p++)
-#define READ64(x)         do {			\
-	(x) = (u64)ntohl(*p++) << 32;		\
-	(x) |= ntohl(*p++);			\
-} while (0)
-#define COPYMEM(x, nbytes) do {			\
-	memcpy((x), p, nbytes);			\
-	p += XDR_QUADLEN(nbytes);		\
-} while (0)
-
-/*
  * draft-ietf-nfsv4-minorversion-22
  * draft-ietf-nfsv4-pnfs-obj-12
  */
@@ -97,18 +78,6 @@ struct pnfs_osd_data_map {
 	u32	odm_raid_algorithm;
 };
 
-static inline int
-pnfs_osd_data_map_xdr_sz(void)
-{
-	return 1 + 2 + 1 + 1 + 1 + 1;
-}
-
-static inline size_t
-pnfs_osd_data_map_incore_sz(void)
-{
-	return sizeof(struct pnfs_osd_data_map);
-}
-
 /*   struct pnfs_osd_objid4 {
  *       deviceid4       oid_device_id;
  *       uint64_t        oid_partition_id;
@@ -121,12 +90,15 @@ struct pnfs_osd_objid {
 	u64			oid_object_id;
 };
 
-/* For printout. I use "dev(%llx:%llx)", _DEVID_LO(), _DEVID_HI BE style */
+/* For printout. I use:
+ * kprint("dev(%llx:%llx)", _DEVID_LO(pointer), _DEVID_HI(pointer));
+ * BE style
+ */
 #define _DEVID_LO(oid_device_id) \
-	(unsigned long long)be64_to_cpup((__be64 *)oid_device_id.data)
+	(unsigned long long)be64_to_cpup((__be64 *)(oid_device_id)->data)
 
 #define _DEVID_HI(oid_device_id) \
-	(unsigned long long)be64_to_cpup(((__be64 *)oid_device_id.data) + 1)
+	(unsigned long long)be64_to_cpup(((__be64 *)(oid_device_id)->data) + 1)
 
 static inline int
 pnfs_osd_objid_xdr_sz(void)
@@ -134,12 +106,6 @@ pnfs_osd_objid_xdr_sz(void)
 	return (NFS4_DEVICEID4_SIZE / 4) + 2 + 2;
 }
 
-static inline size_t
-pnfs_osd_objid_incore_sz(void)
-{
-	return sizeof(struct pnfs_osd_objid);
-}
-
 enum pnfs_osd_version {
 	PNFS_OSD_MISSING              = 0,
 	PNFS_OSD_VERSION_1            = 1,
@@ -148,29 +114,9 @@ enum pnfs_osd_version {
 
 struct pnfs_osd_opaque_cred {
 	u32 cred_len;
-	u8 *cred;
+	void *cred;
 };
 
-static inline int
-pnfs_osd_opaque_cred_xdr_sz(__be32 *p)
-{
-	u32 *start = p;
-	u32 n;
-
-	READ32(n);
-	p += XDR_QUADLEN(n);
-	return p - start;
-}
-
-static inline size_t
-pnfs_osd_opaque_cred_incore_sz(__be32 *p)
-{
-	u32 n;
-
-	READ32(n);
-	return XDR_QUADLEN(n) * 4;
-}
-
 enum pnfs_osd_cap_key_sec {
 	PNFS_OSD_CAP_KEY_SEC_NONE     = 0,
 	PNFS_OSD_CAP_KEY_SEC_SSV      = 1,
@@ -192,29 +138,6 @@ struct pnfs_osd_object_cred {
 	struct pnfs_osd_opaque_cred	oc_cap;
 };
 
-static inline int
-pnfs_osd_object_cred_xdr_sz(__be32 *p)
-{
-	__be32 *start = p;
-
-	p += pnfs_osd_objid_xdr_sz() + 2;
-	p += pnfs_osd_opaque_cred_xdr_sz(p);
-	p += pnfs_osd_opaque_cred_xdr_sz(p);
-	return p - start;
-}
-
-static inline size_t
-pnfs_osd_object_cred_incore_sz(__be32 *p)
-{
-	size_t sz = sizeof(struct pnfs_osd_object_cred);
-
-	p += pnfs_osd_objid_xdr_sz() + 2;
-	sz += pnfs_osd_opaque_cred_incore_sz(p);
-	p += pnfs_osd_opaque_cred_xdr_sz(p);
-	sz += pnfs_osd_opaque_cred_incore_sz(p);
-	return sz;
-}
-
 /*   struct pnfs_osd_layout4 {
  *       pnfs_osd_data_map4      olo_map;
  *       uint32_t                olo_comps_index;
@@ -228,37 +151,7 @@ struct pnfs_osd_layout {
 	struct pnfs_osd_object_cred	*olo_comps;
 };
 
-static inline int
-pnfs_osd_layout_xdr_sz(__be32 *p)
-{
-	__be32 *start = p;
-	u32 n;
-
-	p += pnfs_osd_data_map_xdr_sz() + 1;
-	READ32(n);
-	while ((int)(n--) > 0)
-		p += pnfs_osd_object_cred_xdr_sz(p);
-	return p - start;
-}
-
-static inline size_t
-pnfs_osd_layout_incore_sz(__be32 *p)
-{
-	u32 n;
-	size_t sz;
-
-	p += pnfs_osd_data_map_xdr_sz() + 1;
-	READ32(n);
-	sz = sizeof(struct pnfs_osd_layout);
-	while ((int)(n--) > 0) {
-		sz += pnfs_osd_object_cred_incore_sz(p);
-		p += pnfs_osd_object_cred_xdr_sz(p);
-	}
-	return sz;
-}
-
 /* Device Address */
-
 enum pnfs_osd_targetid_type {
 	OBJ_TARGET_ANON = 1,
 	OBJ_TARGET_SCSI_NAME = 2,
@@ -387,32 +280,68 @@ struct pnfs_osd_ioerr {
 	u32			oer_errno;
 };
 
+/* FIXME: remove */
 static inline unsigned
 pnfs_osd_ioerr_xdr_sz(void)
 {
 	return pnfs_osd_objid_xdr_sz() + 2 + 2 + 1 + 1;
 }
 
-/* OSD XDR API */
 
+/* OSD XDR API */
 /* Layout helpers */
-extern struct pnfs_osd_layout *pnfs_osd_xdr_decode_layout(
-	struct pnfs_osd_layout *layout, __be32 *p);
+/* Layout decoding is done in two parts:
+ * 1. First Call pnfs_osd_xdr_decode_layout_map to read in only the header part
+ *    of the layout. @iter members need not be initialized.
+ *    Returned:
+ *             @layout members are set. (@layout->olo_comps set to NULL).
+ *
+ *             Zero on success, or negative error if passed xdr is broken.
+ *
+ * 2. 2nd Call pnfs_osd_xdr_decode_layout_comp() in a loop until it returns
+ *    false, to decode the next component.
+ *    Returned:
+ *       true if there is more to decode or false if we are done or error.
+ *
+ * Example:
+ *	struct pnfs_osd_xdr_decode_layout_iter iter;
+ *	struct pnfs_osd_layout layout;
+ *	struct pnfs_osd_object_cred comp;
+ *	int status;
+ *
+ *	status = pnfs_osd_xdr_decode_layout_map(&layout, &iter, xdr);
+ *	if (unlikely(status))
+ *		goto err;
+ *	while(pnfs_osd_xdr_decode_layout_comp(&comp, &iter, xdr, &status)) {
+ *		// All of @comp strings point to inside the xdr_buffer
+ *		// or scrach buffer. Copy them out to user memory eg.
+ *		copy_single_comp(dest_comp++, &comp);
+ *	}
+ *	if (unlikely(status))
+ *		goto err;
+ */
+
+struct pnfs_osd_xdr_decode_layout_iter {
+	unsigned total_comps;
+	unsigned decoded_comps;
+};
 
+extern int pnfs_osd_xdr_decode_layout_map(struct pnfs_osd_layout *layout,
+	struct pnfs_osd_xdr_decode_layout_iter *iter, struct xdr_stream *xdr);
+
+extern bool pnfs_osd_xdr_decode_layout_comp(struct pnfs_osd_object_cred *comp,
+	struct pnfs_osd_xdr_decode_layout_iter *iter, struct xdr_stream *xdr,
+	int *err);
+
+/* Layout encoding */
 extern int pnfs_osd_xdr_encode_layout(
 	struct exp_xdr_stream *xdr,
 	struct pnfs_osd_layout *layout);
 
 /* Device Info helpers */
 
-/* First pass calculate total size for space needed */
-extern size_t pnfs_osd_xdr_deviceaddr_incore_sz(__be32 *p);
-
-/* Note: some strings pointed to inside @deviceaddr might point
- * to space inside @p. @p should stay valid while @deviceaddr
- * is in use.
- * It is assumed that @deviceaddr points to bigger memory of size
- * calculated in first pass by pnfs_osd_xdr_deviceaddr_incore_sz()
+/* Note: All strings inside @deviceaddr point to space inside @p.
+ * @p should stay valid while @deviceaddr is in use.
  */
 extern void pnfs_osd_xdr_decode_deviceaddr(
 	struct pnfs_osd_deviceaddr *deviceaddr, __be32 *p);
-- 
1.7.2.3

[PATCH 06/13] SQUASHME: pnfs-obj: Change API of objlayout_io_set_result

[Boaz Harrosh]

Remove the dependency of objlayout_io_set_result() on the existence of
a pnfs_osd_layout at the generic level, by passing the needed pnfs_osd_objid
in question directly. The io-driver has it at hand more readily. It also
decouples the error to be at the index of the original device and enables
a shorter io_err array with only the error devices.

Signed-off-by: Boaz Harrosh <bharrosh@panasas.com>
---
 fs/nfs/objlayout/objio_osd.c |    5 ++++-
 fs/nfs/objlayout/objlayout.c |    7 +++----
 fs/nfs/objlayout/objlayout.h |    4 ++--
 3 files changed, 9 insertions(+), 7 deletions(-)

diff --git a/fs/nfs/objlayout/objio_osd.c b/fs/nfs/objlayout/objio_osd.c
index 1e80f82..f3f7640 100644
--- a/fs/nfs/objlayout/objio_osd.c
+++ b/fs/nfs/objlayout/objio_osd.c
@@ -429,6 +429,7 @@ static int _io_check(struct objio_state *ios, bool is_write)
 	for (i = 0; i <  ios->numdevs; i++) {
 		struct osd_sense_info osi;
 		struct osd_request *or = ios->per_dev[i].or;
+		unsigned dev;
 		int ret;
 
 		if (!or)
@@ -449,7 +450,9 @@ static int _io_check(struct objio_state *ios, bool is_write)
 
 			continue; /* we recovered */
 		}
-		objlayout_io_set_result(&ios->ol_state, ios->per_dev[i].dev,
+		dev = ios->per_dev[i].dev;
+		objlayout_io_set_result(&ios->ol_state, dev,
+					&ios->layout->comps[dev].oc_object_id,
 					osd_pri_2_pnfs_err(osi.osd_err_pri),
 					ios->per_dev[i].offset,
 					ios->per_dev[i].length,
diff --git a/fs/nfs/objlayout/objlayout.c b/fs/nfs/objlayout/objlayout.c
index 1ab9bdc..c80f06c 100644
--- a/fs/nfs/objlayout/objlayout.c
+++ b/fs/nfs/objlayout/objlayout.c
@@ -292,15 +292,14 @@ objlayout_iodone(struct objlayout_io_state *state)
  */
 void
 objlayout_io_set_result(struct objlayout_io_state *state, unsigned index,
-			int osd_error, u64 offset, u64 length, bool is_write)
+			struct pnfs_osd_objid *pooid, int osd_error,
+			u64 offset, u64 length, bool is_write)
 {
 	struct pnfs_osd_ioerr *ioerr = &state->ioerrs[index];
 
 	BUG_ON(index >= state->num_comps);
 	if (osd_error) {
-		struct pnfs_osd_layout *layout = &state->objlseg->layout;
-
-		ioerr->oer_component = layout->olo_comps[index].oc_object_id;
+		ioerr->oer_component = *pooid;
 		ioerr->oer_comp_offset = offset;
 		ioerr->oer_comp_length = length;
 		ioerr->oer_iswrite = is_write;
diff --git a/fs/nfs/objlayout/objlayout.h b/fs/nfs/objlayout/objlayout.h
index 6ea54f6..01a3e4f 100644
--- a/fs/nfs/objlayout/objlayout.h
+++ b/fs/nfs/objlayout/objlayout.h
@@ -132,8 +132,8 @@ extern ssize_t objio_write_pagelist(struct objlayout_io_state *ol_state,
  * callback API
  */
 extern void objlayout_io_set_result(struct objlayout_io_state *state,
-				    unsigned index, int osd_error,
-				    u64 offset, u64 length, bool is_write);
+			unsigned index, struct pnfs_osd_objid *pooid,
+			int osd_error, u64 offset, u64 length, bool is_write);
 
 static inline void
 objlayout_add_delta_space_used(struct objlayout_io_state *state, s64 space_used)
-- 
1.7.2.3

[PATCH 07/13] SQUASHME: pnfs-obj: Avoid double allocation logic in objlayout_alloc_lseg

[Boaz Harrosh]

Push down objlayout_alloc_lseg() logic into the io-engines. This
avoids the generic layer from separately allocating the complete
pnfs_osd_layout structure, then allocating a second structure for
the io-engine's internal segment information. The io-engines have
better knowledge of what is actually needed and what can be
discarded, and can allocate everything at once.

The duplication of code, decoding the pnfs_osd_layout, is very small
since most of it is abstracted inside the _xdr_ layer routines.

TODO: Convert panlayout. It is mostly done (On the old version of
this) In my tree.

Signed-off-by: Boaz Harrosh <bharrosh@panasas.com>
---
 fs/nfs/objlayout/objio_osd.c |  102 +++++++++++++++++++++++++++++++-----------
 fs/nfs/objlayout/objlayout.c |   78 +++++---------------------------
 fs/nfs/objlayout/objlayout.h |   25 ++++-------
 3 files changed, 97 insertions(+), 108 deletions(-)

diff --git a/fs/nfs/objlayout/objio_osd.c b/fs/nfs/objlayout/objio_osd.c
index f3f7640..7e46d2b 100644
--- a/fs/nfs/objlayout/objio_osd.c
+++ b/fs/nfs/objlayout/objio_osd.c
@@ -128,7 +128,14 @@ out:
 	return 0;
 }
 
+struct caps_buffers {
+	u8 caps_key[OSD_CRYPTO_KEYID_SIZE];
+	u8 creds[OSD_CAP_LEN];
+};
+
 struct objio_segment {
+	struct pnfs_layout_segment lseg;
+
 	struct pnfs_osd_object_cred *comps;
 
 	unsigned mirrors_p1;
@@ -143,6 +150,12 @@ struct objio_segment {
 	struct osd_dev	*ods[1];
 };
 
+static inline struct objio_segment *
+OBJIO_LSEG(struct pnfs_layout_segment *lseg)
+{
+	return container_of(lseg, struct objio_segment, lseg);
+}
+
 struct objio_state;
 typedef ssize_t (*objio_done_fn)(struct objio_state *ios);
 
@@ -188,7 +201,8 @@ static struct osd_dev *_device_lookup(struct pnfs_layout_hdr *pnfslay,
 
 	err = objlayout_get_deviceinfo(pnfslay, d_id, &deviceaddr);
 	if (unlikely(err)) {
-		dprintk("%s: objlayout_get_deviceinfo=>%d\n", __func__, err);
+		dprintk("%s: objlayout_get_deviceinfo dev(%llx:%llx) =>%d \n",
+			__func__, _DEVID_LO(d_id), _DEVID_HI(d_id), err);
 		return ERR_PTR(err);
 	}
 
@@ -288,66 +302,102 @@ static int _verify_data_map(struct pnfs_osd_layout *layout)
 	return 0;
 }
 
-int objio_alloc_lseg(void **outp,
+static void copy_single_comp(struct pnfs_osd_object_cred *cur_comp,
+			     struct pnfs_osd_object_cred *src_comp,
+			     struct caps_buffers *caps_p)
+{
+	WARN_ON(src_comp->oc_cap_key.cred_len > sizeof(caps_p->caps_key));
+	WARN_ON(src_comp->oc_cap.cred_len > sizeof(caps_p->creds));
+
+	*cur_comp = *src_comp;
+
+	memcpy(caps_p->caps_key, src_comp->oc_cap_key.cred,
+	       sizeof(caps_p->caps_key));
+	cur_comp->oc_cap_key.cred = caps_p->caps_key;
+
+	memcpy(caps_p->creds, src_comp->oc_cap.cred,
+	       sizeof(caps_p->creds));
+	cur_comp->oc_cap.cred = caps_p->creds;
+}
+
+extern int objio_alloc_lseg(struct pnfs_layout_segment **outp,
 	struct pnfs_layout_hdr *pnfslay,
-	struct pnfs_layout_segment *lseg,
-	struct pnfs_osd_layout *layout)
+	struct pnfs_layout_range *range,
+	struct xdr_stream *xdr)
 {
 	struct objio_segment *objio_seg;
+	struct pnfs_osd_xdr_decode_layout_iter iter;
+	struct pnfs_osd_layout layout;
+	struct pnfs_osd_object_cred *cur_comp, src_comp;
+	struct caps_buffers *caps_p;
+
 	int err;
 
-	err = _verify_data_map(layout);
+	err = pnfs_osd_xdr_decode_layout_map(&layout, &iter, xdr);
+	if (unlikely(err))
+		return err;
+
+	err = _verify_data_map(&layout);
 	if (unlikely(err))
 		return err;
 
 	objio_seg = kzalloc(sizeof(*objio_seg) +
-			(layout->olo_num_comps - 1) * sizeof(objio_seg->ods[0]),
-			GFP_KERNEL);
+			    sizeof(*objio_seg->comps) * layout.olo_num_comps +
+			    sizeof(struct caps_buffers) * layout.olo_num_comps,
+			    GFP_KERNEL);
 	if (!objio_seg)
 		return -ENOMEM;
 
-	objio_seg->comps_index = layout->olo_comps_index;
-	objio_seg->num_comps = layout->olo_num_comps;
-	objio_seg->comps = layout->olo_comps;
+	cur_comp = objio_seg->comps = (void *)(objio_seg + 1);
+	caps_p = (void *)(cur_comp + layout.olo_num_comps);
+	while(pnfs_osd_xdr_decode_layout_comp(&src_comp, &iter, xdr, &err)) {
+		copy_single_comp(cur_comp++, &src_comp, caps_p++);
+	}
+	if (unlikely(err))
+		goto err;
+
+	objio_seg->num_comps = layout.olo_num_comps;
+	objio_seg->comps_index = layout.olo_comps_index;
 	err = objio_devices_lookup(pnfslay, objio_seg);
 	if (err)
-		goto free_seg;
-
-	objio_seg->mirrors_p1 = layout->olo_map.odm_mirror_cnt + 1;
-	objio_seg->stripe_unit = layout->olo_map.odm_stripe_unit;
-	if (layout->olo_map.odm_group_width) {
-		objio_seg->group_width = layout->olo_map.odm_group_width;
-		objio_seg->group_depth = layout->olo_map.odm_group_depth;
-		objio_seg->group_count = layout->olo_map.odm_num_comps /
+		goto err;
+
+	objio_seg->mirrors_p1 = layout.olo_map.odm_mirror_cnt + 1;
+	objio_seg->stripe_unit = layout.olo_map.odm_stripe_unit;
+	if (layout.olo_map.odm_group_width) {
+		objio_seg->group_width = layout.olo_map.odm_group_width;
+		objio_seg->group_depth = layout.olo_map.odm_group_depth;
+		objio_seg->group_count = layout.olo_map.odm_num_comps /
 						objio_seg->mirrors_p1 /
 						objio_seg->group_width;
 	} else {
-		objio_seg->group_width = layout->olo_map.odm_num_comps /
+		objio_seg->group_width = layout.olo_map.odm_num_comps /
 						objio_seg->mirrors_p1;
 		objio_seg->group_depth = -1;
 		objio_seg->group_count = 1;
 	}
 
-	*outp = objio_seg;
+	*outp = &objio_seg->lseg;
 	return 0;
 
-free_seg:
-	dprintk("%s: Error: return %d\n", __func__, err);
+err:
 	kfree(objio_seg);
+	dprintk("%s: Error: return %d\n", __func__, err);
 	*outp = NULL;
 	return err;
 }
 
-void objio_free_lseg(void *p)
+void objio_free_lseg(struct pnfs_layout_segment *lseg)
 {
-	struct objio_segment *objio_seg = p;
+	struct objio_segment *objio_seg = OBJIO_LSEG(lseg);
 
 	kfree(objio_seg);
 }
 
-int objio_alloc_io_state(void *seg, struct objlayout_io_state **outp)
+int objio_alloc_io_state(struct pnfs_layout_segment *lseg,
+			 struct objlayout_io_state **outp)
 {
-	struct objio_segment *objio_seg = seg;
+	struct objio_segment *objio_seg = OBJIO_LSEG(lseg);
 	struct objio_state *ios;
 	const unsigned first_size = sizeof(*ios) +
 				objio_seg->num_comps * sizeof(ios->per_dev[0]);
diff --git a/fs/nfs/objlayout/objlayout.c b/fs/nfs/objlayout/objlayout.c
index c80f06c..e3c42fc 100644
--- a/fs/nfs/objlayout/objlayout.c
+++ b/fs/nfs/objlayout/objlayout.c
@@ -75,28 +75,6 @@ objlayout_free_layout_hdr(struct pnfs_layout_hdr *lo)
 	kfree(objlay);
 }
 
-struct caps_buffers {
-	u8 caps_key[OSD_CRYPTO_KEYID_SIZE];
-	u8 creds[OSD_CAP_LEN];
-};
-
-static void copy_single_comp(struct pnfs_osd_object_cred *cur_comp,
-			     struct pnfs_osd_object_cred *src_comp,
-			     struct caps_buffers *caps_p)
-{
-	WARN_ON(src_comp->oc_cap_key.cred_len > sizeof(caps_p->caps_key) >> 2);
-	WARN_ON(src_comp->oc_cap.cred_len > sizeof(caps_p->creds) >> 2);
-	*cur_comp = *src_comp;
-
-	memcpy(caps_p->caps_key, src_comp->oc_cap_key.cred,
-	       sizeof(caps_p->caps_key));
-	cur_comp->oc_cap_key.cred = caps_p->caps_key;
-
-	memcpy(caps_p->creds, src_comp->oc_cap.cred,
-	       sizeof(caps_p->creds));
-	cur_comp->oc_cap.cred = caps_p->creds;
-}
-
 /*
  * Unmarshall layout and store it in pnfslay.
  */
@@ -113,11 +91,7 @@ objlayout_alloc_lseg(struct pnfs_layout_hdr *pnfslay,
 		.len = lgr->layoutp->len,
 	};
 	struct page *scratch;
-	struct objlayout_segment *objlseg = NULL;
-	struct pnfs_osd_layout *layout;
-	struct pnfs_osd_object_cred *cur_comp, src_comp;
-	struct caps_buffers *caps_p;
-	struct pnfs_osd_xdr_decode_layout_iter iter;
+	struct pnfs_layout_segment *lseg;
 
 	dprintk("%s: Begin pnfslay %p\n", __func__, pnfslay);
 
@@ -128,44 +102,22 @@ objlayout_alloc_lseg(struct pnfs_layout_hdr *pnfslay,
 	xdr_init_decode(&stream, &buf, NULL);
 	xdr_set_scratch_buffer(&stream, page_address(scratch), PAGE_SIZE);
 
-	layout = &objlseg->layout;
-	status = pnfs_osd_xdr_decode_layout_map(layout, &iter, &stream);
-	if (unlikely(status))
-		goto err;
-
-	objlseg = kzalloc(sizeof(*objlseg) +
-			  sizeof(*layout->olo_comps) * layout->olo_num_comps +
-			  sizeof(struct caps_buffers) * layout->olo_num_comps,
-			  GFP_KERNEL);
-	if (unlikely(!objlseg)) {
-		status = -ENOMEM;
+	status = objio_alloc_lseg(&lseg, pnfslay, &lgr->range, &stream);
+	if (unlikely(status)) {
+		dprintk("%s: objio_alloc_lseg Return err %d\n", __func__,
+			status);
 		goto err;
 	}
 
-	cur_comp = layout->olo_comps = (void *)(objlseg + 1);
-	caps_p = (void *)(cur_comp + layout->olo_num_comps);
-	while(pnfs_osd_xdr_decode_layout_comp(&src_comp, &iter, &stream,
-					      &status)) {
-		copy_single_comp(cur_comp++, &src_comp, caps_p++);
-	}
-	if (unlikely(status))
-		goto err;
-
-	objlseg->lseg.pls_range = lgr->range;
-	status = objio_alloc_lseg(&objlseg->internal, pnfslay, &objlseg->lseg,
-				  layout);
-	if (status)
-		goto err;
-
 	__free_page(scratch);
 
-	dprintk("%s: Return %p\n", __func__, &objlseg->lseg);
-	return &objlseg->lseg;
+	dprintk("%s: Return %p\n", __func__, lseg);
+	return lseg;
 
 err:
-	kfree(objlseg);
 	__free_page(scratch);
 err_nofree:
+	dprintk("%s: Err Return=>%d\n", __func__, status);
 	return ERR_PTR(status);
 }
 
@@ -175,16 +127,12 @@ err_nofree:
 void
 objlayout_free_lseg(struct pnfs_layout_segment *lseg)
 {
-	struct objlayout_segment *objlseg;
-
 	dprintk("%s: freeing layout segment %p\n", __func__, lseg);
 
 	if (unlikely(!lseg))
 		return;
 
-	objlseg = container_of(lseg, struct objlayout_segment, lseg);
-	objio_free_lseg(objlseg->internal);
-	kfree(objlseg);
+	objio_free_lseg(lseg);
 }
 
 /*
@@ -219,13 +167,11 @@ objlayout_alloc_io_state(struct pnfs_layout_hdr *pnfs_layout_type,
 			struct pnfs_layout_segment *lseg,
 			void *rpcdata)
 {
-	struct objlayout_segment *objlseg =
-		container_of(lseg, struct objlayout_segment, lseg);
 	struct objlayout_io_state *state;
 	u64 lseg_end_offset;
 
 	dprintk("%s: allocating io_state\n", __func__);
-	if (objio_alloc_io_state(objlseg->internal, &state))
+	if (objio_alloc_io_state(lseg, &state))
 		return NULL;
 
 	BUG_ON(offset < lseg->pls_range.offset);
@@ -242,7 +188,7 @@ objlayout_alloc_io_state(struct pnfs_layout_hdr *pnfs_layout_type,
 	}
 
 	INIT_LIST_HEAD(&state->err_list);
-	state->objlseg = objlseg;
+	state->lseg = lseg;
 	state->rpcdata = rpcdata;
 	state->pages = pages;
 	state->pgbase = pgbase;
@@ -275,7 +221,7 @@ objlayout_iodone(struct objlayout_io_state *state)
 	if (likely(state->status >= 0)) {
 		objlayout_free_io_state(state);
 	} else {
-		struct objlayout *objlay = OBJLAYOUT(state->objlseg->lseg.pls_layout);
+		struct objlayout *objlay = OBJLAYOUT(state->lseg->pls_layout);
 
 		spin_lock(&objlay->lock);
 		objlay->delta_space_valid = OBJ_DSU_INVALID;
diff --git a/fs/nfs/objlayout/objlayout.h b/fs/nfs/objlayout/objlayout.h
index 01a3e4f..83bc36f 100644
--- a/fs/nfs/objlayout/objlayout.h
+++ b/fs/nfs/objlayout/objlayout.h
@@ -46,15 +46,6 @@
 #include "../pnfs.h"
 
 /*
- * in-core layout segment
- */
-struct objlayout_segment {
-	struct pnfs_layout_segment lseg;
-	void *internal;    /* for provider internal use */
-	struct pnfs_osd_layout layout;
-};
-
-/*
  * per-inode layout
  */
 struct objlayout {
@@ -84,7 +75,7 @@ OBJLAYOUT(struct pnfs_layout_hdr *lo)
  * embedded in objects provider io_state data structure
  */
 struct objlayout_io_state {
-	struct objlayout_segment *objlseg;
+	struct pnfs_layout_segment *lseg;
 
 	struct page **pages;
 	unsigned pgbase;
@@ -115,13 +106,15 @@ struct objlayout_io_state {
 extern void *objio_init_mt(void);
 extern void objio_fini_mt(void *mt);
 
-extern int objio_alloc_lseg(void **outp,
+extern int objio_alloc_lseg(struct pnfs_layout_segment **outp,
 	struct pnfs_layout_hdr *pnfslay,
-	struct pnfs_layout_segment *lseg,
-	struct pnfs_osd_layout *layout);
-extern void objio_free_lseg(void *p);
+	struct pnfs_layout_range *range,
+	struct xdr_stream *xdr);
+extern void objio_free_lseg(struct pnfs_layout_segment *lseg);
 
-extern int objio_alloc_io_state(void *seg, struct objlayout_io_state **outp);
+extern int objio_alloc_io_state(
+	struct pnfs_layout_segment *lseg,
+	struct objlayout_io_state **outp);
 extern void objio_free_io_state(struct objlayout_io_state *state);
 
 extern ssize_t objio_read_pagelist(struct objlayout_io_state *ol_state);
@@ -138,7 +131,7 @@ extern void objlayout_io_set_result(struct objlayout_io_state *state,
 static inline void
 objlayout_add_delta_space_used(struct objlayout_io_state *state, s64 space_used)
 {
-	struct objlayout *objlay = OBJLAYOUT(state->objlseg->lseg.pls_layout);
+	struct objlayout *objlay = OBJLAYOUT(state->lseg->pls_layout);
 
 	/* If one of the I/Os errored out and the delta_space_used was
 	 * invalid we render the complete report as invalid. Protocol mandate
-- 
1.7.2.3

[PATCH 08/13] SQUASHME: pnfs_osd_xdr: Remove Server API declarations

[Boaz Harrosh]

Server API will be added later with it's implementation patch.

This also removes the dependency on the exp_xdr_stream header.

Signed-off-by: Boaz Harrosh <bharrosh@panasas.com>
---
 include/linux/pnfs_osd_xdr.h |   10 ----------
 1 files changed, 0 insertions(+), 10 deletions(-)

diff --git a/include/linux/pnfs_osd_xdr.h b/include/linux/pnfs_osd_xdr.h
index e4c2b25..db22471 100644
--- a/include/linux/pnfs_osd_xdr.h
+++ b/include/linux/pnfs_osd_xdr.h
@@ -41,7 +41,6 @@
 
 #include <linux/nfs_fs.h>
 #include <linux/nfs_page.h>
-#include <linux/exp_xdr.h>
 #include <scsi/osd_protocol.h>
 
 #define PNFS_OSD_OSDNAME_MAXSIZE 256
@@ -333,11 +332,6 @@ extern bool pnfs_osd_xdr_decode_layout_comp(struct pnfs_osd_object_cred *comp,
 	struct pnfs_osd_xdr_decode_layout_iter *iter, struct xdr_stream *xdr,
 	int *err);
 
-/* Layout encoding */
-extern int pnfs_osd_xdr_encode_layout(
-	struct exp_xdr_stream *xdr,
-	struct pnfs_osd_layout *layout);
-
 /* Device Info helpers */
 
 /* Note: All strings inside @deviceaddr point to space inside @p.
@@ -346,10 +340,6 @@ extern int pnfs_osd_xdr_encode_layout(
 extern void pnfs_osd_xdr_decode_deviceaddr(
 	struct pnfs_osd_deviceaddr *deviceaddr, __be32 *p);
 
-/* For Servers */
-extern int pnfs_osd_xdr_encode_deviceaddr(
-	struct exp_xdr_stream *xdr, struct pnfs_osd_deviceaddr *devaddr);
-
 /* layoutupdate (layout_commit) xdr helpers */
 extern int
 pnfs_osd_xdr_encode_layoutupdate(struct xdr_stream *xdr,
-- 
1.7.2.3

[PATCH 09/13] SQUASHME: pnfs_osd_xdr: Avoid using xdr_rewind_stream

[Boaz Harrosh]

In encoding the io_err xdr. We used the xdr_rewind_stream
API when we run out of space and want to encode the last
accumulated error.

We can avoid that by splitting the encoding API into two.
first call: to reserved space.
Second: do the encoding.
If reserving space fails we can now use the previous
place for the accumulated error

Signed-off-by: Boaz Harrosh <bharrosh@panasas.com>
---
 fs/nfs/objlayout/objlayout.c        |   25 ++++++++++++--------
 fs/nfs/objlayout/pnfs_osd_xdr_cli.c |   41 ++++++++++++++--------------------
 include/linux/pnfs_osd_xdr.h        |   14 +++--------
 3 files changed, 36 insertions(+), 44 deletions(-)

diff --git a/fs/nfs/objlayout/objlayout.c b/fs/nfs/objlayout/objlayout.c
index e3c42fc..8892130 100644
--- a/fs/nfs/objlayout/objlayout.c
+++ b/fs/nfs/objlayout/objlayout.c
@@ -532,7 +532,7 @@ merge_ioerr(struct pnfs_osd_ioerr *dest_err,
 }
 
 static void
-encode_accumulated_error(struct objlayout *objlay, struct xdr_stream *xdr)
+encode_accumulated_error(struct objlayout *objlay, __be32 *p)
 {
 	struct objlayout_io_state *state, *tmp;
 	struct pnfs_osd_ioerr accumulated_err = {.oer_errno = 0};
@@ -564,7 +564,7 @@ encode_accumulated_error(struct objlayout *objlay, struct xdr_stream *xdr)
 		objlayout_free_io_state(state);
 	}
 
-	BUG_ON(pnfs_osd_xdr_encode_ioerr(xdr, &accumulated_err));
+	pnfs_osd_xdr_encode_ioerr(p, &accumulated_err);
 }
 
 void
@@ -574,7 +574,7 @@ objlayout_encode_layoutreturn(struct pnfs_layout_hdr *pnfslay,
 {
 	struct objlayout *objlay = OBJLAYOUT(pnfslay);
 	struct objlayout_io_state *state, *tmp;
-	__be32 *start, *uninitialized_var(last_xdr);
+	__be32 *start;
 
 	dprintk("%s: Begin\n", __func__);
 	start = xdr_reserve_space(xdr, 4);
@@ -583,10 +583,11 @@ objlayout_encode_layoutreturn(struct pnfs_layout_hdr *pnfslay,
 	spin_lock(&objlay->lock);
 
 	list_for_each_entry_safe(state, tmp, &objlay->err_list, err_list) {
+		__be32 *last_xdr = NULL, *p;
 		unsigned i;
 		int res = 0;
 
-		for (i = 0; i < state->num_comps && !res; i++) {
+		for (i = 0; i < state->num_comps; i++) {
 			struct pnfs_osd_ioerr *ioerr = &state->ioerrs[i];
 
 			if (!ioerr->oer_errno)
@@ -604,23 +605,27 @@ objlayout_encode_layoutreturn(struct pnfs_layout_hdr *pnfslay,
 				ioerr->oer_comp_offset,
 				ioerr->oer_comp_length);
 
-			last_xdr = xdr->p;
-			res = pnfs_osd_xdr_encode_ioerr(xdr, &state->ioerrs[i]);
+			p = pnfs_osd_xdr_ioerr_reserve_space(xdr);
+			if (unlikely(!p)) {
+				res = -E2BIG;
+				break; /* accumulated_error */
+			}
+
+			last_xdr = p;
+			pnfs_osd_xdr_encode_ioerr(p, &state->ioerrs[i]);
 		}
 
 		/* TODO: use xdr_write_pages */
 		if (unlikely(res)) {
 			/* no space for even one error descriptor */
-			BUG_ON(last_xdr == start + 1);
+			BUG_ON(!last_xdr);
 
 			/* we've encountered a situation with lots and lots of
 			 * errors and no space to encode them all. Use the last
 			 * available slot to report the union of all the
 			 * remaining errors.
 			 */
-			xdr_rewind_stream(xdr, last_xdr -
-					       pnfs_osd_ioerr_xdr_sz() / 4);
-			encode_accumulated_error(objlay, xdr);
+			encode_accumulated_error(objlay, last_xdr);
 			goto loop_done;
 		}
 		list_del(&state->err_list);
diff --git a/fs/nfs/objlayout/pnfs_osd_xdr_cli.c b/fs/nfs/objlayout/pnfs_osd_xdr_cli.c
index b55131c..2aba671 100644
--- a/fs/nfs/objlayout/pnfs_osd_xdr_cli.c
+++ b/fs/nfs/objlayout/pnfs_osd_xdr_cli.c
@@ -369,22 +369,17 @@ pnfs_osd_xdr_encode_layoutupdate(struct xdr_stream *xdr,
  * 	struct pnfs_deviceid	oid_device_id;
  * 	u64			oid_partition_id;
  * 	u64			oid_object_id;
+ * }; // xdr size 32 bytes
  */
-static inline int pnfs_osd_xdr_encode_objid(struct xdr_stream *xdr,
-					    struct pnfs_osd_objid *object_id)
+static inline __be32 *
+pnfs_osd_xdr_encode_objid(__be32 *p, struct pnfs_osd_objid *object_id)
 {
-	__be32 *p;
-
-	p = xdr_reserve_space(xdr, 32);
-	if (!p)
-		return -E2BIG;
-
 	p = xdr_encode_opaque_fixed(p, &object_id->oid_device_id.data,
 				    sizeof(object_id->oid_device_id.data));
 	p = xdr_encode_hyper(p, object_id->oid_partition_id);
 	p = xdr_encode_hyper(p, object_id->oid_object_id);
 
-	return 0;
+	return p;
 }
 
 /*
@@ -394,26 +389,24 @@ static inline int pnfs_osd_xdr_encode_objid(struct xdr_stream *xdr,
  * 	u64			oer_comp_length;
  * 	u32			oer_iswrite;
  * 	u32			oer_errno;
- * };
+ * }; // xdr size 32 + 24 bytes
  */
-int pnfs_osd_xdr_encode_ioerr(struct xdr_stream *xdr,
-			      struct pnfs_osd_ioerr *ioerr)
+void pnfs_osd_xdr_encode_ioerr(__be32 *p, struct pnfs_osd_ioerr *ioerr)
 {
-	__be32 *p;
-	int ret;
-
-	ret = pnfs_osd_xdr_encode_objid(xdr, &ioerr->oer_component);
-	if (ret)
-		return ret;
-
-	p = xdr_reserve_space(xdr, 24);
-	if (!p)
-		return -E2BIG;
-
+	p = pnfs_osd_xdr_encode_objid(p, &ioerr->oer_component);
 	p = xdr_encode_hyper(p, ioerr->oer_comp_offset);
 	p = xdr_encode_hyper(p, ioerr->oer_comp_length);
 	*p++ = cpu_to_be32(ioerr->oer_iswrite);
 	*p   = cpu_to_be32(ioerr->oer_errno);
+}
 
-	return 0;
+__be32 *pnfs_osd_xdr_ioerr_reserve_space(struct xdr_stream *xdr)
+{
+	__be32 * p;
+	
+	p = xdr_reserve_space(xdr, 32 + 24);
+	if (unlikely(!p))
+		dprintk("%s: out of xdr space\n", __func__);
+
+	return p;
 }
diff --git a/include/linux/pnfs_osd_xdr.h b/include/linux/pnfs_osd_xdr.h
index db22471..747d06d 100644
--- a/include/linux/pnfs_osd_xdr.h
+++ b/include/linux/pnfs_osd_xdr.h
@@ -279,14 +279,6 @@ struct pnfs_osd_ioerr {
 	u32			oer_errno;
 };
 
-/* FIXME: remove */
-static inline unsigned
-pnfs_osd_ioerr_xdr_sz(void)
-{
-	return pnfs_osd_objid_xdr_sz() + 2 + 2 + 1 + 1;
-}
-
-
 /* OSD XDR API */
 /* Layout helpers */
 /* Layout decoding is done in two parts:
@@ -348,8 +340,10 @@ extern __be32 *
 pnfs_osd_xdr_decode_layoutupdate(struct pnfs_osd_layoutupdate *lou, __be32 *p);
 
 /* osd_ioerror encoding/decoding (layout_return) */
-extern int
-pnfs_osd_xdr_encode_ioerr(struct xdr_stream *xdr, struct pnfs_osd_ioerr *ioerr);
+/* Client */
+extern __be32 *pnfs_osd_xdr_ioerr_reserve_space(struct xdr_stream *xdr);
+extern void pnfs_osd_xdr_encode_ioerr(__be32 *p, struct pnfs_osd_ioerr *ioerr);
+/* Server*/
 extern __be32 *
 pnfs_osd_xdr_decode_ioerr(struct pnfs_osd_ioerr *ioerr, __be32 *p);
 
-- 
1.7.2.3

[PATCH 10/13] [RFC] Bugs in new pnfs write path

[Boaz Harrosh]

1. In nfs4_write_done_cb: data->write_done_cb comes with a NULL.
   Just as a guess I call nfs4_write_done_cb() just above it
   it looked like the right thing todo. With that in, I'm able
   to write things to file When converting pnfs.c:258 to a WARN_ON.

   Benny we might want to set data->write_done_cb somewhere in the
   none-rpc path? where is it best to do that?

2. In pnfs_ld_write_done:
	put_lseg(data->lseg);
	data->lseg = NULL;
   was done before the call to pnfs_set_layoutcommit()
   which trys to get_lseg() on that same data->lseg.

3. In pnfs_ld_write_done:
   data->mds_ops->rpc_call_done(NULL, data);
   crashes with a NULL task. Just pass it with &data->task

   Which calls for a cleanup. There is bunch of functions
   with [task, write_data] API. And the task is always
   write_data->task

Signed-off-by: Boaz Harrosh <bharrosh@panasas.com>
---
 fs/nfs/nfs4proc.c |    3 ++-
 fs/nfs/pnfs.c     |   10 ++++++----
 2 files changed, 8 insertions(+), 5 deletions(-)

diff --git a/fs/nfs/nfs4proc.c b/fs/nfs/nfs4proc.c
index 759523a..1a53187 100644
--- a/fs/nfs/nfs4proc.c
+++ b/fs/nfs/nfs4proc.c
@@ -3250,7 +3250,8 @@ static int nfs4_write_done(struct rpc_task *task, struct nfs_write_data *data)
 {
 	if (!nfs4_sequence_done(task, &data->res.seq_res))
 		return -EAGAIN;
-	return data->write_done_cb(task, data);
+	return data->write_done_cb ? data->write_done_cb(task, data) :
+		nfs4_write_done_cb(task, data);
 }
 
 /* Reset the the nfs_write_data to send the write to the MDS. */
diff --git a/fs/nfs/pnfs.c b/fs/nfs/pnfs.c
index 17d0c4c..b04cdb4 100644
--- a/fs/nfs/pnfs.c
+++ b/fs/nfs/pnfs.c
@@ -255,7 +255,7 @@ put_lseg_common(struct pnfs_layout_segment *lseg)
 {
 	struct inode *inode = lseg->pls_layout->plh_inode;
 
-	BUG_ON(test_bit(NFS_LSEG_VALID, &lseg->pls_flags));
+ 	WARN_ON(test_bit(NFS_LSEG_VALID, &lseg->pls_flags));
 	list_del_init(&lseg->pls_list);
 	if (list_empty(&lseg->pls_layout->plh_segs)) {
 		set_bit(NFS_LAYOUT_DESTROYED, &lseg->pls_layout->plh_flags);
@@ -1124,15 +1124,17 @@ pnfs_ld_write_done(struct nfs_write_data *data)
 {
 	int status;
 
-	put_lseg(data->lseg);
-	data->lseg = NULL;
 	if (!data->pnfs_error) {
 		pnfs_set_layoutcommit(data);
-		data->mds_ops->rpc_call_done(NULL, data);
+		data->mds_ops->rpc_call_done(&data->task, data);
 		data->mds_ops->rpc_release(data);
+		put_lseg(data->lseg);
+		data->lseg = NULL;
 		return 0;
 	}
 
+	put_lseg(data->lseg);
+	data->lseg = NULL;
 	dprintk("%s: pnfs_error=%d, retry via MDS\n", __func__,
 		data->pnfs_error);
 	status = nfs_initiate_write(data, NFS_CLIENT(data->inode), data->mds_ops, NFS_FILE_SYNC);
-- 
1.7.2.3

[PATCH 11/13] SQUASHME: pnfs_osd_xdr: Add Server API for encoding/decoding osd XDRs

[Boaz Harrosh]

This patch adds declarations and implementation fixes
to the objects Server XDR implementation. Needs squashing to
the patch That intruduces the pnfs_osd_xdr_srv.c file

Signed-off-by: Boaz Harrosh <bharrosh@panasas.com>
---
 fs/exportfs/pnfs_osd_xdr_srv.c |   23 +++++++++++++----------
 include/linux/pnfs_osd_xdr.h   |   14 ++++++++++++--
 2 files changed, 25 insertions(+), 12 deletions(-)

diff --git a/fs/exportfs/pnfs_osd_xdr_srv.c b/fs/exportfs/pnfs_osd_xdr_srv.c
index 60df0df..cf2df46 100644
--- a/fs/exportfs/pnfs_osd_xdr_srv.c
+++ b/fs/exportfs/pnfs_osd_xdr_srv.c
@@ -1,12 +1,11 @@
 /*
- *  pnfs_osd_xdr_enc.c
+ *  Object-Based pNFS Layout XDR layer for the Server side
  *
- *  Object-Based pNFS Layout XDR layer
- *
- *  Copyright (C) 2007-2009 Panasas Inc.
+ *  Copyright (C) 2007 and on Panasas Inc.
  *  All rights reserved.
  *
  *  Benny Halevy <bhalevy@panasas.com>
+ *  Boaz Harrosh <bharrosh@panasas.com>
  *
  *  This program is free software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License version 2
@@ -253,7 +252,7 @@ EXPORT_SYMBOL(pnfs_osd_xdr_decode_layoutupdate);
  * 	struct pnfs_deviceid	oid_device_id;
  * 	u64			oid_partition_id;
  * 	u64			oid_object_id;
- * };
+ * }; xdr size 32
  */
 static inline __be32 *
 pnfs_osd_xdr_decode_objid(__be32 *p, struct pnfs_osd_objid *objid)
@@ -274,16 +273,20 @@ pnfs_osd_xdr_decode_objid(__be32 *p, struct pnfs_osd_objid *objid)
  * 	u64			oer_comp_length;
  * 	u32			oer_iswrite;
  * 	u32			oer_errno;
- * };
+ * }; xdr size 32 + 24
  */
-__be32 *
-pnfs_osd_xdr_decode_ioerr(struct pnfs_osd_ioerr *ioerr, __be32 *p)
+bool pnfs_osd_xdr_decode_ioerr(struct pnfs_osd_ioerr *ioerr,
+			       struct exp_xdr_stream *xdr)
 {
+	__be32 *p = exp_xdr_reserve_space(xdr, 32 + 24);
+	if (!p)
+		return false;
+
 	p = pnfs_osd_xdr_decode_objid(p, &ioerr->oer_component);
 	p = xdr_decode_hyper(p, &ioerr->oer_comp_offset);
 	p = xdr_decode_hyper(p, &ioerr->oer_comp_length);
 	ioerr->oer_iswrite = be32_to_cpu(*p++);
-	ioerr->oer_errno = be32_to_cpu(*p++);
-	return p;
+	ioerr->oer_errno = be32_to_cpu(*p);
+	return true;
 }
 EXPORT_SYMBOL(pnfs_osd_xdr_decode_ioerr);
diff --git a/include/linux/pnfs_osd_xdr.h b/include/linux/pnfs_osd_xdr.h
index 747d06d..9581d2b 100644
--- a/include/linux/pnfs_osd_xdr.h
+++ b/include/linux/pnfs_osd_xdr.h
@@ -41,6 +41,7 @@
 
 #include <linux/nfs_fs.h>
 #include <linux/nfs_page.h>
+#include <linux/exp_xdr.h>
 #include <scsi/osd_protocol.h>
 
 #define PNFS_OSD_OSDNAME_MAXSIZE 256
@@ -324,6 +325,11 @@ extern bool pnfs_osd_xdr_decode_layout_comp(struct pnfs_osd_object_cred *comp,
 	struct pnfs_osd_xdr_decode_layout_iter *iter, struct xdr_stream *xdr,
 	int *err);
 
+/* Layout encoding */
+extern int pnfs_osd_xdr_encode_layout(
+	struct exp_xdr_stream *xdr,
+	struct pnfs_osd_layout *layout);
+
 /* Device Info helpers */
 
 /* Note: All strings inside @deviceaddr point to space inside @p.
@@ -332,6 +338,10 @@ extern bool pnfs_osd_xdr_decode_layout_comp(struct pnfs_osd_object_cred *comp,
 extern void pnfs_osd_xdr_decode_deviceaddr(
 	struct pnfs_osd_deviceaddr *deviceaddr, __be32 *p);
 
+/* For Servers */
+extern int pnfs_osd_xdr_encode_deviceaddr(
+	struct exp_xdr_stream *xdr, struct pnfs_osd_deviceaddr *devaddr);
+
 /* layoutupdate (layout_commit) xdr helpers */
 extern int
 pnfs_osd_xdr_encode_layoutupdate(struct xdr_stream *xdr,
@@ -344,7 +354,7 @@ pnfs_osd_xdr_decode_layoutupdate(struct pnfs_osd_layoutupdate *lou, __be32 *p);
 extern __be32 *pnfs_osd_xdr_ioerr_reserve_space(struct xdr_stream *xdr);
 extern void pnfs_osd_xdr_encode_ioerr(__be32 *p, struct pnfs_osd_ioerr *ioerr);
 /* Server*/
-extern __be32 *
-pnfs_osd_xdr_decode_ioerr(struct pnfs_osd_ioerr *ioerr, __be32 *p);
+extern bool pnfs_osd_xdr_decode_ioerr(struct pnfs_osd_ioerr *ioerr,
+				      struct exp_xdr_stream *xdr);
 
 #endif /* __PNFS_OSD_XDR_H__ */
-- 
1.7.2.3

[PATCH 12/13] SQUASHME: XDR API changes to pnfs_osd_xdr_decode_ioerr()

[Boaz Harrosh]

Now accepts an exp_xdr_stream. And returns a bool when
done. Easier on the coding.

Signed-off-by: Boaz Harrosh <bharrosh@panasas.com>
---
 fs/exofs/export.c |   18 ++++++++----------
 1 files changed, 8 insertions(+), 10 deletions(-)

diff --git a/fs/exofs/export.c b/fs/exofs/export.c
index 69bce46..31e16b4 100644
--- a/fs/exofs/export.c
+++ b/fs/exofs/export.c
@@ -246,19 +246,17 @@ static int exofs_layout_return(
 	struct inode *inode,
 	const struct nfsd4_pnfs_layoutreturn_arg *args)
 {
-	__be32 *p = args->lrf_body;
-	unsigned len = exp_xdr_qwords(args->lrf_body_len);
+	struct exp_xdr_stream xdr = {
+		.p = args->lrf_body,
+		.end = args->lrf_body + exp_xdr_qwords(args->lrf_body_len),
+	};
+	struct pnfs_osd_ioerr ioerr;
 
-	EXOFS_DBGMSG("(0x%lx) cookie %p xdr_len %d\n",
-		     inode->i_ino, args->lr_cookie, len);
+	EXOFS_DBGMSG("(0x%lx) cookie %p body_len %d\n",
+		     inode->i_ino, args->lr_cookie, args->lrf_body_len);
 
-	while (len >= pnfs_osd_ioerr_xdr_sz()) {
-		struct pnfs_osd_ioerr ioerr;
-
-		p = pnfs_osd_xdr_decode_ioerr(&ioerr, p);
-		len -= pnfs_osd_ioerr_xdr_sz();
+	while (pnfs_osd_xdr_decode_ioerr(&ioerr, &xdr))
 		exofs_handle_error(&ioerr);
-	}
 
 	if (args->lr_cookie) {
 		struct exofs_i_info *oi = exofs_i(inode);
-- 
1.7.2.3

[PATCH 13/13] SQUASHME: dbg Print the full device_id returned

[Boaz Harrosh]

Signed-off-by: Boaz Harrosh <bharrosh@panasas.com>
---
 fs/exofs/export.c |    5 +++--
 1 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/fs/exofs/export.c b/fs/exofs/export.c
index 31e16b4..99768ab 100644
--- a/fs/exofs/export.c
+++ b/fs/exofs/export.c
@@ -312,8 +312,9 @@ int exofs_get_device_info(struct super_block *sb, struct exp_xdr_stream *xdr,
 
 	exp_xdr_encode_opaque_len(start, xdr->p);
 
-	EXOFS_DBGMSG("xdr_bytes=%Zu devno=%lld osdname-%s\n",
-		     exp_xdr_qbytes(xdr->p - start), devno, odi->osdname);
+	EXOFS_DBGMSG("xdr_bytes=%Zu devid=(%llx,%llx) osdname-%s\n",
+		     exp_xdr_qbytes(xdr->p - start), devid->sbid, devno,
+		     odi->osdname);
 	return 0;
 
 err:
-- 
1.7.2.3

pnfs-obj: git diff pnfs-all-2.6.38 to pnfs-submit. Without the reordering

[Boaz Harrosh]

Just for review. These are the actual code changes made by Benny
since the last tests at Connectathon

Thanks
Boaz
---
git diff --stat -p -M pnfs/pnfs-all-2.6.38 REORDER_REVERT -- fs/nfs/objlayout/ include/linux/pnfs_osd_xdr.h
 fs/nfs/objlayout/Kbuild             |    2 +-
 fs/nfs/objlayout/objio_osd.c        |   39 +----------
 fs/nfs/objlayout/objlayout.c        |  121 ++++++++++++++++-------------------
 fs/nfs/objlayout/objlayout.h        |   19 ++----
 fs/nfs/objlayout/panfs_shim.c       |   18 +-----
 fs/nfs/objlayout/pnfs_osd_xdr_cli.c |   55 +++++++---------
 include/linux/pnfs_osd_xdr.h        |   26 ++++----
 7 files changed, 102 insertions(+), 178 deletions(-)

diff --git a/fs/nfs/objlayout/Kbuild b/fs/nfs/objlayout/Kbuild
index 9addfe8..6cb585f 100644
--- a/fs/nfs/objlayout/Kbuild
+++ b/fs/nfs/objlayout/Kbuild
@@ -1,7 +1,7 @@
 #
 # Makefile for the pNFS Objects Layout Driver kernel module
 #
-objlayoutdriver-y := pnfs_osd_xdr_cli.o objlayout.o objio_osd.o
+objlayoutdriver-y := objio_osd.o pnfs_osd_xdr_cli.o objlayout.o
 obj-$(CONFIG_PNFS_OBJLAYOUT) += objlayoutdriver.o
 
 #
diff --git a/fs/nfs/objlayout/objio_osd.c b/fs/nfs/objlayout/objio_osd.c
index ea19404..a2cb9a3 100644
--- a/fs/nfs/objlayout/objio_osd.c
+++ b/fs/nfs/objlayout/objio_osd.c
@@ -1,12 +1,10 @@
 /*
- *  objio_osd.c
- *
  *  pNFS Objects layout implementation over open-osd initiator library
  *
- *  Copyright (C) 2009 Panasas Inc.
+ *  Copyright (C) 2009 Panasas Inc. [year of first publication]
  *  All rights reserved.
  *
- *  Benny Halevy <bharrosh@panasas.com>
+ *  Benny Halevy <bhalevy@panasas.com>
  *  Boaz Harrosh <bharrosh@panasas.com>
  *
  *  This program is free software; you can redistribute it and/or modify
@@ -40,11 +38,7 @@
  */
 
 #include <linux/module.h>
-#include <scsi/scsi_device.h>
-#include <scsi/osd_attributes.h>
 #include <scsi/osd_initiator.h>
-#include <scsi/osd_sec.h>
-#include <scsi/osd_sense.h>
 
 #include "objlayout.h"
 
@@ -855,7 +849,7 @@ static ssize_t _write_done(struct objio_state *ios)
 	if (likely(!ret)) {
 		/* FIXME: should be based on the OSD's persistence model
 		 * See OSD2r05 Section 4.13 Data persistence model */
-		ios->ol_state.committed = NFS_UNSTABLE; //NFS_FILE_SYNC;
+		ios->ol_state.committed = NFS_FILE_SYNC;
 		status = ios->length;
 	} else {
 		status = ret;
@@ -909,7 +903,6 @@ static int _write_mirrors(struct objio_state *ios, unsigned cur_comp)
 			per_dev->offset =  master_dev->offset;
 		} else {
 			bio = master_dev->bio;
-			/* FIXME: bio_set_dir() */
 			bio->bi_rw |= REQ_WRITE;
 		}
 
@@ -966,34 +959,13 @@ ssize_t objio_write_pagelist(struct objlayout_io_state *ol_state, bool stable)
 	return _write_exec(ios);
 }
 
-/*
- * Policy Operations
- */
-
-/*
- * Get the max [rw]size
- */
-static ssize_t
-objlayout_get_blocksize(void)
-{
-	ssize_t sz = BIO_MAX_PAGES_KMALLOC * PAGE_SIZE;
-
-	return sz;
-}
-
-/*
- * Don't gather across stripes, but rather gather (coalesce) up to
- * the stripe size.
- *
- * FIXME: change interface to use merge_align, merge_count
- */
 static struct pnfs_layoutdriver_type objlayout_type = {
 	.id = LAYOUT_OSD2_OBJECTS,
 	.name = "LAYOUT_OSD2_OBJECTS",
 	.flags                   = PNFS_LAYOUTRET_ON_SETATTR,
 
 	.set_layoutdriver        = objlayout_set_layoutdriver,
-	.clear_layoutdriver      = objlayout_clear_layoutdriver,
+	.unset_layoutdriver      = objlayout_unset_layoutdriver,
 
 	.alloc_layout_hdr        = objlayout_alloc_layout_hdr,
 	.free_layout_hdr         = objlayout_free_layout_hdr,
@@ -1001,11 +973,8 @@ static struct pnfs_layoutdriver_type objlayout_type = {
 	.alloc_lseg              = objlayout_alloc_lseg,
 	.free_lseg               = objlayout_free_lseg,
 
-	.get_blocksize           = objlayout_get_blocksize,
-
 	.read_pagelist           = objlayout_read_pagelist,
 	.write_pagelist          = objlayout_write_pagelist,
-	.commit                  = objlayout_commit,
 
 	.encode_layoutcommit	 = objlayout_encode_layoutcommit,
 	.encode_layoutreturn     = objlayout_encode_layoutreturn,
diff --git a/fs/nfs/objlayout/objlayout.c b/fs/nfs/objlayout/objlayout.c
index 14a1e97..a04e232 100644
--- a/fs/nfs/objlayout/objlayout.c
+++ b/fs/nfs/objlayout/objlayout.c
@@ -1,9 +1,7 @@
 /*
- *  objlayout.c
+ *  pNFS Objects layout driver high level definitions
  *
- *  pNFS layout driver for Panasas OSDs
- *
- *  Copyright (C) 2007-2009 Panasas Inc.
+ *  Copyright (C) 2007 Panasas Inc. [year of first publication]
  *  All rights reserved.
  *
  *  Benny Halevy <bhalevy@panasas.com>
@@ -84,23 +82,39 @@ struct pnfs_layout_segment *
 objlayout_alloc_lseg(struct pnfs_layout_hdr *pnfslay,
 		     struct nfs4_layoutget_res *lgr)
 {
-	int status;
-	void *layout = lgr->layout.buf;
-	struct objlayout_segment *objlseg;
+	int status = -ENOMEM;
+	struct xdr_stream stream;
+	struct xdr_buf buf = {
+		.pages =  lgr->layoutp->pages,
+		.page_len =  lgr->layoutp->len,
+		.buflen =  lgr->layoutp->len,
+		.len = lgr->layoutp->len,
+	};
+	struct page *scratch;
+	__be32 *p;
+	struct objlayout_segment *objlseg = NULL;
 	struct pnfs_osd_layout *pnfs_osd_layout;
 
-	dprintk("%s: Begin pnfslay %p layout %p\n", __func__, pnfslay, layout);
+	dprintk("%s: Begin pnfslay %p\n", __func__, pnfslay);
+
+	scratch = alloc_page(GFP_KERNEL);
+	if (!scratch)
+		goto err_nofree;
+
+	xdr_init_decode(&stream, &buf, NULL);
+	xdr_set_scratch_buffer(&stream, page_address(scratch), PAGE_SIZE);
 
-	BUG_ON(!layout);
+	p = xdr_inline_decode(&stream, pnfs_osd_data_map_xdr_sz() << 2);
+	if (unlikely(!p))
+		goto err;
 
-	status = -ENOMEM;
 	objlseg = kzalloc(sizeof(*objlseg) +
-			  pnfs_osd_layout_incore_sz(layout), GFP_KERNEL);
+			  pnfs_osd_layout_incore_sz(p), GFP_KERNEL);
 	if (!objlseg)
 		goto err;
 
 	pnfs_osd_layout = (struct pnfs_osd_layout *)objlseg->pnfs_osd_layout;
-	pnfs_osd_xdr_decode_layout(pnfs_osd_layout, layout);
+	pnfs_osd_xdr_decode_layout(pnfs_osd_layout, p);
 
 	objlseg->lseg.pls_range = lgr->range;
 	status = objio_alloc_lseg(&objlseg->internal, pnfslay, &objlseg->lseg,
@@ -108,11 +122,15 @@ objlayout_alloc_lseg(struct pnfs_layout_hdr *pnfslay,
 	if (status)
 		goto err;
 
+	__free_page(scratch);
+
 	dprintk("%s: Return %p\n", __func__, &objlseg->lseg);
 	return &objlseg->lseg;
 
- err:
+err:
 	kfree(objlseg);
+	__free_page(scratch);
+err_nofree:
 	return ERR_PTR(status);
 }
 
@@ -161,7 +179,6 @@ static struct objlayout_io_state *
 objlayout_alloc_io_state(struct pnfs_layout_hdr *pnfs_layout_type,
 			struct page **pages,
 			unsigned pgbase,
-			unsigned nr_pages,
 			loff_t offset,
 			size_t count,
 			struct pnfs_layout_segment *lseg,
@@ -171,7 +188,6 @@ objlayout_alloc_io_state(struct pnfs_layout_hdr *pnfs_layout_type,
 		container_of(lseg, struct objlayout_segment, lseg);
 	struct objlayout_io_state *state;
 	u64 lseg_end_offset;
-	size_t size_nr_pages;
 
 	dprintk("%s: allocating io_state\n", __func__);
 	if (objio_alloc_io_state(objlseg->internal, &state))
@@ -186,24 +202,16 @@ objlayout_alloc_io_state(struct pnfs_layout_hdr *pnfs_layout_type,
 	}
 
 	if (pgbase > PAGE_SIZE) {
-		unsigned n = pgbase >> PAGE_SHIFT;
-
+		pages += pgbase >> PAGE_SHIFT;
 		pgbase &= ~PAGE_MASK;
-		pages += n;
-		nr_pages -= n;
 	}
 
-	size_nr_pages = (pgbase + count + PAGE_SIZE - 1) >> PAGE_SHIFT;
-	BUG_ON(nr_pages < size_nr_pages);
-	if (nr_pages > size_nr_pages)
-		nr_pages = size_nr_pages;
-
 	INIT_LIST_HEAD(&state->err_list);
 	state->objlseg = objlseg;
 	state->rpcdata = rpcdata;
 	state->pages = pages;
 	state->pgbase = pgbase;
-	state->nr_pages = nr_pages;
+	state->nr_pages = (pgbase + count + PAGE_SIZE - 1) >> PAGE_SHIFT;
 	state->offset = offset;
 	state->count = count;
 	state->sync = 0;
@@ -280,32 +288,6 @@ objlayout_io_set_result(struct objlayout_io_state *state, unsigned index,
 	}
 }
 
-static void _rpc_commit_complete(struct work_struct *work)
-{
-	struct rpc_task *task;
-	struct nfs_write_data *wdata;
-
-	dprintk("%s enter\n", __func__);
-	task = container_of(work, struct rpc_task, u.tk_work);
-	wdata = container_of(task, struct nfs_write_data, task);
-
-	pnfs_commit_done(wdata);
-}
-
-/*
- * Commit data remotely on OSDs
- */
-enum pnfs_try_status
-objlayout_commit(struct nfs_write_data *wdata, int how)
-{
-	int status = PNFS_ATTEMPTED;
-
-	INIT_WORK(&wdata->task.u.tk_work, _rpc_commit_complete);
-	schedule_work(&wdata->task.u.tk_work);
-	dprintk("%s: Return %d\n", __func__, status);
-	return status;
-}
-
 /* Function scheduled on rpc workqueue to call ->nfs_readlist_complete().
  * This is because the osd completion is called with ints-off from
  * the block layer
@@ -319,7 +301,7 @@ static void _rpc_read_complete(struct work_struct *work)
 	task = container_of(work, struct rpc_task, u.tk_work);
 	rdata = container_of(task, struct nfs_read_data, task);
 
-	pnfs_read_done(rdata);
+	pnfs_ld_read_done(rdata);
 }
 
 void
@@ -340,7 +322,7 @@ objlayout_read_done(struct objlayout_io_state *state, ssize_t status, bool sync)
 	/* must not use state after this point */
 
 	if (sync)
-		pnfs_read_done(rdata);
+		pnfs_ld_read_done(rdata);
 	else {
 		INIT_WORK(&rdata->task.u.tk_work, _rpc_read_complete);
 		schedule_work(&rdata->task.u.tk_work);
@@ -351,7 +333,7 @@ objlayout_read_done(struct objlayout_io_state *state, ssize_t status, bool sync)
  * Perform sync or async reads.
  */
 enum pnfs_try_status
-objlayout_read_pagelist(struct nfs_read_data *rdata, unsigned nr_pages)
+objlayout_read_pagelist(struct nfs_read_data *rdata)
 {
 	loff_t offset = rdata->args.offset;
 	size_t count = rdata->args.count;
@@ -375,8 +357,8 @@ objlayout_read_pagelist(struct nfs_read_data *rdata, unsigned nr_pages)
 
 	state = objlayout_alloc_io_state(NFS_I(rdata->inode)->layout,
 					 rdata->args.pages, rdata->args.pgbase,
-					 nr_pages, offset, count,
-					 rdata->pdata.lseg, rdata);
+					 offset, count,
+					 rdata->lseg, rdata);
 	if (unlikely(!state)) {
 		status = -ENOMEM;
 		goto out;
@@ -387,7 +369,7 @@ objlayout_read_pagelist(struct nfs_read_data *rdata, unsigned nr_pages)
 	status = objio_read_pagelist(state);
  out:
 	dprintk("%s: Return status %Zd\n", __func__, status);
-	rdata->pdata.pnfs_error = status;
+	rdata->pnfs_error = status;
 	return PNFS_ATTEMPTED;
 }
 
@@ -404,7 +386,7 @@ static void _rpc_write_complete(struct work_struct *work)
 	task = container_of(work, struct rpc_task, u.tk_work);
 	wdata = container_of(task, struct nfs_write_data, task);
 
-	pnfs_writeback_done(wdata);
+	pnfs_ld_write_done(wdata);
 }
 
 void
@@ -430,7 +412,7 @@ objlayout_write_done(struct objlayout_io_state *state, ssize_t status,
 	/* must not use state after this point */
 
 	if (sync)
-		pnfs_writeback_done(wdata);
+		pnfs_ld_write_done(wdata);
 	else {
 		INIT_WORK(&wdata->task.u.tk_work, _rpc_write_complete);
 		schedule_work(&wdata->task.u.tk_work);
@@ -442,7 +424,6 @@ objlayout_write_done(struct objlayout_io_state *state, ssize_t status,
  */
 enum pnfs_try_status
 objlayout_write_pagelist(struct nfs_write_data *wdata,
-			 unsigned nr_pages,
 			 int how)
 {
 	struct objlayout_io_state *state;
@@ -454,10 +435,9 @@ objlayout_write_pagelist(struct nfs_write_data *wdata,
 	state = objlayout_alloc_io_state(NFS_I(wdata->inode)->layout,
 					 wdata->args.pages,
 					 wdata->args.pgbase,
-					 nr_pages,
 					 wdata->args.offset,
 					 wdata->args.count,
-					 wdata->pdata.lseg, wdata);
+					 wdata->lseg, wdata);
 	if (unlikely(!state)) {
 		status = -ENOMEM;
 		goto out;
@@ -468,7 +448,7 @@ objlayout_write_pagelist(struct nfs_write_data *wdata,
 	status = objio_write_pagelist(state, how & FLUSH_STABLE);
  out:
 	dprintk("%s: Return status %Zd\n", __func__, status);
-	wdata->pdata.pnfs_error = status;
+	wdata->pnfs_error = status;
 	return PNFS_ATTEMPTED;
 }
 
@@ -648,6 +628,8 @@ objlayout_encode_layoutreturn(struct pnfs_layout_hdr *pnfslay,
 			last_xdr = xdr->p;
 			res = pnfs_osd_xdr_encode_ioerr(xdr, &state->ioerrs[i]);
 		}
+
+		/* TODO: use xdr_write_pages */
 		if (unlikely(res)) {
 			/* no space for even one error descriptor */
 			BUG_ON(last_xdr == start + 1);
@@ -672,6 +654,10 @@ loop_done:
 	dprintk("%s: Return\n", __func__);
 }
 
+
+/*
+ * Get Device Info API for io engines
+ */
 struct objlayout_deviceinfo {
 	struct page *page;
 	struct pnfs_osd_deviceaddr da; /* This must be last */
@@ -687,7 +673,7 @@ int objlayout_get_deviceinfo(struct pnfs_layout_hdr *pnfslay,
 	struct objlayout_deviceinfo *odi;
 	struct pnfs_device pd;
 	struct super_block *sb;
-	struct page *page;
+	struct page *page, **pages;
 	size_t sz;
 	u32 *p;
 	int err;
@@ -696,7 +682,8 @@ int objlayout_get_deviceinfo(struct pnfs_layout_hdr *pnfslay,
 	if (!page)
 		return -ENOMEM;
 
-	pd.area = page_address(page);
+	pages = &page;
+	pd.pages = pages;
 
 	memcpy(&pd.dev_id, d_id, sizeof(*d_id));
 	pd.layout_type = LAYOUT_OSD2_OBJECTS;
@@ -711,7 +698,7 @@ int objlayout_get_deviceinfo(struct pnfs_layout_hdr *pnfslay,
 	if (err)
 		goto err_out;
 
-	p = pd.area;
+	p = page_address(page);
 	sz = pnfs_osd_xdr_deviceaddr_incore_sz(p);
 	odi = kzalloc(sz + (sizeof(*odi) - sizeof(odi->da)), GFP_KERNEL);
 	if (!odi) {
@@ -765,7 +752,7 @@ objlayout_set_layoutdriver(struct nfs_server *server,
  * layoutdriver private data.
  */
 int
-objlayout_clear_layoutdriver(struct nfs_server *server)
+objlayout_unset_layoutdriver(struct nfs_server *server)
 {
 	dprintk("%s: Begin %p\n", __func__, server->pnfs_ld_data);
 	objio_fini_mt(server->pnfs_ld_data);
diff --git a/fs/nfs/objlayout/objlayout.h b/fs/nfs/objlayout/objlayout.h
index 57c29ec..38abb01 100644
--- a/fs/nfs/objlayout/objlayout.h
+++ b/fs/nfs/objlayout/objlayout.h
@@ -1,10 +1,8 @@
 /*
- *  objlayout.h
- *
  *  Data types and function declerations for interfacing with the
  *  pNFS standard object layout driver.
  *
- *  Copyright (C) 2007-2009 Panasas Inc.
+ *  Copyright (C) 2007 Panasas Inc. [year of first publication]
  *  All rights reserved.
  *
  *  Benny Halevy <bhalevy@panasas.com>
@@ -167,10 +165,9 @@ extern void objlayout_put_deviceinfo(struct pnfs_osd_deviceaddr *deviceaddr);
  * exported generic objects function vectors
  */
 
-extern int objlayout_set_layoutdriver(
-	struct nfs_server *,
-	const struct nfs_fh *);
-extern int objlayout_clear_layoutdriver(struct nfs_server *);
+extern int objlayout_set_layoutdriver(struct nfs_server *,
+	const struct nfs_fh *mntfh);
+extern int objlayout_unset_layoutdriver(struct nfs_server *);
 
 extern struct pnfs_layout_hdr *objlayout_alloc_layout_hdr(struct inode *);
 extern void objlayout_free_layout_hdr(struct pnfs_layout_hdr *);
@@ -181,16 +178,10 @@ extern struct pnfs_layout_segment *objlayout_alloc_lseg(
 extern void objlayout_free_lseg(struct pnfs_layout_segment *);
 
 extern enum pnfs_try_status objlayout_read_pagelist(
-	struct nfs_read_data *,
-	unsigned nr_pages);
+	struct nfs_read_data *);
 
 extern enum pnfs_try_status objlayout_write_pagelist(
 	struct nfs_write_data *,
-	unsigned nr_pages,
-	int how);
-
-extern enum pnfs_try_status objlayout_commit(
-	struct nfs_write_data *,
 	int how);
 
 extern void objlayout_encode_layoutcommit(
diff --git a/fs/nfs/objlayout/panfs_shim.c b/fs/nfs/objlayout/panfs_shim.c
index cc18bbf..718ea0c 100644
--- a/fs/nfs/objlayout/panfs_shim.c
+++ b/fs/nfs/objlayout/panfs_shim.c
@@ -627,19 +627,6 @@ EXPORT_SYMBOL(panfs_shim_unregister);
 #define PANLAYOUT_MAX_GATHER_STRIPES 8
 
 /*
- * Get the max [rw]size
- */
-static ssize_t
-panlayout_get_blocksize(void)
-{
-	ssize_t sz = (PANLAYOUT_MAX_STRIPE_WIDTH-1) *
-		      PANLAYOUT_DEF_STRIPE_UNIT *
-		      PANLAYOUT_MAX_GATHER_STRIPES;
-	dprintk("%s: Return %Zd\n", __func__, sz);
-	return sz;
-}
-
-/*
  * Don't gather across stripes, but rather gather (coalesce) up to
  * the stripe size.
  *
@@ -653,7 +640,7 @@ static struct pnfs_layoutdriver_type panlayout_type = {
 	.flags                   = PNFS_LAYOUTRET_ON_SETATTR,
 
 	.set_layoutdriver        = objlayout_set_layoutdriver,
-	.clear_layoutdriver      = objlayout_clear_layoutdriver,
+	.unset_layoutdriver      = objlayout_unset_layoutdriver,
 
 	.alloc_layout_hdr        = objlayout_alloc_layout_hdr,
 	.free_layout_hdr         = objlayout_free_layout_hdr,
@@ -661,11 +648,8 @@ static struct pnfs_layoutdriver_type panlayout_type = {
 	.alloc_lseg              = objlayout_alloc_lseg,
 	.free_lseg               = objlayout_free_lseg,
 
-	.get_blocksize           = panlayout_get_blocksize,
-
 	.read_pagelist           = objlayout_read_pagelist,
 	.write_pagelist          = objlayout_write_pagelist,
-	.commit                  = objlayout_commit,
 
 	.encode_layoutcommit	 = objlayout_encode_layoutcommit,
 	.encode_layoutreturn     = objlayout_encode_layoutreturn,
diff --git a/fs/nfs/objlayout/pnfs_osd_xdr_cli.c b/fs/nfs/objlayout/pnfs_osd_xdr_cli.c
index d05c6be..4dea458 100644
--- a/fs/nfs/objlayout/pnfs_osd_xdr_cli.c
+++ b/fs/nfs/objlayout/pnfs_osd_xdr_cli.c
@@ -1,9 +1,7 @@
 /*
- *  pnfs_osd_xdr.c
- *
  *  Object-Based pNFS Layout XDR layer
  *
- *  Copyright (C) 2007-2009 Panasas Inc.
+ *  Copyright (C) 2007 Panasas Inc. [year of first publication]
  *  All rights reserved.
  *
  *  Benny Halevy <bhalevy@panasas.com>
@@ -43,10 +41,7 @@
 #define NFSDBG_FACILITY         NFSDBG_PNFS_LD
 
 /*
- * The following implementation is based on these Internet Drafts:
- *
- * draft-ietf-nfsv4-minorversion-21
- * draft-ietf-nfsv4-pnfs-obj-12
+ * The following implementation is based on RFC5664
  */
 
 /*
@@ -56,8 +51,8 @@
  * 	u64			oid_object_id;
  * };
  */
-static inline u32 *
-pnfs_osd_xdr_decode_objid(u32 *p, struct pnfs_osd_objid *objid)
+static inline __be32 *
+pnfs_osd_xdr_decode_objid(__be32 *p, struct pnfs_osd_objid *objid)
 {
 	COPYMEM(objid->oid_device_id.data, sizeof(objid->oid_device_id.data));
 	READ64(objid->oid_partition_id);
@@ -65,8 +60,8 @@ pnfs_osd_xdr_decode_objid(u32 *p, struct pnfs_osd_objid *objid)
 	return p;
 }
 
-static inline u32 *
-pnfs_osd_xdr_decode_opaque_cred(u32 *p,
+static inline __be32 *
+pnfs_osd_xdr_decode_opaque_cred(__be32 *p,
 				struct pnfs_osd_opaque_cred *opaque_cred)
 {
 	READ32(opaque_cred->cred_len);
@@ -83,8 +78,8 @@ pnfs_osd_xdr_decode_opaque_cred(u32 *p,
  * 	struct pnfs_osd_opaque_cred	oc_cap;
  * };
  */
-static inline u32 *
-pnfs_osd_xdr_decode_object_cred(u32 *p, struct pnfs_osd_object_cred *comp,
+static inline __be32 *
+pnfs_osd_xdr_decode_object_cred(__be32 *p, struct pnfs_osd_object_cred *comp,
 				u8 **credp)
 {
 	u8 *cred;
@@ -116,7 +111,7 @@ pnfs_osd_xdr_decode_object_cred(u32 *p, struct pnfs_osd_object_cred *comp,
  * };
  */
 static inline u32 *
-pnfs_osd_xdr_decode_data_map(u32 *p, struct pnfs_osd_data_map *data_map)
+pnfs_osd_xdr_decode_data_map(__be32 *p, struct pnfs_osd_data_map *data_map)
 {
 	READ32(data_map->odm_num_comps);
 	READ64(data_map->odm_stripe_unit);
@@ -137,10 +132,10 @@ pnfs_osd_xdr_decode_data_map(u32 *p, struct pnfs_osd_data_map *data_map)
 }
 
 struct pnfs_osd_layout *
-pnfs_osd_xdr_decode_layout(struct pnfs_osd_layout *layout, u32 *p)
+pnfs_osd_xdr_decode_layout(struct pnfs_osd_layout *layout, __be32 *p)
 {
 	int i;
-	u32 *start = p;
+	__be32 *start = p;
 	struct pnfs_osd_object_cred *comp;
 	u8 *cred;
 
@@ -178,8 +173,8 @@ pnfs_osd_xdr_decode_layout(struct pnfs_osd_layout *layout, u32 *p)
  *       should not be freed while the returned information is in use.
  */
 
-u32 *__xdr_read_calc_nfs4_string(
-	u32 *p, struct nfs4_string *str, u8 **freespace)
+__be32 *__xdr_read_calc_nfs4_string(
+	__be32 *p, struct nfs4_string *str, u8 **freespace)
 {
 	u32 len;
 	char *data;
@@ -209,8 +204,8 @@ u32 *__xdr_read_calc_nfs4_string(
 	return p;
 }
 
-u32 *__xdr_read_calc_u8_opaque(
-	u32 *p, struct nfs4_string *str)
+__be32 *__xdr_read_calc_u8_opaque(
+	__be32 *p, struct nfs4_string *str)
 {
 	u32 len;
 
@@ -231,8 +226,8 @@ u32 *__xdr_read_calc_u8_opaque(
  * 	struct nfs4_string	oti_scsi_device_id;
  * };
  */
-u32 *__xdr_read_calc_targetid(
-	u32 *p, struct pnfs_osd_targetid* targetid, u8 **freespace)
+__be32 *__xdr_read_calc_targetid(
+	__be32 *p, struct pnfs_osd_targetid* targetid, u8 **freespace)
 {
 	u32 oti_type;
 
@@ -256,8 +251,8 @@ u32 *__xdr_read_calc_targetid(
  * 	struct nfs4_string	r_addr;
  * };
  */
-u32 *__xdr_read_calc_net_addr(
-	u32 *p, struct pnfs_osd_net_addr* netaddr, u8 **freespace)
+__be32 *__xdr_read_calc_net_addr(
+	__be32 *p, struct pnfs_osd_net_addr* netaddr, u8 **freespace)
 {
 
 	p = __xdr_read_calc_nfs4_string(p,
@@ -277,8 +272,8 @@ u32 *__xdr_read_calc_net_addr(
  * 	struct pnfs_osd_net_addr	ota_netaddr;
  * };
  */
-u32 *__xdr_read_calc_targetaddr(
-	u32 *p, struct pnfs_osd_targetaddr *targetaddr, u8 **freespace)
+__be32 *__xdr_read_calc_targetaddr(
+	__be32 *p, struct pnfs_osd_targetaddr *targetaddr, u8 **freespace)
 {
 	u32 ota_available;
 
@@ -305,8 +300,8 @@ u32 *__xdr_read_calc_targetaddr(
  * 	struct nfs4_string		oda_osdname;
  * };
  */
-u32 *__xdr_read_calc_deviceaddr(
-	u32 *p, struct pnfs_osd_deviceaddr *deviceaddr, u8 **freespace)
+__be32 *__xdr_read_calc_deviceaddr(
+	__be32 *p, struct pnfs_osd_deviceaddr *deviceaddr, u8 **freespace)
 {
 	p = __xdr_read_calc_targetid(p,
 			deviceaddr ? &deviceaddr->oda_targetid : NULL,
@@ -338,7 +333,7 @@ u32 *__xdr_read_calc_deviceaddr(
 	return p;
 }
 
-size_t pnfs_osd_xdr_deviceaddr_incore_sz(u32 *p)
+size_t pnfs_osd_xdr_deviceaddr_incore_sz(__be32 *p)
 {
 	u8 *null_freespace = NULL;
 	size_t sz;
@@ -350,7 +345,7 @@ size_t pnfs_osd_xdr_deviceaddr_incore_sz(u32 *p)
 }
 
 void pnfs_osd_xdr_decode_deviceaddr(
-	struct pnfs_osd_deviceaddr *deviceaddr, u32 *p)
+	struct pnfs_osd_deviceaddr *deviceaddr, __be32 *p)
 {
 	u8 *freespace = (u8 *)(deviceaddr + 1);
 
diff --git a/include/linux/pnfs_osd_xdr.h b/include/linux/pnfs_osd_xdr.h
index b404f33..aed693f 100644
--- a/include/linux/pnfs_osd_xdr.h
+++ b/include/linux/pnfs_osd_xdr.h
@@ -1,9 +1,7 @@
 /*
- *  pnfs_osd_xdr.h
- *
  *  pNFS-osd on-the-wire data structures
  *
- *  Copyright (C) 2007-2009 Panasas Inc.
+ *  Copyright (C) 2007 Panasas Inc. [year of first publication]
  *  All rights reserved.
  *
  *  Benny Halevy <bhalevy@panasas.com>
@@ -154,7 +152,7 @@ struct pnfs_osd_opaque_cred {
 };
 
 static inline int
-pnfs_osd_opaque_cred_xdr_sz(u32 *p)
+pnfs_osd_opaque_cred_xdr_sz(__be32 *p)
 {
 	u32 *start = p;
 	u32 n;
@@ -165,7 +163,7 @@ pnfs_osd_opaque_cred_xdr_sz(u32 *p)
 }
 
 static inline size_t
-pnfs_osd_opaque_cred_incore_sz(u32 *p)
+pnfs_osd_opaque_cred_incore_sz(__be32 *p)
 {
 	u32 n;
 
@@ -195,9 +193,9 @@ struct pnfs_osd_object_cred {
 };
 
 static inline int
-pnfs_osd_object_cred_xdr_sz(u32 *p)
+pnfs_osd_object_cred_xdr_sz(__be32 *p)
 {
-	u32 *start = p;
+	__be32 *start = p;
 
 	p += pnfs_osd_objid_xdr_sz() + 2;
 	p += pnfs_osd_opaque_cred_xdr_sz(p);
@@ -206,7 +204,7 @@ pnfs_osd_object_cred_xdr_sz(u32 *p)
 }
 
 static inline size_t
-pnfs_osd_object_cred_incore_sz(u32 *p)
+pnfs_osd_object_cred_incore_sz(__be32 *p)
 {
 	size_t sz = sizeof(struct pnfs_osd_object_cred);
 
@@ -231,9 +229,9 @@ struct pnfs_osd_layout {
 };
 
 static inline int
-pnfs_osd_layout_xdr_sz(u32 *p)
+pnfs_osd_layout_xdr_sz(__be32 *p)
 {
-	u32 *start = p;
+	__be32 *start = p;
 	u32 n;
 
 	p += pnfs_osd_data_map_xdr_sz() + 1;
@@ -244,7 +242,7 @@ pnfs_osd_layout_xdr_sz(u32 *p)
 }
 
 static inline size_t
-pnfs_osd_layout_incore_sz(u32 *p)
+pnfs_osd_layout_incore_sz(__be32 *p)
 {
 	u32 n;
 	size_t sz;
@@ -399,7 +397,7 @@ pnfs_osd_ioerr_xdr_sz(void)
 
 /* Layout helpers */
 extern struct pnfs_osd_layout *pnfs_osd_xdr_decode_layout(
-	struct pnfs_osd_layout *layout, u32 *p);
+	struct pnfs_osd_layout *layout, __be32 *p);
 
 extern int pnfs_osd_xdr_encode_layout(
 	struct exp_xdr_stream *xdr,
@@ -408,7 +406,7 @@ extern int pnfs_osd_xdr_encode_layout(
 /* Device Info helpers */
 
 /* First pass calculate total size for space needed */
-extern size_t pnfs_osd_xdr_deviceaddr_incore_sz(u32 *p);
+extern size_t pnfs_osd_xdr_deviceaddr_incore_sz(__be32 *p);
 
 /* Note: some strings pointed to inside @deviceaddr might point
  * to space inside @p. @p should stay valid while @deviceaddr
@@ -417,7 +415,7 @@ extern size_t pnfs_osd_xdr_deviceaddr_incore_sz(u32 *p);
  * calculated in first pass by pnfs_osd_xdr_deviceaddr_incore_sz()
  */
 extern void pnfs_osd_xdr_decode_deviceaddr(
-	struct pnfs_osd_deviceaddr *deviceaddr, u32 *p);
+	struct pnfs_osd_deviceaddr *deviceaddr, __be32 *p);
 
 /* For Servers */
 extern int pnfs_osd_xdr_encode_deviceaddr(

Re: [PATCH 10/13] [RFC] Bugs in new pnfs write path

[Benny Halevy]

On 2011-05-21 13:33, Boaz Harrosh wrote:
> 1. In nfs4_write_done_cb: data->write_done_cb comes with a NULL.

what's the call path?

>    Just as a guess I call nfs4_write_done_cb() just above it
>    it looked like the right thing todo. With that in, I'm able
>    to write things to file When converting pnfs.c:258 to a WARN_ON.
> 
>    Benny we might want to set data->write_done_cb somewhere in the
>    none-rpc path? where is it best to do that?

we're not supposed to get there in the non-rpc path...
The non-rpc drivers must call pnfs_ld_write_done.

> 
> 2. In pnfs_ld_write_done:
> 	put_lseg(data->lseg);
> 	data->lseg = NULL;
>    was done before the call to pnfs_set_layoutcommit()
>    which trys to get_lseg() on that same data->lseg.

good catch, thanks!

> 
> 3. In pnfs_ld_write_done:
>    data->mds_ops->rpc_call_done(NULL, data);
>    crashes with a NULL task. Just pass it with &data->task

As we don't go through nfs_initiate_write data->task is not initialized.
Where's the crash exactly?
We better fix it than fake a task structure...

Benny

> 
>    Which calls for a cleanup. There is bunch of functions
>    with [task, write_data] API. And the task is always
>    write_data->task
> 
> Signed-off-by: Boaz Harrosh <bharrosh@panasas.com>
> ---
>  fs/nfs/nfs4proc.c |    3 ++-
>  fs/nfs/pnfs.c     |   10 ++++++----
>  2 files changed, 8 insertions(+), 5 deletions(-)
> 
> diff --git a/fs/nfs/nfs4proc.c b/fs/nfs/nfs4proc.c
> index 759523a..1a53187 100644
> --- a/fs/nfs/nfs4proc.c
> +++ b/fs/nfs/nfs4proc.c
> @@ -3250,7 +3250,8 @@ static int nfs4_write_done(struct rpc_task *task, struct nfs_write_data *data)
>  {
>  	if (!nfs4_sequence_done(task, &data->res.seq_res))
>  		return -EAGAIN;
> -	return data->write_done_cb(task, data);
> +	return data->write_done_cb ? data->write_done_cb(task, data) :
> +		nfs4_write_done_cb(task, data);
>  }
>  
>  /* Reset the the nfs_write_data to send the write to the MDS. */
> diff --git a/fs/nfs/pnfs.c b/fs/nfs/pnfs.c
> index 17d0c4c..b04cdb4 100644
> --- a/fs/nfs/pnfs.c
> +++ b/fs/nfs/pnfs.c
> @@ -255,7 +255,7 @@ put_lseg_common(struct pnfs_layout_segment *lseg)
>  {
>  	struct inode *inode = lseg->pls_layout->plh_inode;
>  
> -	BUG_ON(test_bit(NFS_LSEG_VALID, &lseg->pls_flags));
> + 	WARN_ON(test_bit(NFS_LSEG_VALID, &lseg->pls_flags));
>  	list_del_init(&lseg->pls_list);
>  	if (list_empty(&lseg->pls_layout->plh_segs)) {
>  		set_bit(NFS_LAYOUT_DESTROYED, &lseg->pls_layout->plh_flags);
> @@ -1124,15 +1124,17 @@ pnfs_ld_write_done(struct nfs_write_data *data)
>  {
>  	int status;
>  
> -	put_lseg(data->lseg);
> -	data->lseg = NULL;
>  	if (!data->pnfs_error) {
>  		pnfs_set_layoutcommit(data);
> -		data->mds_ops->rpc_call_done(NULL, data);
> +		data->mds_ops->rpc_call_done(&data->task, data);
>  		data->mds_ops->rpc_release(data);
> +		put_lseg(data->lseg);
> +		data->lseg = NULL;
>  		return 0;
>  	}
>  
> +	put_lseg(data->lseg);
> +	data->lseg = NULL;
>  	dprintk("%s: pnfs_error=%d, retry via MDS\n", __func__,
>  		data->pnfs_error);
>  	status = nfs_initiate_write(data, NFS_CLIENT(data->inode), data->mds_ops, NFS_FILE_SYNC);

Re: [PATCH 10/13] [RFC] Bugs in new pnfs write path

[Boaz Harrosh]

On 05/22/2011 09:08 PM, Benny Halevy wrote:

This is the old patch please see the one that actually works
I sent it by itself after this one

> On 2011-05-21 13:33, Boaz Harrosh wrote:
>> 1. In nfs4_write_done_cb: data->write_done_cb comes with a NULL.
> 
> what's the call path?
> 

I sent it in my mails
it is eventually called from pnfs_ld_write_done trhough
data->mds_ops->rpc_call_done(&data->task, data);

>>    Just as a guess I call nfs4_write_done_cb() just above it
>>    it looked like the right thing todo. With that in, I'm able
>>    to write things to file When converting pnfs.c:258 to a WARN_ON.
>>
>>    Benny we might want to set data->write_done_cb somewhere in the
>>    none-rpc path? where is it best to do that?
> 
> we're not supposed to get there in the non-rpc path...
> The non-rpc drivers must call pnfs_ld_write_done.
> 

Yep! we do. And please don't touch anything, everything works perfectly
now.

As I said it gets called from pnfs_ld_write_done through 
data->mds_ops->rpc_call_done(&data->task, data);

>>
>> 2. In pnfs_ld_write_done:
>> 	put_lseg(data->lseg);
>> 	data->lseg = NULL;
>>    was done before the call to pnfs_set_layoutcommit()
>>    which trys to get_lseg() on that same data->lseg.
> 
> good catch, thanks!
> 
>>
>> 3. In pnfs_ld_write_done:
>>    data->mds_ops->rpc_call_done(NULL, data);
>>    crashes with a NULL task. Just pass it with &data->task
> 
> As we don't go through nfs_initiate_write data->task is not initialized.
> Where's the crash exactly?
> We better fix it than fake a task structure...
> 
> Benny
> 
>>
>>    Which calls for a cleanup. There is bunch of functions
>>    with [task, write_data] API. And the task is always
>>    write_data->task
>>
>> Signed-off-by: Boaz Harrosh <bharrosh@panasas.com>
>> ---
>>  fs/nfs/nfs4proc.c |    3 ++-
>>  fs/nfs/pnfs.c     |   10 ++++++----
>>  2 files changed, 8 insertions(+), 5 deletions(-)
>>
>> diff --git a/fs/nfs/nfs4proc.c b/fs/nfs/nfs4proc.c
>> index 759523a..1a53187 100644
>> --- a/fs/nfs/nfs4proc.c
>> +++ b/fs/nfs/nfs4proc.c
>> @@ -3250,7 +3250,8 @@ static int nfs4_write_done(struct rpc_task *task, struct nfs_write_data *data)
>>  {
>>  	if (!nfs4_sequence_done(task, &data->res.seq_res))
>>  		return -EAGAIN;
>> -	return data->write_done_cb(task, data);
>> +	return data->write_done_cb ? data->write_done_cb(task, data) :
>> +		nfs4_write_done_cb(task, data);
>>  }
>>  
>>  /* Reset the the nfs_write_data to send the write to the MDS. */
>> diff --git a/fs/nfs/pnfs.c b/fs/nfs/pnfs.c
>> index 17d0c4c..b04cdb4 100644
>> --- a/fs/nfs/pnfs.c
>> +++ b/fs/nfs/pnfs.c
>> @@ -255,7 +255,7 @@ put_lseg_common(struct pnfs_layout_segment *lseg)
>>  {
>>  	struct inode *inode = lseg->pls_layout->plh_inode;
>>  
>> -	BUG_ON(test_bit(NFS_LSEG_VALID, &lseg->pls_flags));
>> + 	WARN_ON(test_bit(NFS_LSEG_VALID, &lseg->pls_flags));
>>  	list_del_init(&lseg->pls_list);
>>  	if (list_empty(&lseg->pls_layout->plh_segs)) {
>>  		set_bit(NFS_LAYOUT_DESTROYED, &lseg->pls_layout->plh_flags);
>> @@ -1124,15 +1124,17 @@ pnfs_ld_write_done(struct nfs_write_data *data)
>>  {
>>  	int status;
>>  
>> -	put_lseg(data->lseg);
>> -	data->lseg = NULL;
>>  	if (!data->pnfs_error) {
>>  		pnfs_set_layoutcommit(data);
>> -		data->mds_ops->rpc_call_done(NULL, data);
>> +		data->mds_ops->rpc_call_done(&data->task, data);
>>  		data->mds_ops->rpc_release(data);
>> +		put_lseg(data->lseg);
>> +		data->lseg = NULL;
>>  		return 0;
>>  	}
>>  
>> +	put_lseg(data->lseg);
>> +	data->lseg = NULL;
>>  	dprintk("%s: pnfs_error=%d, retry via MDS\n", __func__,
>>  		data->pnfs_error);
>>  	status = nfs_initiate_write(data, NFS_CLIENT(data->inode), data->mds_ops, NFS_FILE_SYNC);
> 

Re: [PATCH 10/13] [RFC] Bugs in new pnfs write path

[Boaz Harrosh]

On 05/23/2011 07:19 AM, Boaz Harrosh wrote:
>>> 3. In pnfs_ld_write_done:
>>>    data->mds_ops->rpc_call_done(NULL, data);
>>>    crashes with a NULL task. Just pass it with &data->task
>>
>> As we don't go through nfs_initiate_write data->task is not initialized.
>> Where's the crash exactly?
>> We better fix it than fake a task structure...
>>

We better do have a valid data->task because we use it all over
in the objects layout driver, to schedule out of the interrupt.

It works fine I tested it

>> Benny

Boaz