Merge commit 'v3.4.9' into android-t114-3.4
[linux-2.6.git] / fs / exofs / ore.c
index f1b7180..1585db1 100644 (file)
  */
 
 #include <linux/slab.h>
+#include <linux/module.h>
 #include <asm/div64.h>
+#include <linux/lcm.h>
 
-#include <scsi/osd_ore.h>
+#include "ore_raid.h"
 
-#define ORE_ERR(fmt, a...) printk(KERN_ERR "ore: " fmt, ##a)
+MODULE_AUTHOR("Boaz Harrosh <bharrosh@panasas.com>");
+MODULE_DESCRIPTION("Objects Raid Engine ore.ko");
+MODULE_LICENSE("GPL");
 
-#ifdef CONFIG_EXOFS_DEBUG
-#define ORE_DBGMSG(fmt, a...) \
-       printk(KERN_NOTICE "ore @%s:%d: " fmt, __func__, __LINE__, ##a)
-#else
-#define ORE_DBGMSG(fmt, a...) \
-       do { if (0) printk(fmt, ##a); } while (0)
-#endif
+/* ore_verify_layout does a couple of things:
+ * 1. Given a minimum number of needed parameters fixes up the rest of the
+ *    members to be operatonals for the ore. The needed parameters are those
+ *    that are defined by the pnfs-objects layout STD.
+ * 2. Check to see if the current ore code actually supports these parameters
+ *    for example stripe_unit must be a multple of the system PAGE_SIZE,
+ *    and etc...
+ * 3. Cache some havily used calculations that will be needed by users.
+ */
 
-/* u64 has problems with printk this will cast it to unsigned long long */
-#define _LLU(x) (unsigned long long)(x)
+enum { BIO_MAX_PAGES_KMALLOC =
+               (PAGE_SIZE - sizeof(struct bio)) / sizeof(struct bio_vec),};
 
-#define ORE_DBGMSG2(M...) do {} while (0)
-/* #define ORE_DBGMSG2 ORE_DBGMSG */
+int ore_verify_layout(unsigned total_comps, struct ore_layout *layout)
+{
+       u64 stripe_length;
+
+       switch (layout->raid_algorithm) {
+       case PNFS_OSD_RAID_0:
+               layout->parity = 0;
+               break;
+       case PNFS_OSD_RAID_5:
+               layout->parity = 1;
+               break;
+       case PNFS_OSD_RAID_PQ:
+       case PNFS_OSD_RAID_4:
+       default:
+               ORE_ERR("Only RAID_0/5 for now\n");
+               return -EINVAL;
+       }
+       if (0 != (layout->stripe_unit & ~PAGE_MASK)) {
+               ORE_ERR("Stripe Unit(0x%llx)"
+                         " must be Multples of PAGE_SIZE(0x%lx)\n",
+                         _LLU(layout->stripe_unit), PAGE_SIZE);
+               return -EINVAL;
+       }
+       if (layout->group_width) {
+               if (!layout->group_depth) {
+                       ORE_ERR("group_depth == 0 && group_width != 0\n");
+                       return -EINVAL;
+               }
+               if (total_comps < (layout->group_width * layout->mirrors_p1)) {
+                       ORE_ERR("Data Map wrong, "
+                               "numdevs=%d < group_width=%d * mirrors=%d\n",
+                               total_comps, layout->group_width,
+                               layout->mirrors_p1);
+                       return -EINVAL;
+               }
+               layout->group_count = total_comps / layout->mirrors_p1 /
+                                               layout->group_width;
+       } else {
+               if (layout->group_depth) {
+                       printk(KERN_NOTICE "Warning: group_depth ignored "
+                               "group_width == 0 && group_depth == %lld\n",
+                               _LLU(layout->group_depth));
+               }
+               layout->group_width = total_comps / layout->mirrors_p1;
+               layout->group_depth = -1;
+               layout->group_count = 1;
+       }
 
-MODULE_AUTHOR("Boaz Harrosh <bharrosh@panasas.com>");
-MODULE_DESCRIPTION("Objects Raid Engine ore.ko");
-MODULE_LICENSE("GPL");
+       stripe_length = (u64)layout->group_width * layout->stripe_unit;
+       if (stripe_length >= (1ULL << 32)) {
+               ORE_ERR("Stripe_length(0x%llx) >= 32bit is not supported\n",
+                       _LLU(stripe_length));
+               return -EINVAL;
+       }
 
-static void ore_calc_stripe_info(struct ore_layout *layout, u64 file_offset,
-                                struct ore_striping_info *si);
+       layout->max_io_length =
+               (BIO_MAX_PAGES_KMALLOC * PAGE_SIZE - layout->stripe_unit) *
+                                                       layout->group_width;
+       if (layout->parity) {
+               unsigned stripe_length =
+                               (layout->group_width - layout->parity) *
+                               layout->stripe_unit;
+
+               layout->max_io_length /= stripe_length;
+               layout->max_io_length *= stripe_length;
+       }
+       return 0;
+}
+EXPORT_SYMBOL(ore_verify_layout);
 
 static u8 *_ios_cred(struct ore_io_state *ios, unsigned index)
 {
@@ -69,21 +135,81 @@ static struct osd_dev *_ios_od(struct ore_io_state *ios, unsigned index)
        return ore_comp_dev(ios->oc, index);
 }
 
-static int  _get_io_state(struct ore_layout *layout,
-                         struct ore_components *oc, unsigned numdevs,
-                         struct ore_io_state **pios)
+int  _ore_get_io_state(struct ore_layout *layout,
+                       struct ore_components *oc, unsigned numdevs,
+                       unsigned sgs_per_dev, unsigned num_par_pages,
+                       struct ore_io_state **pios)
 {
        struct ore_io_state *ios;
+       struct page **pages;
+       struct osd_sg_entry *sgilist;
+       struct __alloc_all_io_state {
+               struct ore_io_state ios;
+               struct ore_per_dev_state per_dev[numdevs];
+               union {
+                       struct osd_sg_entry sglist[sgs_per_dev * numdevs];
+                       struct page *pages[num_par_pages];
+               };
+       } *_aios;
+
+       if (likely(sizeof(*_aios) <= PAGE_SIZE)) {
+               _aios = kzalloc(sizeof(*_aios), GFP_KERNEL);
+               if (unlikely(!_aios)) {
+                       ORE_DBGMSG("Failed kzalloc bytes=%zd\n",
+                                  sizeof(*_aios));
+                       *pios = NULL;
+                       return -ENOMEM;
+               }
+               pages = num_par_pages ? _aios->pages : NULL;
+               sgilist = sgs_per_dev ? _aios->sglist : NULL;
+               ios = &_aios->ios;
+       } else {
+               struct __alloc_small_io_state {
+                       struct ore_io_state ios;
+                       struct ore_per_dev_state per_dev[numdevs];
+               } *_aio_small;
+               union __extra_part {
+                       struct osd_sg_entry sglist[sgs_per_dev * numdevs];
+                       struct page *pages[num_par_pages];
+               } *extra_part;
+
+               _aio_small = kzalloc(sizeof(*_aio_small), GFP_KERNEL);
+               if (unlikely(!_aio_small)) {
+                       ORE_DBGMSG("Failed alloc first part bytes=%zd\n",
+                                  sizeof(*_aio_small));
+                       *pios = NULL;
+                       return -ENOMEM;
+               }
+               extra_part = kzalloc(sizeof(*extra_part), GFP_KERNEL);
+               if (unlikely(!extra_part)) {
+                       ORE_DBGMSG("Failed alloc second part bytes=%zd\n",
+                                  sizeof(*extra_part));
+                       kfree(_aio_small);
+                       *pios = NULL;
+                       return -ENOMEM;
+               }
 
-       /*TODO: Maybe use kmem_cach per sbi of size
-        * exofs_io_state_size(layout->s_numdevs)
-        */
-       ios = kzalloc(ore_io_state_size(numdevs), GFP_KERNEL);
-       if (unlikely(!ios)) {
-               ORE_DBGMSG("Failed kzalloc bytes=%d\n",
-                          ore_io_state_size(numdevs));
-               *pios = NULL;
-               return -ENOMEM;
+               pages = num_par_pages ? extra_part->pages : NULL;
+               sgilist = sgs_per_dev ? extra_part->sglist : NULL;
+               /* In this case the per_dev[0].sgilist holds the pointer to
+                * be freed
+                */
+               ios = &_aio_small->ios;
+               ios->extra_part_alloc = true;
+       }
+
+       if (pages) {
+               ios->parity_pages = pages;
+               ios->max_par_pages = num_par_pages;
+       }
+       if (sgilist) {
+               unsigned d;
+
+               for (d = 0; d < numdevs; ++d) {
+                       ios->per_dev[d].sglist = sgilist;
+                       sgilist += sgs_per_dev;
+               }
+               ios->sgs_per_dev = sgs_per_dev;
        }
 
        ios->layout = layout;
@@ -114,9 +240,42 @@ int  ore_get_rw_state(struct ore_layout *layout, struct ore_components *oc,
 {
        struct ore_io_state *ios;
        unsigned numdevs = layout->group_width * layout->mirrors_p1;
+       unsigned sgs_per_dev = 0, max_par_pages = 0;
        int ret;
 
-       ret = _get_io_state(layout, oc, numdevs, pios);
+       if (layout->parity && length) {
+               unsigned data_devs = layout->group_width - layout->parity;
+               unsigned stripe_size = layout->stripe_unit * data_devs;
+               unsigned pages_in_unit = layout->stripe_unit / PAGE_SIZE;
+               u32 remainder;
+               u64 num_stripes;
+               u64 num_raid_units;
+
+               num_stripes = div_u64_rem(length, stripe_size, &remainder);
+               if (remainder)
+                       ++num_stripes;
+
+               num_raid_units =  num_stripes * layout->parity;
+
+               if (is_reading) {
+                       /* For reads add per_dev sglist array */
+                       /* TODO: Raid 6 we need twice more. Actually:
+                       *         num_stripes / LCMdP(W,P);
+                       *         if (W%P != 0) num_stripes *= parity;
+                       */
+
+                       /* first/last seg is split */
+                       num_raid_units += layout->group_width;
+                       sgs_per_dev = div_u64(num_raid_units, data_devs) + 2;
+               } else {
+                       /* For Writes add parity pages array. */
+                       max_par_pages = num_raid_units * pages_in_unit *
+                                               sizeof(struct page *);
+               }
+       }
+
+       ret = _ore_get_io_state(layout, oc, numdevs, sgs_per_dev, max_par_pages,
+                               pios);
        if (unlikely(ret))
                return ret;
 
@@ -125,10 +284,11 @@ int  ore_get_rw_state(struct ore_layout *layout, struct ore_components *oc,
        ios->offset = offset;
 
        if (length) {
-               ore_calc_stripe_info(layout, offset, &ios->si);
-               ios->length = (length <= ios->si.group_length) ? length :
-                                                       ios->si.group_length;
+               ore_calc_stripe_info(layout, offset, length, &ios->si);
+               ios->length = ios->si.length;
                ios->nr_pages = (ios->length + PAGE_SIZE - 1) / PAGE_SIZE;
+               if (layout->parity)
+                       _ore_post_alloc_raid_stuff(ios);
        }
 
        return 0;
@@ -145,7 +305,7 @@ EXPORT_SYMBOL(ore_get_rw_state);
 int  ore_get_io_state(struct ore_layout *layout, struct ore_components *oc,
                      struct ore_io_state **pios)
 {
-       return _get_io_state(layout, oc, oc->numdevs, pios);
+       return _ore_get_io_state(layout, oc, oc->numdevs, 0, 0, pios);
 }
 EXPORT_SYMBOL(ore_get_io_state);
 
@@ -163,6 +323,7 @@ void ore_put_io_state(struct ore_io_state *ios)
                                bio_put(per_dev->bio);
                }
 
+               _ore_free_raid_stuff(ios);
                kfree(ios);
        }
 }
@@ -190,7 +351,7 @@ static void _done_io(struct osd_request *or, void *p)
        kref_put(&ios->kref, _last_io);
 }
 
-static int ore_io_execute(struct ore_io_state *ios)
+int ore_io_execute(struct ore_io_state *ios)
 {
        DECLARE_COMPLETION_ONSTACK(wait);
        bool sync = (ios->done == NULL);
@@ -250,7 +411,7 @@ static void _clear_bio(struct bio *bio)
        }
 }
 
-int ore_check_io(struct ore_io_state *ios, u64 *resid)
+int ore_check_io(struct ore_io_state *ios, ore_on_dev_error on_dev_error)
 {
        enum osd_err_priority acumulated_osd_err = 0;
        int acumulated_lin_err = 0;
@@ -258,7 +419,8 @@ int ore_check_io(struct ore_io_state *ios, u64 *resid)
 
        for (i = 0; i < ios->numdevs; i++) {
                struct osd_sense_info osi;
-               struct osd_request *or = ios->per_dev[i].or;
+               struct ore_per_dev_state *per_dev = &ios->per_dev[i];
+               struct osd_request *or = per_dev->or;
                int ret;
 
                if (unlikely(!or))
@@ -270,29 +432,31 @@ int ore_check_io(struct ore_io_state *ios, u64 *resid)
 
                if (OSD_ERR_PRI_CLEAR_PAGES == osi.osd_err_pri) {
                        /* start read offset passed endof file */
-                       _clear_bio(ios->per_dev[i].bio);
+                       _clear_bio(per_dev->bio);
                        ORE_DBGMSG("start read offset passed end of file "
                                "offset=0x%llx, length=0x%llx\n",
-                               _LLU(ios->per_dev[i].offset),
-                               _LLU(ios->per_dev[i].length));
+                               _LLU(per_dev->offset),
+                               _LLU(per_dev->length));
 
                        continue; /* we recovered */
                }
 
+               if (on_dev_error) {
+                       u64 residual = ios->reading ?
+                                       or->in.residual : or->out.residual;
+                       u64 offset = (ios->offset + ios->length) - residual;
+                       unsigned dev = per_dev->dev - ios->oc->first_dev;
+                       struct ore_dev *od = ios->oc->ods[dev];
+
+                       on_dev_error(ios, od, dev, osi.osd_err_pri,
+                                    offset, residual);
+               }
                if (osi.osd_err_pri >= acumulated_osd_err) {
                        acumulated_osd_err = osi.osd_err_pri;
                        acumulated_lin_err = ret;
                }
        }
 
-       /* TODO: raid specific residual calculations */
-       if (resid) {
-               if (likely(!acumulated_lin_err))
-                       *resid = 0;
-               else
-                       *resid = ios->length;
-       }
-
        return acumulated_lin_err;
 }
 EXPORT_SYMBOL(ore_check_io);
@@ -300,53 +464,65 @@ EXPORT_SYMBOL(ore_check_io);
 /*
  * L - logical offset into the file
  *
- * U - The number of bytes in a stripe within a group
+ * D - number of Data devices
+ *     D = group_width - parity
  *
- *     U = stripe_unit * group_width
+ * U - The number of bytes in a stripe within a group
+ *     U =  stripe_unit * D
  *
  * T - The number of bytes striped within a group of component objects
  *     (before advancing to the next group)
- *
- *     T = stripe_unit * group_width * group_depth
+ *     T = U * group_depth
  *
  * S - The number of bytes striped across all component objects
  *     before the pattern repeats
+ *     S = T * group_count
  *
- *     S = stripe_unit * group_width * group_depth * group_count
- *
- * M - The "major" (i.e., across all components) stripe number
- *
+ * M - The "major" (i.e., across all components) cycle number
  *     M = L / S
  *
- * G - Counts the groups from the beginning of the major stripe
- *
+ * G - Counts the groups from the beginning of the major cycle
  *     G = (L - (M * S)) / T   [or (L % S) / T]
  *
  * H - The byte offset within the group
- *
  *     H = (L - (M * S)) % T   [or (L % S) % T]
  *
  * N - The "minor" (i.e., across the group) stripe number
- *
  *     N = H / U
  *
  * C - The component index coresponding to L
  *
- *     C = (H - (N * U)) / stripe_unit + G * group_width
- *     [or (L % U) / stripe_unit + G * group_width]
+ *     C = (H - (N * U)) / stripe_unit + G * D
+ *     [or (L % U) / stripe_unit + G * D]
  *
  * O - The component offset coresponding to L
- *
  *     O = L % stripe_unit + N * stripe_unit + M * group_depth * stripe_unit
+ *
+ * LCMdP – Parity cycle: Lowest Common Multiple of group_width, parity
+ *          divide by parity
+ *     LCMdP = lcm(group_width, parity) / parity
+ *
+ * R - The parity Rotation stripe
+ *     (Note parity cycle always starts at a group's boundary)
+ *     R = N % LCMdP
+ *
+ * I = the first parity device index
+ *     I = (group_width + group_width - R*parity - parity) % group_width
+ *
+ * Craid - The component index Rotated
+ *     Craid = (group_width + C - R*parity) % group_width
+ *      (We add the group_width to avoid negative numbers modulo math)
  */
-static void ore_calc_stripe_info(struct ore_layout *layout, u64 file_offset,
-                                struct ore_striping_info *si)
+void ore_calc_stripe_info(struct ore_layout *layout, u64 file_offset,
+                         u64 length, struct ore_striping_info *si)
 {
        u32     stripe_unit = layout->stripe_unit;
        u32     group_width = layout->group_width;
        u64     group_depth = layout->group_depth;
+       u32     parity      = layout->parity;
 
-       u32     U = stripe_unit * group_width;
+       u32     D = group_width - parity;
+       u32     U = D * stripe_unit;
        u64     T = U * group_depth;
        u64     S = T * layout->group_count;
        u64     M = div64_u64(file_offset, S);
@@ -362,21 +538,43 @@ static void ore_calc_stripe_info(struct ore_layout *layout, u64 file_offset,
        u32     N = div_u64(H, U);
 
        /* "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 *= layout->mirrors_p1;
+       u32     C = (u32)(H - (N * U)) / stripe_unit + G * group_width;
 
        div_u64_rem(file_offset, stripe_unit, &si->unit_off);
 
        si->obj_offset = si->unit_off + (N * stripe_unit) +
                                  (M * group_depth * stripe_unit);
 
-       si->group_length = T - H;
+       if (parity) {
+               u32 LCMdP = lcm(group_width, parity) / parity;
+               /* R     = N % LCMdP; */
+               u32 RxP   = (N % LCMdP) * parity;
+               u32 first_dev = C - C % group_width;
+
+               si->par_dev = (group_width + group_width - parity - RxP) %
+                             group_width + first_dev;
+               si->dev = (group_width + C - RxP) % group_width + first_dev;
+               si->bytes_in_stripe = U;
+               si->first_stripe_start = M * S + G * T + N * U;
+       } else {
+               /* Make the math correct see _prepare_one_group */
+               si->par_dev = group_width;
+               si->dev = C;
+       }
+
+       si->dev *= layout->mirrors_p1;
+       si->par_dev *= layout->mirrors_p1;
+       si->offset = file_offset;
+       si->length = T - H;
+       if (si->length > length)
+               si->length = length;
        si->M = M;
 }
+EXPORT_SYMBOL(ore_calc_stripe_info);
 
-static int _add_stripe_unit(struct ore_io_state *ios,  unsigned *cur_pg,
-               unsigned pgbase, struct ore_per_dev_state *per_dev,
-               int cur_len)
+int _ore_add_stripe_unit(struct ore_io_state *ios,  unsigned *cur_pg,
+                        unsigned pgbase, struct page **pages,
+                        struct ore_per_dev_state *per_dev, int cur_len)
 {
        unsigned pg = *cur_pg;
        struct request_queue *q =
@@ -387,8 +585,11 @@ static int _add_stripe_unit(struct ore_io_state *ios,  unsigned *cur_pg,
        if (per_dev->bio == NULL) {
                unsigned pages_in_stripe = ios->layout->group_width *
                                        (ios->layout->stripe_unit / PAGE_SIZE);
-               unsigned bio_size = (ios->nr_pages + pages_in_stripe) /
-                                               ios->layout->group_width;
+               unsigned nr_pages = ios->nr_pages * ios->layout->group_width /
+                                       (ios->layout->group_width -
+                                        ios->layout->parity);
+               unsigned bio_size = (nr_pages + pages_in_stripe) /
+                                       ios->layout->group_width;
 
                per_dev->bio = bio_kmalloc(GFP_KERNEL, bio_size);
                if (unlikely(!per_dev->bio)) {
@@ -403,15 +604,18 @@ static int _add_stripe_unit(struct ore_io_state *ios,  unsigned *cur_pg,
                unsigned pglen = min_t(unsigned, PAGE_SIZE - pgbase, cur_len);
                unsigned added_len;
 
-               BUG_ON(ios->nr_pages <= pg);
                cur_len -= pglen;
 
-               added_len = bio_add_pc_page(q, per_dev->bio, ios->pages[pg],
+               added_len = bio_add_pc_page(q, per_dev->bio, pages[pg],
                                            pglen, pgbase);
                if (unlikely(pglen != added_len)) {
+                       ORE_DBGMSG("Failed bio_add_pc_page bi_vcnt=%u\n",
+                                  per_dev->bio->bi_vcnt);
                        ret = -ENOMEM;
                        goto out;
                }
+               _add_stripe_page(ios->sp2d, &ios->si, pages[pg]);
+
                pgbase = 0;
                ++pg;
        }
@@ -433,9 +637,11 @@ static int _prepare_for_striping(struct ore_io_state *ios)
        struct ore_striping_info *si = &ios->si;
        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 group_width = ios->layout->group_width;
+       unsigned devs_in_group = group_width * mirrors_p1;
        unsigned dev = si->dev;
        unsigned first_dev = dev - (dev % devs_in_group);
+       unsigned dev_order;
        unsigned cur_pg = ios->pages_consumed;
        u64 length = ios->length;
        int ret = 0;
@@ -445,7 +651,11 @@ static int _prepare_for_striping(struct ore_io_state *ios)
                return 0;
        }
 
-       BUG_ON(length > si->group_length);
+       BUG_ON(length > si->length);
+
+       dev_order = _dev_order(devs_in_group, mirrors_p1, si->par_dev, dev);
+       si->cur_comp = dev_order;
+       si->cur_pg = si->unit_off / PAGE_SIZE;
 
        while (length) {
                unsigned comp = dev - first_dev;
@@ -454,17 +664,20 @@ static int _prepare_for_striping(struct ore_io_state *ios)
 
                if (!per_dev->length) {
                        per_dev->dev = dev;
-                       if (dev < si->dev) {
-                               per_dev->offset = si->obj_offset + stripe_unit -
-                                                                  si->unit_off;
-                               cur_len = stripe_unit;
-                       } else if (dev == si->dev) {
+                       if (dev == si->dev) {
+                               WARN_ON(dev == si->par_dev);
                                per_dev->offset = si->obj_offset;
                                cur_len = stripe_unit - si->unit_off;
                                page_off = si->unit_off & ~PAGE_MASK;
                                BUG_ON(page_off && (page_off != ios->pgbase));
-                       } else { /* dev > si->dev */
-                               per_dev->offset = si->obj_offset - si->unit_off;
+                       } else {
+                               if (si->cur_comp > dev_order)
+                                       per_dev->offset =
+                                               si->obj_offset - si->unit_off;
+                               else /* si->cur_comp < dev_order */
+                                       per_dev->offset =
+                                               si->obj_offset + stripe_unit -
+                                                                  si->unit_off;
                                cur_len = stripe_unit;
                        }
                } else {
@@ -473,8 +686,8 @@ static int _prepare_for_striping(struct ore_io_state *ios)
                if (cur_len >= length)
                        cur_len = length;
 
-               ret = _add_stripe_unit(ios, &cur_pg, page_off , per_dev,
-                                      cur_len);
+               ret = _ore_add_stripe_unit(ios, &cur_pg, page_off, ios->pages,
+                                          per_dev, cur_len);
                if (unlikely(ret))
                        goto out;
 
@@ -482,17 +695,47 @@ static int _prepare_for_striping(struct ore_io_state *ios)
                dev = (dev % devs_in_group) + first_dev;
 
                length -= cur_len;
+
+               si->cur_comp = (si->cur_comp + 1) % group_width;
+               if (unlikely((dev == si->par_dev) || (!length && ios->sp2d))) {
+                       if (!length && ios->sp2d) {
+                               /* If we are writing and this is the very last
+                                * stripe. then operate on parity dev.
+                                */
+                               dev = si->par_dev;
+                       }
+                       if (ios->sp2d)
+                               /* In writes cur_len just means if it's the
+                                * last one. See _ore_add_parity_unit.
+                                */
+                               cur_len = length;
+                       per_dev = &ios->per_dev[dev - first_dev];
+                       if (!per_dev->length) {
+                               /* Only/always the parity unit of the first
+                                * stripe will be empty. So this is a chance to
+                                * initialize the per_dev info.
+                                */
+                               per_dev->dev = dev;
+                               per_dev->offset = si->obj_offset - si->unit_off;
+                       }
+
+                       ret = _ore_add_parity_unit(ios, si, per_dev, cur_len);
+                       if (unlikely(ret))
+                                       goto out;
+
+                       /* Rotate next par_dev backwards with wraping */
+                       si->par_dev = (devs_in_group + si->par_dev -
+                                      ios->layout->parity * mirrors_p1) %
+                                     devs_in_group + first_dev;
+                       /* Next stripe, start fresh */
+                       si->cur_comp = 0;
+                       si->cur_pg = 0;
+               }
        }
 out:
        ios->numdevs = devs_in_group;
        ios->pages_consumed = cur_pg;
-       if (unlikely(ret)) {
-               if (length == ios->length)
-                       return ret;
-               else
-                       ios->length -= length;
-       }
-       return 0;
+       return ret;
 }
 
 int ore_create(struct ore_io_state *ios)
@@ -594,11 +837,11 @@ static int _write_mirror(struct ore_io_state *ios, int cur_comp)
                                bio->bi_rw |= REQ_WRITE;
                        }
 
-                       osd_req_write(or, _ios_obj(ios, dev), per_dev->offset,
-                                     bio, per_dev->length);
+                       osd_req_write(or, _ios_obj(ios, cur_comp),
+                                     per_dev->offset, bio, per_dev->length);
                        ORE_DBGMSG("write(0x%llx) offset=0x%llx "
                                      "length=0x%llx dev=%d\n",
-                                    _LLU(_ios_obj(ios, dev)->id),
+                                    _LLU(_ios_obj(ios, cur_comp)->id),
                                     _LLU(per_dev->offset),
                                     _LLU(per_dev->length), dev);
                } else if (ios->kern_buff) {
@@ -610,20 +853,20 @@ static int _write_mirror(struct ore_io_state *ios, int cur_comp)
                               (ios->si.unit_off + ios->length >
                                ios->layout->stripe_unit));
 
-                       ret = osd_req_write_kern(or, _ios_obj(ios, per_dev->dev),
+                       ret = osd_req_write_kern(or, _ios_obj(ios, cur_comp),
                                                 per_dev->offset,
                                                 ios->kern_buff, ios->length);
                        if (unlikely(ret))
                                goto out;
                        ORE_DBGMSG2("write_kern(0x%llx) offset=0x%llx "
                                      "length=0x%llx dev=%d\n",
-                                    _LLU(_ios_obj(ios, dev)->id),
+                                    _LLU(_ios_obj(ios, cur_comp)->id),
                                     _LLU(per_dev->offset),
                                     _LLU(ios->length), per_dev->dev);
                } else {
-                       osd_req_set_attributes(or, _ios_obj(ios, dev));
+                       osd_req_set_attributes(or, _ios_obj(ios, cur_comp));
                        ORE_DBGMSG2("obj(0x%llx) set_attributes=%d dev=%d\n",
-                                    _LLU(_ios_obj(ios, dev)->id),
+                                    _LLU(_ios_obj(ios, cur_comp)->id),
                                     ios->out_attr_len, dev);
                }
 
@@ -645,6 +888,14 @@ int ore_write(struct ore_io_state *ios)
        int i;
        int ret;
 
+       if (unlikely(ios->sp2d && !ios->r4w)) {
+               /* A library is attempting a RAID-write without providing
+                * a pages lock interface.
+                */
+               WARN_ON_ONCE(1);
+               return -ENOTSUPP;
+       }
+
        ret = _prepare_for_striping(ios);
        if (unlikely(ret))
                return ret;
@@ -660,7 +911,7 @@ int ore_write(struct ore_io_state *ios)
 }
 EXPORT_SYMBOL(ore_write);
 
-static int _read_mirror(struct ore_io_state *ios, unsigned cur_comp)
+int _ore_read_mirror(struct ore_io_state *ios, unsigned cur_comp)
 {
        struct osd_request *or;
        struct ore_per_dev_state *per_dev = &ios->per_dev[cur_comp];
@@ -679,12 +930,24 @@ static int _read_mirror(struct ore_io_state *ios, unsigned cur_comp)
        per_dev->or = or;
 
        if (ios->pages) {
-               osd_req_read(or, obj, per_dev->offset,
-                               per_dev->bio, per_dev->length);
+               if (per_dev->cur_sg) {
+                       /* finalize the last sg_entry */
+                       _ore_add_sg_seg(per_dev, 0, false);
+                       if (unlikely(!per_dev->cur_sg))
+                               return 0; /* Skip parity only device */
+
+                       osd_req_read_sg(or, obj, per_dev->bio,
+                                       per_dev->sglist, per_dev->cur_sg);
+               } else {
+                       /* The no raid case */
+                       osd_req_read(or, obj, per_dev->offset,
+                                    per_dev->bio, per_dev->length);
+               }
+
                ORE_DBGMSG("read(0x%llx) offset=0x%llx length=0x%llx"
-                            " dev=%d\n", _LLU(obj->id),
+                            " dev=%d sg_len=%d\n", _LLU(obj->id),
                             _LLU(per_dev->offset), _LLU(per_dev->length),
-                            first_dev);
+                            first_dev, per_dev->cur_sg);
        } else {
                BUG_ON(ios->kern_buff);
 
@@ -712,7 +975,7 @@ int ore_read(struct ore_io_state *ios)
                return ret;
 
        for (i = 0; i < ios->numdevs; i += ios->layout->mirrors_p1) {
-               ret = _read_mirror(ios, i);
+               ret = _ore_read_mirror(ios, i);
                if (unlikely(ret))
                        return ret;
        }
@@ -781,7 +1044,7 @@ static void _calc_trunk_info(struct ore_layout *layout, u64 file_offset,
 {
        unsigned stripe_unit = layout->stripe_unit;
 
-       ore_calc_stripe_info(layout, file_offset, &ti->si);
+       ore_calc_stripe_info(layout, file_offset, 0, &ti->si);
 
        ti->prev_group_obj_off = ti->si.M * stripe_unit;
        ti->next_group_obj_off = ti->si.M ? (ti->si.M - 1) * stripe_unit : 0;