1 files changed, 231 insertions, 1 deletions

diff --git a/fs/nfs/objlayout/objlayout.c b/fs/nfs/objlayout/objlayout.c
index 5157ef6d0041..f7caecff6b4d 100644
--- a/fs/nfs/objlayout/objlayout.c
+++ b/fs/nfs/objlayout/objlayout.c
@@ -50,6 +50,10 @@ objlayout_alloc_layout_hdr(struct inode *inode, gfp_t gfp_flags)
         struct objlayout *objlay;
 
         objlay = kzalloc(sizeof(struct objlayout), gfp_flags);
+        if (objlay) {
+                spin_lock_init(&objlay->lock);
+                INIT_LIST_HEAD(&objlay->err_list);
+        }
         dprintk("%s: Return %p\n", __func__, objlay);
         return &objlay->pnfs_layout;
 }
@@ -64,6 +68,7 @@ objlayout_free_layout_hdr(struct pnfs_layout_hdr *lo)
 
         dprintk("%s: objlay %p\n", __func__, objlay);
 
+        WARN_ON(!list_empty(&objlay->err_list));
         kfree(objlay);
 }
 
@@ -183,6 +188,7 @@ objlayout_alloc_io_state(struct pnfs_layout_hdr *pnfs_layout_type,
                 pgbase &= ~PAGE_MASK;
         }
 
+        INIT_LIST_HEAD(&state->err_list);
         state->lseg = lseg;
         state->rpcdata = rpcdata;
         state->pages = pages;
@@ -213,7 +219,52 @@ objlayout_iodone(struct objlayout_io_state *state)
 {
         dprintk("%s: state %p status\n", __func__, state);
 
-        objlayout_free_io_state(state);
+        if (likely(state->status >= 0)) {
+                objlayout_free_io_state(state);
+        } else {
+                struct objlayout *objlay = OBJLAYOUT(state->lseg->pls_layout);
+
+                spin_lock(&objlay->lock);
+                list_add(&objlay->err_list, &state->err_list);
+                spin_unlock(&objlay->lock);
+        }
+}
+
+/*
+ * objlayout_io_set_result - Set an osd_error code on a specific osd comp.
+ *
+ * The @index component IO failed (error returned from target). Register
+ * the error for later reporting at layout-return.
+ */
+void
+objlayout_io_set_result(struct objlayout_io_state *state, unsigned index,
+                        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) {
+                ioerr->oer_component = *pooid;
+                ioerr->oer_comp_offset = offset;
+                ioerr->oer_comp_length = length;
+                ioerr->oer_iswrite = is_write;
+                ioerr->oer_errno = osd_error;
+
+                dprintk("%s: err[%d]: errno=%d is_write=%d dev(%llx:%llx) "
+                        "par=0x%llx obj=0x%llx offset=0x%llx length=0x%llx\n",
+                        __func__, index, ioerr->oer_errno,
+                        ioerr->oer_iswrite,
+                        _DEVID_LO(&ioerr->oer_component.oid_device_id),
+                        _DEVID_HI(&ioerr->oer_component.oid_device_id),
+                        ioerr->oer_component.oid_partition_id,
+                        ioerr->oer_component.oid_object_id,
+                        ioerr->oer_comp_offset,
+                        ioerr->oer_comp_length);
+        } else {
+                /* User need not call if no error is reported */
+                ioerr->oer_errno = 0;
+        }
 }
 
 /* Function scheduled on rpc workqueue to call ->nfs_readlist_complete().
@@ -382,6 +433,185 @@ objlayout_write_pagelist(struct nfs_write_data *wdata,
         return PNFS_ATTEMPTED;
 }
 
+static int
+err_prio(u32 oer_errno)
+{
+        switch (oer_errno) {
+        case 0:
+                return 0;
+
+        case PNFS_OSD_ERR_RESOURCE:
+                return OSD_ERR_PRI_RESOURCE;
+        case PNFS_OSD_ERR_BAD_CRED:
+                return OSD_ERR_PRI_BAD_CRED;
+        case PNFS_OSD_ERR_NO_ACCESS:
+                return OSD_ERR_PRI_NO_ACCESS;
+        case PNFS_OSD_ERR_UNREACHABLE:
+                return OSD_ERR_PRI_UNREACHABLE;
+        case PNFS_OSD_ERR_NOT_FOUND:
+                return OSD_ERR_PRI_NOT_FOUND;
+        case PNFS_OSD_ERR_NO_SPACE:
+                return OSD_ERR_PRI_NO_SPACE;
+        default:
+                WARN_ON(1);
+                /* fallthrough */
+        case PNFS_OSD_ERR_EIO:
+                return OSD_ERR_PRI_EIO;
+        }
+}
+
+static void
+merge_ioerr(struct pnfs_osd_ioerr *dest_err,
+            const struct pnfs_osd_ioerr *src_err)
+{
+        u64 dest_end, src_end;
+
+        if (!dest_err->oer_errno) {
+                *dest_err = *src_err;
+                /* accumulated device must be blank */
+                memset(&dest_err->oer_component.oid_device_id, 0,
+                        sizeof(dest_err->oer_component.oid_device_id));
+
+                return;
+        }
+
+        if (dest_err->oer_component.oid_partition_id !=
+                                src_err->oer_component.oid_partition_id)
+                dest_err->oer_component.oid_partition_id = 0;
+
+        if (dest_err->oer_component.oid_object_id !=
+                                src_err->oer_component.oid_object_id)
+                dest_err->oer_component.oid_object_id = 0;
+
+        if (dest_err->oer_comp_offset > src_err->oer_comp_offset)
+                dest_err->oer_comp_offset = src_err->oer_comp_offset;
+
+        dest_end = end_offset(dest_err->oer_comp_offset,
+                              dest_err->oer_comp_length);
+        src_end =  end_offset(src_err->oer_comp_offset,
+                              src_err->oer_comp_length);
+        if (dest_end < src_end)
+                dest_end = src_end;
+
+        dest_err->oer_comp_length = dest_end - dest_err->oer_comp_offset;
+
+        if ((src_err->oer_iswrite == dest_err->oer_iswrite) &&
+            (err_prio(src_err->oer_errno) > err_prio(dest_err->oer_errno))) {
+                        dest_err->oer_errno = src_err->oer_errno;
+        } else if (src_err->oer_iswrite) {
+                dest_err->oer_iswrite = true;
+                dest_err->oer_errno = src_err->oer_errno;
+        }
+}
+
+static void
+encode_accumulated_error(struct objlayout *objlay, __be32 *p)
+{
+        struct objlayout_io_state *state, *tmp;
+        struct pnfs_osd_ioerr accumulated_err = {.oer_errno = 0};
+
+        list_for_each_entry_safe(state, tmp, &objlay->err_list, err_list) {
+                unsigned i;
+
+                for (i = 0; i < state->num_comps; i++) {
+                        struct pnfs_osd_ioerr *ioerr = &state->ioerrs[i];
+
+                        if (!ioerr->oer_errno)
+                                continue;
+
+                        printk(KERN_ERR "%s: err[%d]: errno=%d is_write=%d "
+                                "dev(%llx:%llx) par=0x%llx obj=0x%llx "
+                                "offset=0x%llx length=0x%llx\n",
+                                __func__, i, ioerr->oer_errno,
+                                ioerr->oer_iswrite,
+                                _DEVID_LO(&ioerr->oer_component.oid_device_id),
+                                _DEVID_HI(&ioerr->oer_component.oid_device_id),
+                                ioerr->oer_component.oid_partition_id,
+                                ioerr->oer_component.oid_object_id,
+                                ioerr->oer_comp_offset,
+                                ioerr->oer_comp_length);
+
+                        merge_ioerr(&accumulated_err, ioerr);
+                }
+                list_del(&state->err_list);
+                objlayout_free_io_state(state);
+        }
+
+        pnfs_osd_xdr_encode_ioerr(p, &accumulated_err);
+}
+
+void
+objlayout_encode_layoutreturn(struct pnfs_layout_hdr *pnfslay,
+                              struct xdr_stream *xdr,
+                              const struct nfs4_layoutreturn_args *args)
+{
+        struct objlayout *objlay = OBJLAYOUT(pnfslay);
+        struct objlayout_io_state *state, *tmp;
+        __be32 *start;
+
+        dprintk("%s: Begin\n", __func__);
+        start = xdr_reserve_space(xdr, 4);
+        BUG_ON(!start);
+
+        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; i++) {
+                        struct pnfs_osd_ioerr *ioerr = &state->ioerrs[i];
+
+                        if (!ioerr->oer_errno)
+                                continue;
+
+                        dprintk("%s: err[%d]: errno=%d is_write=%d "
+                                "dev(%llx:%llx) par=0x%llx obj=0x%llx "
+                                "offset=0x%llx length=0x%llx\n",
+                                __func__, i, ioerr->oer_errno,
+                                ioerr->oer_iswrite,
+                                _DEVID_LO(&ioerr->oer_component.oid_device_id),
+                                _DEVID_HI(&ioerr->oer_component.oid_device_id),
+                                ioerr->oer_component.oid_partition_id,
+                                ioerr->oer_component.oid_object_id,
+                                ioerr->oer_comp_offset,
+                                ioerr->oer_comp_length);
+
+                        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);
+
+                        /* 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.
+                         */
+                        encode_accumulated_error(objlay, last_xdr);
+                        goto loop_done;
+                }
+                list_del(&state->err_list);
+                objlayout_free_io_state(state);
+        }
+loop_done:
+        spin_unlock(&objlay->lock);
+
+        *start = cpu_to_be32((xdr->p - start - 1) * 4);
+        dprintk("%s: Return\n", __func__);
+}
+
+
 /*
  * Get Device Info API for io engines
  */