Whamcloud - gitweb
*** empty log message ***
[fs/lustre-release.git] / lustre / lov / lov_pack.c
index a719aac..5f14160 100644 (file)
  *
  */
 
-#define DEBUG_SUBSYSTEM S_LLITE
+#define DEBUG_SUBSYSTEM S_LOV
 #ifndef __KERNEL__
 #include <liblustre.h>
 #endif
 
-#include <linux/lustre_net.h>
-#include <linux/obd.h>
-#include <linux/obd_lov.h>
-#include <linux/obd_class.h>
-#include <linux/obd_support.h>
+#include <lustre_net.h>
+#include <obd.h>
+#include <obd_lov.h>
+#include <obd_class.h>
+#include <obd_support.h>
+#include <lustre/lustre_user.h>
 
 #include "lov_internal.h"
 
-void lov_dump_lmm(int level, struct lov_mds_md *lmm)
+void lov_dump_lmm_v1(int level, struct lov_mds_md_v1 *lmm)
 {
-        struct lov_object_id *loi;
-        int idx;
-
-        CDEBUG(level, "objid "LPX64", magic %#08x, ost_count %u\n",
-               lmm->lmm_object_id, lmm->lmm_magic, lmm->lmm_ost_count);
-        CDEBUG(level,"stripe_size %u, stripe_count %u, stripe_offset %u\n",
-               lmm->lmm_stripe_size, lmm->lmm_stripe_count,
-               lmm->lmm_stripe_offset);
-        for (idx = 0, loi = lmm->lmm_objects; idx < lmm->lmm_ost_count;
-             idx++, loi++)
-                CDEBUG(level, "ost idx %u subobj "LPX64"\n", idx,
-                       loi->l_object_id);
+        struct lov_ost_data_v1 *lod;
+        int i;
+        
+        CDEBUG(level, "objid "LPX64", magic 0x%08x, pattern %#x\n",
+               le64_to_cpu(lmm->lmm_object_id), le32_to_cpu(lmm->lmm_magic),
+               le32_to_cpu(lmm->lmm_pattern));
+        CDEBUG(level,"stripe_size %u, stripe_count %u\n",
+               le32_to_cpu(lmm->lmm_stripe_size),
+               le32_to_cpu(lmm->lmm_stripe_count));
+
+        if (le32_to_cpu(lmm->lmm_stripe_count) <= LOV_V1_INSANE_STRIPE_COUNT) {
+                for (i = 0, lod = lmm->lmm_objects;
+                     i < (int)le32_to_cpu(lmm->lmm_stripe_count); i++, lod++)
+                         CDEBUG(level,
+                                "stripe %u idx %u subobj "LPX64"/"LPX64"\n",
+                                i, le32_to_cpu(lod->l_ost_idx),
+                                le64_to_cpu(lod->l_object_gr),
+                                le64_to_cpu(lod->l_object_id));
+        } else {
+                CDEBUG(level, "bad stripe_count %u > max_stripe_count %u\n",
+                       le32_to_cpu(lmm->lmm_stripe_count),
+                       LOV_V1_INSANE_STRIPE_COUNT);
+        }
+}
+
+void lov_dump_lmm_join(int level, struct lov_mds_md_join *lmmj)
+{
+
+        CDEBUG(level, "objid "LPX64", magic 0x%08X, pattern %#X\n",
+               le64_to_cpu(lmmj->lmmj_md.lmm_object_id),
+               le32_to_cpu(lmmj->lmmj_md.lmm_magic),
+               le32_to_cpu(lmmj->lmmj_md.lmm_pattern));
+        CDEBUG(level,"stripe_size %u, stripe_count %u extent_count %u \n",
+               le32_to_cpu(lmmj->lmmj_md.lmm_stripe_size),
+               le32_to_cpu(lmmj->lmmj_md.lmm_stripe_count),
+               le32_to_cpu(lmmj->lmmj_extent_count));
 }
 
 #define LMM_ASSERT(test)                                                \
@@ -66,44 +91,40 @@ do {                                                                    \
  *     LOVs properly.  For now lov_mds_md_size() just assumes one obd_id
  *     per stripe.
  */
-int lov_packmd(struct lustre_handle *conn, struct lov_mds_md **lmmp,
+int lov_packmd(struct obd_export *exp, struct lov_mds_md **lmmp,
                struct lov_stripe_md *lsm)
 {
-        struct obd_device *obd = class_conn2obd(conn);
+        struct obd_device *obd = class_exp2obd(exp);
         struct lov_obd *lov = &obd->u.lov;
-        struct lov_oinfo *loi;
         struct lov_mds_md *lmm;
-        int ost_count = lov->desc.ld_tgt_count;
-        int stripe_count = ost_count;
+        int stripe_count = lov->desc.ld_tgt_count;
         int lmm_size;
         int i;
         ENTRY;
 
         if (lsm) {
-                int i, max = 0;
                 if (lsm->lsm_magic != LOV_MAGIC) {
-                        CERROR("bad mem LOV MAGIC: %#010x != %#010x\n",
+                        CERROR("bad mem LOV MAGIC: 0x%08X != 0x%08X\n",
                                lsm->lsm_magic, LOV_MAGIC);
                         RETURN(-EINVAL);
                 }
-                stripe_count = lsm->lsm_stripe_count;
-
-                for (i = 0,loi = lsm->lsm_oinfo; i < stripe_count; i++,loi++) {
-                        if (loi->loi_ost_idx > max)
-                                max = loi->loi_ost_idx;
+                if (!lmmp) {
+                        stripe_count = lov_get_stripecnt(lov, lsm->lsm_stripe_count);
+                        lsm->lsm_stripe_count = stripe_count;
+                } else {
+                        stripe_count = lsm->lsm_stripe_count;
                 }
-                ost_count = max + 1;
         }
 
         /* XXX LOV STACKING call into osc for sizes */
-        lmm_size = lov_mds_md_size(ost_count);
+        lmm_size = lov_mds_md_size(stripe_count);
 
         if (!lmmp)
                 RETURN(lmm_size);
 
         if (*lmmp && !lsm) {
-                ost_count = le32_to_cpu ((*lmmp)->lmm_ost_count);
-                OBD_FREE(*lmmp, lov_mds_md_size(ost_count));
+                stripe_count = le32_to_cpu((*lmmp)->lmm_stripe_count);
+                OBD_FREE(*lmmp, lov_mds_md_size(stripe_count));
                 *lmmp = NULL;
                 RETURN(0);
         }
@@ -115,164 +136,141 @@ int lov_packmd(struct lustre_handle *conn, struct lov_mds_md **lmmp,
         }
 
         lmm = *lmmp;
-        lmm->lmm_magic = cpu_to_le32 (LOV_MAGIC);
-        lmm->lmm_ost_count = cpu_to_le16 (ost_count);
+        lmm->lmm_magic = cpu_to_le32(LOV_MAGIC); /* only write new format */
 
         if (!lsm)
                 RETURN(lmm_size);
 
-        lmm->lmm_object_id = cpu_to_le64 (lsm->lsm_object_id);
-        lmm->lmm_stripe_count = cpu_to_le16 (stripe_count);
-        lmm->lmm_stripe_size = cpu_to_le32 (lsm->lsm_stripe_size);
-        lmm->lmm_stripe_offset = cpu_to_le32 (lsm->lsm_stripe_offset);
-
-        /* Only fill in the object ids which we are actually using.
-         * Assumes lmm_objects is otherwise zero-filled. */
-        for (i = 0, loi = lsm->lsm_oinfo; i < stripe_count; i++, loi++) {
-                /* XXX call down to osc_packmd() to do the packing */
-                LASSERT (loi->loi_id);
-                lmm->lmm_objects[loi->loi_ost_idx].l_object_id =
-                        cpu_to_le64 (loi->loi_id);
+        lmm->lmm_object_id = cpu_to_le64(lsm->lsm_object_id);
+        lmm->lmm_object_gr = cpu_to_le64(lsm->lsm_object_gr);
+        lmm->lmm_stripe_size = cpu_to_le32(lsm->lsm_stripe_size);
+        lmm->lmm_stripe_count = cpu_to_le32(stripe_count);
+        lmm->lmm_pattern = cpu_to_le32(lsm->lsm_pattern);
+
+        for (i = 0; i < stripe_count; i++) {
+                struct lov_oinfo *loi = lsm->lsm_oinfo[i];
+
+                /* XXX LOV STACKING call down to osc_packmd() to do packing */
+                LASSERTF(loi->loi_id, "lmm_oid "LPU64" stripe %u/%u idx %u\n",
+                         lmm->lmm_object_id, i, stripe_count, loi->loi_ost_idx);
+                lmm->lmm_objects[i].l_object_id = cpu_to_le64(loi->loi_id);
+                lmm->lmm_objects[i].l_object_gr = cpu_to_le64(loi->loi_gr);
+                lmm->lmm_objects[i].l_ost_gen = cpu_to_le32(loi->loi_ost_gen);
+                lmm->lmm_objects[i].l_ost_idx = cpu_to_le32(loi->loi_ost_idx);
         }
 
         RETURN(lmm_size);
 }
 
-int lov_get_stripecnt(struct lov_obd *lov, int stripe_count)
+/* Find the max stripecount we should use */
+int lov_get_stripecnt(struct lov_obd *lov, __u32 stripe_count)
 {
         if (!stripe_count)
                 stripe_count = lov->desc.ld_default_stripe_count;
-        if (!stripe_count || stripe_count > lov->desc.ld_active_tgt_count)
+        if (stripe_count > lov->desc.ld_active_tgt_count)
                 stripe_count = lov->desc.ld_active_tgt_count;
+        if (!stripe_count)
+                stripe_count = 1;
+        /* for now, we limit the stripe count directly, when bug 4424 is
+         * fixed this needs to be somewhat dynamic based on whether ext3
+         * can handle larger EA sizes. */
+        if (stripe_count > LOV_MAX_STRIPE_COUNT)
+                stripe_count = LOV_MAX_STRIPE_COUNT;
 
         return stripe_count;
 }
 
-static int lov_verify_lmm(struct lov_mds_md *lmm, int lmm_bytes,
-                          int *ost_count, int *stripe_count, int *ost_offset)
-{
-        if (lmm_bytes < sizeof(*lmm)) {
-                CERROR("lov_mds_md too small: %d, need at least %d\n",
-                       lmm_bytes, (int)sizeof(*lmm));
-                return -EINVAL;
-        }
 
-        if (le32_to_cpu(lmm->lmm_magic) != LOV_MAGIC) {
-                CERROR("bad disk LOV MAGIC: %#08x != %#08x\n",
-                       le32_to_cpu(lmm->lmm_magic), LOV_MAGIC);
-                lov_dump_lmm(D_WARNING, lmm);
-                return -EINVAL;
-        }
-
-        *ost_count = le16_to_cpu(lmm->lmm_ost_count);
-        *stripe_count = le16_to_cpu(lmm->lmm_stripe_count);
-        *ost_offset = le32_to_cpu(lmm->lmm_stripe_offset);
-
-        if (*ost_count == 0 || *stripe_count == 0) {
-                CERROR("zero OST count %d or stripe count %d\n",
-                       *ost_count, *stripe_count);
-                lov_dump_lmm(D_WARNING, lmm);
-                return -EINVAL;
-        }
-
-        if (lmm_bytes < lov_mds_md_size(*ost_count)) {
-                CERROR("lov_mds_md too small: %d, need %d\n",
-                       lmm_bytes, lov_mds_md_size(*ost_count));
-                lov_dump_lmm(D_WARNING, lmm);
-                return -EINVAL;
-        }
-
-        if (*ost_offset > *ost_count) {
-                CERROR("starting OST offset %d > number of OSTs %d\n",
-                       *ost_offset, *ost_count);
-                lov_dump_lmm(D_WARNING, lmm);
-                return -EINVAL;
-        }
-
-        if (*stripe_count > *ost_count) {
-                CERROR("stripe count %d > number of OSTs %d\n",
-                       *stripe_count, *ost_count);
-                lov_dump_lmm(D_WARNING, lmm);
-                return -EINVAL;
-        }
+static int lov_verify_lmm(void *lmm, int lmm_bytes, int *stripe_count)
+{
+        int rc;
 
-        if (lmm->lmm_object_id == 0) {
-                CERROR("zero object id\n");
-                lov_dump_lmm(D_WARNING, lmm);
+        if (lsm_op_find(le32_to_cpu(*(__u32 *)lmm)) == NULL) {
+                CERROR("bad disk LOV MAGIC: 0x%08X; dumping V1 LMM:\n",
+                       le32_to_cpu(*(__u32 *)lmm));
+                lov_dump_lmm_v1(D_WARNING, lmm);
                 return -EINVAL;
         }
-
-        return 0;
+        rc = lsm_op_find(le32_to_cpu(*(__u32 *)lmm))->lsm_lmm_verify(lmm,
+                                     lmm_bytes, stripe_count);
+        return rc;
 }
 
-int lov_alloc_memmd(struct lov_stripe_md **lsmp, int stripe_count)
+int lov_alloc_memmd(struct lov_stripe_md **lsmp, int stripe_count,
+                      int pattern, int magic)
 {
-        int lsm_size = lov_stripe_md_size(stripe_count);
-        struct lov_oinfo *loi;
-        int i;
+        int i, lsm_size;
+        ENTRY;
+
+        CDEBUG(D_INFO, "alloc lsm, stripe_count %d\n", stripe_count);
 
-        OBD_ALLOC(*lsmp, lsm_size);
-        if (!*lsmp)
-                return -ENOMEM;
+        *lsmp = lsm_alloc_plain(stripe_count, &lsm_size);
+        if (!*lsmp) {
+                CERROR("can't allocate lsmp stripe_count %d\n", stripe_count);
+                RETURN(-ENOMEM);
+        }
 
-        (*lsmp)->lsm_magic = LOV_MAGIC;
+        spin_lock_init(&(*lsmp)->lsm_lock);
+        (*lsmp)->lsm_magic = magic;
         (*lsmp)->lsm_stripe_count = stripe_count;
         (*lsmp)->lsm_maxbytes = LUSTRE_STRIPE_MAXBYTES * stripe_count;
+        (*lsmp)->lsm_pattern = pattern;
+        (*lsmp)->lsm_oinfo[0]->loi_ost_idx = ~0;
 
-        for (i = 0, loi = (*lsmp)->lsm_oinfo; i < stripe_count; i++, loi++){
-                loi->loi_dirty_ot = &loi->loi_dirty_ot_inline;
-                ot_init(loi->loi_dirty_ot);
-        }
-        return lsm_size;
+        for (i = 0; i < stripe_count; i++)
+                loi_init((*lsmp)->lsm_oinfo[i]);
+
+        RETURN(lsm_size);
 }
 
 void lov_free_memmd(struct lov_stripe_md **lsmp)
 {
-        OBD_FREE(*lsmp, lov_stripe_md_size((*lsmp)->lsm_stripe_count));
+        struct lov_stripe_md *lsm = *lsmp;
+
+        LASSERT(lsm_op_find(lsm->lsm_magic) != NULL);
+        lsm_op_find(lsm->lsm_magic)->lsm_free(lsm);
+
         *lsmp = NULL;
 }
 
+
 /* Unpack LOV object metadata from disk storage.  It is packed in LE byte
  * order and is opaque to the networking layer.
  */
-int lov_unpackmd(struct lustre_handle *conn, struct lov_stripe_md **lsmp,
+int lov_unpackmd(struct obd_export *exp,  struct lov_stripe_md **lsmp,
                  struct lov_mds_md *lmm, int lmm_bytes)
 {
-        struct obd_device *obd = class_conn2obd(conn);
+        struct obd_device *obd = class_exp2obd(exp);
         struct lov_obd *lov = &obd->u.lov;
-        struct lov_stripe_md *lsm;
-        struct lov_oinfo *loi;
-        int ost_count;
-        int ost_offset;
-        int stripe_count;
-        int lsm_size;
-        int i;
+        int rc = 0, stripe_count, lsm_size;
+        __u32 magic;
         ENTRY;
 
         /* If passed an MDS struct use values from there, otherwise defaults */
         if (lmm) {
-                i = lov_verify_lmm(lmm, lmm_bytes, &ost_count, &stripe_count,
-                                   &ost_offset);
-                if (i)
-                        RETURN(i);
+                rc = lov_verify_lmm(lmm, lmm_bytes, &stripe_count);
+                if (rc)
+                        RETURN(rc);
+                magic = le32_to_cpu(lmm->lmm_magic);
         } else {
-                ost_count = 0;
                 stripe_count = lov_get_stripecnt(lov, 0);
-                ost_offset = 0;
+                magic = LOV_MAGIC;
         }
 
         /* If we aren't passed an lsmp struct, we just want the size */
-        if (!lsmp)
+        if (!lsmp) {
                 /* XXX LOV STACKING call into osc for sizes */
+                LBUG();
                 RETURN(lov_stripe_md_size(stripe_count));
-
+        }
         /* If we are passed an allocated struct but nothing to unpack, free */
         if (*lsmp && !lmm) {
                 lov_free_memmd(lsmp);
                 RETURN(0);
         }
 
-        lsm_size = lov_alloc_memmd(lsmp, stripe_count);
+        lsm_size = lov_alloc_memmd(lsmp, stripe_count, LOV_PATTERN_RAID0,
+                                   magic);
         if (lsm_size < 0)
                 RETURN(lsm_size);
 
@@ -280,138 +278,201 @@ int lov_unpackmd(struct lustre_handle *conn, struct lov_stripe_md **lsmp,
         if (!lmm)
                 RETURN(lsm_size);
 
-        lsm = *lsmp;
-        lsm->lsm_object_id = le64_to_cpu(lmm->lmm_object_id);
-        lsm->lsm_stripe_size = le32_to_cpu(lmm->lmm_stripe_size);
-        lsm->lsm_stripe_offset = ost_offset;
-
-        for (i = 0, loi = lsm->lsm_oinfo; i < ost_count; i++, ost_offset++) {
-                ost_offset %= ost_count;
-
-                if (!lmm->lmm_objects[ost_offset].l_object_id)
-                        continue;
-
-                /* XXX LOV STACKING call down to osc_unpackmd() */
-                loi->loi_id =
-                        le64_to_cpu(lmm->lmm_objects[ost_offset].l_object_id);
-                loi->loi_ost_idx = ost_offset;
-                loi++;
-        }
-
-        if (loi - lsm->lsm_oinfo != stripe_count) {
-                CERROR("missing objects in lmm struct\n");
-                lov_dump_lmm(D_WARNING, lmm);
+        LASSERT(lsm_op_find(magic) != NULL);
+        rc = lsm_op_find(magic)->lsm_unpackmd(lov, *lsmp, lmm);
+        if (rc) {
                 lov_free_memmd(lsmp);
-                RETURN(-EINVAL);
+                RETURN(rc);
         }
 
-
         RETURN(lsm_size);
 }
 
-/* Configure object striping information on a new file.
- *
- * @lmmu is a pointer to a user struct with one or more of the fields set to
- * indicate the application preference: lmm_stripe_count, lmm_stripe_size,
- * lmm_stripe_offset, and lmm_stripe_pattern.  lmm_magic must be LOV_MAGIC.
- * @lsmp is a pointer to an in-core stripe MD that needs to be filled in.
- */
-int lov_setstripe(struct lustre_handle *conn, struct lov_stripe_md **lsmp,
-                  struct lov_mds_md *lmmu)
+static int __lov_setstripe(struct obd_export *exp, struct lov_stripe_md **lsmp,
+                           struct lov_user_md *lump)
 {
-        struct obd_device *obd = class_conn2obd(conn);
+        struct obd_device *obd = class_exp2obd(exp);
         struct lov_obd *lov = &obd->u.lov;
-        struct lov_mds_md lmm;
+        struct lov_user_md lum;
         int stripe_count;
         int rc;
         ENTRY;
 
-        rc = copy_from_user(&lmm, lmmu, sizeof(lmm));
+        rc = copy_from_user(&lum, lump, sizeof(lum));
         if (rc)
                 RETURN(-EFAULT);
 
-        /* Bug 1185 FIXME: struct lov_mds_md is little-endian everywhere else */
+        if (lum.lmm_magic != LOV_USER_MAGIC) {
+                if (lum.lmm_magic == __swab32(LOV_USER_MAGIC)) {
+                        lustre_swab_lov_user_md(&lum);
+                } else {
+                        CDEBUG(D_IOCTL, "bad userland LOV MAGIC:"
+                               " %#08x != %#08x\n",
+                               lum.lmm_magic, LOV_USER_MAGIC);
+                        RETURN(-EINVAL);
+                }
+        }
 
-        if (lmm.lmm_magic != LOV_MAGIC) {
-                CDEBUG(D_IOCTL, "bad userland LOV MAGIC: %#08x != %#08x\n",
-                       lmm.lmm_magic, LOV_MAGIC);
-                RETURN(-EINVAL);
+        if (lum.lmm_pattern == 0) {
+                lum.lmm_pattern = lov->desc.ld_pattern ?
+                        lov->desc.ld_pattern : LOV_PATTERN_RAID0;
         }
-#if 0   /* the stripe_count/offset is "advisory", and it gets fixed later */
-        if (lmm.lmm_stripe_count > lov->desc.ld_tgt_count &&
-            lmm.lmm_stripe_count != 0xffffffff) {
-                CERROR("stripe count %u more than OST count %d\n",
-                       lmm.lmm_stripe_count, lov->desc.ld_tgt_count);
+
+        if (lum.lmm_pattern != LOV_PATTERN_RAID0) {
+                CDEBUG(D_IOCTL, "bad userland stripe pattern: %#x\n",
+                       lum.lmm_pattern);
                 RETURN(-EINVAL);
         }
-        if (lmm.lmm_stripe_offset >= lov->desc.ld_tgt_count &&
-            lmm.lmm_stripe_offset != 0xffffffff) {
-                CERROR("stripe offset %u more than max OST index %d\n",
-                       lmm.lmm_stripe_offset, lov->desc.ld_tgt_count);
-                RETURN(-EINVAL);
+
+        /* 64kB is the largest common page size we see (ia64), and matches the
+         * check in lfs */
+        if (lum.lmm_stripe_size & (LOV_MIN_STRIPE_SIZE - 1)) {
+                CDEBUG(D_IOCTL, "stripe size %u not multiple of %u, fixing\n",
+                       lum.lmm_stripe_size, LOV_MIN_STRIPE_SIZE);
+                lum.lmm_stripe_size = LOV_MIN_STRIPE_SIZE;
         }
-#endif
-        if (lmm.lmm_stripe_size & (PAGE_SIZE - 1)) {
-                CDEBUG(D_IOCTL, "stripe size %u not multiple of %lu\n",
-                       lmm.lmm_stripe_size, PAGE_SIZE);
+
+        if ((lum.lmm_stripe_offset >= lov->desc.ld_tgt_count) &&
+            (lum.lmm_stripe_offset != (typeof(lum.lmm_stripe_offset))(-1))) {
+                CDEBUG(D_IOCTL, "stripe offset %u > number of OSTs %u\n",
+                       lum.lmm_stripe_offset, lov->desc.ld_tgt_count);
                 RETURN(-EINVAL);
         }
-        stripe_count = lov_get_stripecnt(lov, lmm.lmm_stripe_count);
+        stripe_count = lov_get_stripecnt(lov, lum.lmm_stripe_count);
 
-        if ((__u64)lmm.lmm_stripe_size * stripe_count > ~0UL) {
-                CDEBUG(D_IOCTL, "stripe width %ux%u > %lu on 32-bit system\n",
-                       lmm.lmm_stripe_size, (int)lmm.lmm_stripe_count, ~0UL);
+        if ((__u64)lum.lmm_stripe_size * stripe_count > ~0UL) {
+                CDEBUG(D_IOCTL, "stripe width %ux%i exeeds %lu bytes\n",
+                       lum.lmm_stripe_size, (int)lum.lmm_stripe_count, ~0UL);
                 RETURN(-EINVAL);
         }
 
-        rc = lov_alloc_memmd(lsmp, stripe_count);
+        rc = lov_alloc_memmd(lsmp, stripe_count, lum.lmm_pattern, LOV_MAGIC);
 
-        if (rc < 0)
-                RETURN(rc);
+        if (rc >= 0) {
+                (*lsmp)->lsm_oinfo[0]->loi_ost_idx = lum.lmm_stripe_offset;
+                (*lsmp)->lsm_stripe_size = lum.lmm_stripe_size;
+                rc = 0;
+        }
+
+        RETURN(0);
+}
+
+/* Configure object striping information on a new file.
+ *
+ * @lmmu is a pointer to a user struct with one or more of the fields set to
+ * indicate the application preference: lmm_stripe_count, lmm_stripe_size,
+ * lmm_stripe_offset, and lmm_stripe_pattern.  lmm_magic must be LOV_MAGIC.
+ * @lsmp is a pointer to an in-core stripe MD that needs to be filled in.
+ */
+int lov_setstripe(struct obd_export *exp, struct lov_stripe_md **lsmp,
+                  struct lov_user_md *lump)
+{
+        int rc;
+        mm_segment_t seg;
+
+        seg = get_fs();
+        set_fs(KERNEL_DS);
+
+        rc = __lov_setstripe(exp, lsmp, lump);
+        set_fs(seg);
+        RETURN(rc);
+}
+
+int lov_setea(struct obd_export *exp, struct lov_stripe_md **lsmp,
+              struct lov_user_md *lump)
+{
+        int i;
+        int rc;
+        struct obd_export *oexp;
+        struct lov_obd *lov = &exp->exp_obd->u.lov;
+        obd_id last_id = 0;
 
-        (*lsmp)->lsm_stripe_offset = lmm.lmm_stripe_offset;
-        (*lsmp)->lsm_stripe_size = lmm.lmm_stripe_size;
+        ENTRY;
+        for (i = 0; i < lump->lmm_stripe_count; i++) {
+                __u32 len = sizeof(last_id);
+                oexp = lov->lov_tgts[lump->lmm_objects[i].l_ost_idx]->ltd_exp;
+                rc = obd_get_info(oexp, strlen("last_id"), "last_id",
+                                  &len, &last_id);
+                if (rc)
+                        RETURN(rc);
+                if (lump->lmm_objects[i].l_object_id > last_id) {
+                        CERROR("Setting EA for object > than last id on "
+                               "ost idx %d "LPD64" > "LPD64" \n",
+                               lump->lmm_objects[i].l_ost_idx,
+                               lump->lmm_objects[i].l_object_id, last_id);
+                        RETURN(-EINVAL);
+                }
+        }
 
+        rc = lov_setstripe(exp, lsmp, lump);
+        if (rc)
+                RETURN(rc);
+
+        for (i = 0; i < lump->lmm_stripe_count; i++) {
+                (*lsmp)->lsm_oinfo[i]->loi_ost_idx =
+                        lump->lmm_objects[i].l_ost_idx;
+                (*lsmp)->lsm_oinfo[i]->loi_id = lump->lmm_objects[i].l_object_id;
+                (*lsmp)->lsm_oinfo[i]->loi_gr = lump->lmm_objects[i].l_object_gr;
+        }
         RETURN(0);
 }
 
+
 /* Retrieve object striping information.
  *
- * @lmmu is a pointer to an in-core struct with lmm_ost_count indicating
+ * @lump is a pointer to an in-core struct with lmm_ost_count indicating
  * the maximum number of OST indices which will fit in the user buffer.
- * lmm_magic must be LOV_MAGIC.
+ * lmm_magic must be LOV_USER_MAGIC.
  */
-int lov_getstripe(struct lustre_handle *conn, struct lov_stripe_md *lsm,
-                  struct lov_mds_md *lmmu)
+int lov_getstripe(struct obd_export *exp, struct lov_stripe_md *lsm,
+                  struct lov_user_md *lump)
 {
-        struct lov_mds_md lmm, *lmmk = NULL;
+        /*
+         * XXX huge struct allocated on stack.
+         */
+        struct lov_user_md lum;
+        struct lov_mds_md *lmmk = NULL;
         int rc, lmm_size;
+        mm_segment_t seg;
         ENTRY;
 
         if (!lsm)
                 RETURN(-ENODATA);
 
-        rc = copy_from_user(&lmm, lmmu, sizeof(lmm));
+        /*
+         * "Switch to kernel segment" to allow copying from kernel space by
+         * copy_{to,from}_user().
+         */
+        seg = get_fs();
+        set_fs(KERNEL_DS);
+        rc = copy_from_user(&lum, lump, sizeof(lum));
         if (rc)
-                RETURN(-EFAULT);
-
-        if (lmm.lmm_magic != LOV_MAGIC)
-                RETURN(-EINVAL);
-
-        rc = lov_packmd(conn, &lmmk, lsm);
-        if (rc < 0)
-                RETURN(rc);
-        /* Bug 1185 FIXME: convert lmmk to big-endian before copy to userspace */
-        lmm_size = rc;
-        rc = 0;
-
-        /* User wasn't expecting this many OST entries */
-        if (lmm.lmm_ost_count < lmmk->lmm_ost_count)
-                rc = -EOVERFLOW;
-        else if (copy_to_user(lmmu, lmmk, lmm_size))
                 rc = -EFAULT;
-
-        obd_free_diskmd (conn, &lmmk);
-
+        else if (lum.lmm_magic != LOV_USER_MAGIC)
+                rc = -EINVAL;
+        else {
+                rc = lov_packmd(exp, &lmmk, lsm);
+                if (rc < 0)
+                        RETURN(rc);
+                lmm_size = rc;
+                rc = 0;
+
+                /* FIXME: Bug 1185 - copy fields properly when structs change */
+                CLASSERT(sizeof lum == sizeof *lmmk);
+                CLASSERT(sizeof lum.lmm_objects[0] ==
+                         sizeof lmmk->lmm_objects[0]);
+
+                /* User wasn't expecting this many OST entries */
+                if (lum.lmm_stripe_count == 0) {
+                        if (copy_to_user(lump, lmmk, sizeof lum))
+                                rc = -EFAULT;
+                } else if (lum.lmm_stripe_count < lmmk->lmm_stripe_count) {
+                        rc = -EOVERFLOW;
+                } else if (copy_to_user(lump, lmmk, sizeof lum))
+                        rc = -EFAULT;
+
+                obd_free_diskmd(exp, &lmmk);
+        }
+        set_fs(seg);
         RETURN(rc);
 }