staging: lustre: lmv: implement lmv version of read_page
authorwang di <di.wang@intel.com>
Fri, 19 Aug 2016 18:07:25 +0000 (14:07 -0400)
committerGreg Kroah-Hartman <gregkh@linuxfoundation.org>
Sun, 21 Aug 2016 14:09:27 +0000 (16:09 +0200)
All the code needed to implement read_page. This
will eventually replace lmv_readpage.

Signed-off-by: wang di <di.wang@intel.com>
Reviewed-on: http://review.whamcloud.com/10761
Intel-bug-id: https://jira.hpdd.intel.com/browse/LU-4906
Reviewed-by: John L. Hammond <john.hammond@intel.com>
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>
Signed-off-by: James Simmons <jsimmons@infradead.org>
Signed-off-by: Greg Kroah-Hartman <gregkh@linuxfoundation.org>
drivers/staging/lustre/lustre/lmv/lmv_obd.c

index 3e41f49..1f01be4 100644 (file)
@@ -2368,6 +2368,334 @@ static int lmv_readpage(struct obd_export *exp, struct md_op_data *op_data,
        return rc;
 }
 
+/**
+ * Get current minimum entry from striped directory
+ *
+ * This function will search the dir entry, whose hash value is the
+ * closest(>=) to @hash_offset, from all of sub-stripes, and it is
+ * only being called for striped directory.
+ *
+ * \param[in] exp              export of LMV
+ * \param[in] op_data          parameters transferred beween client MD stack
+ *                             stripe_information will be included in this
+ *                             parameter
+ * \param[in] cb_op            ldlm callback being used in enqueue in
+ *                             mdc_read_page
+ * \param[in] hash_offset      the hash value, which is used to locate
+ *                             minum(closet) dir entry
+ * \param[in|out] stripe_offset        the caller use this to indicate the stripe
+ *                             index of last entry, so to avoid hash conflict
+ *                             between stripes. It will also be used to
+ *                             return the stripe index of current dir entry.
+ * \param[in|out] entp         the minum entry and it also is being used
+ *                             to input the last dir entry to resolve the
+ *                             hash conflict
+ *
+ * \param[out] ppage           the page which holds the minum entry
+ *
+ * \retval                     = 0 get the entry successfully
+ *                             negative errno (< 0) does not get the entry
+ */
+static int lmv_get_min_striped_entry(struct obd_export *exp,
+                                    struct md_op_data *op_data,
+                                    struct md_callback *cb_op,
+                                    __u64 hash_offset, int *stripe_offset,
+                                    struct lu_dirent **entp,
+                                    struct page **ppage)
+{
+       struct lmv_stripe_md *lsm = op_data->op_mea1;
+       struct obd_device *obd = exp->exp_obd;
+       struct lmv_obd *lmv = &obd->u.lmv;
+       struct lu_dirent *min_ent = NULL;
+       struct page *min_page = NULL;
+       struct lmv_tgt_desc *tgt;
+       int stripe_count;
+       int min_idx = 0;
+       int rc = 0;
+       int i;
+
+       stripe_count = lsm->lsm_md_stripe_count;
+       for (i = 0; i < stripe_count; i++) {
+               __u64 stripe_hash = hash_offset;
+               struct lu_dirent *ent = NULL;
+               struct page *page = NULL;
+               struct lu_dirpage *dp;
+
+               tgt = lmv_get_target(lmv, lsm->lsm_md_oinfo[i].lmo_mds, NULL);
+               if (IS_ERR(tgt)) {
+                       rc = PTR_ERR(tgt);
+                       goto out;
+               }
+
+               /*
+                * op_data will be shared by each stripe, so we need
+                * reset these value for each stripe
+                */
+               op_data->op_fid1 = lsm->lsm_md_oinfo[i].lmo_fid;
+               op_data->op_fid2 = lsm->lsm_md_oinfo[i].lmo_fid;
+               op_data->op_data = lsm->lsm_md_oinfo[i].lmo_root;
+next:
+               rc = md_read_page(tgt->ltd_exp, op_data, cb_op, stripe_hash,
+                                 &page);
+               if (rc)
+                       goto out;
+
+               dp = page_address(page);
+               for (ent = lu_dirent_start(dp); ent;
+                    ent = lu_dirent_next(ent)) {
+                       /* Skip dummy entry */
+                       if (!le16_to_cpu(ent->lde_namelen))
+                               continue;
+
+                       if (le64_to_cpu(ent->lde_hash) < hash_offset)
+                               continue;
+
+                       if (le64_to_cpu(ent->lde_hash) == hash_offset &&
+                           (*entp == ent || i < *stripe_offset))
+                               continue;
+
+                       /* skip . and .. for other stripes */
+                       if (i && (!strncmp(ent->lde_name, ".",
+                                          le16_to_cpu(ent->lde_namelen)) ||
+                                 !strncmp(ent->lde_name, "..",
+                                          le16_to_cpu(ent->lde_namelen))))
+                               continue;
+                       break;
+               }
+
+               if (!ent) {
+                       stripe_hash = le64_to_cpu(dp->ldp_hash_end);
+
+                       kunmap(page);
+                       put_page(page);
+                       page = NULL;
+
+                       /*
+                        * reach the end of current stripe, go to next stripe
+                        */
+                       if (stripe_hash == MDS_DIR_END_OFF)
+                               continue;
+                       else
+                               goto next;
+               }
+
+               if (min_ent) {
+                       if (le64_to_cpu(min_ent->lde_hash) >
+                           le64_to_cpu(ent->lde_hash)) {
+                               min_ent = ent;
+                               kunmap(min_page);
+                               put_page(min_page);
+                               min_idx = i;
+                               min_page = page;
+                       } else {
+                               kunmap(page);
+                               put_page(page);
+                               page = NULL;
+                       }
+               } else {
+                       min_ent = ent;
+                       min_page = page;
+                       min_idx = i;
+               }
+       }
+
+out:
+       if (*ppage) {
+               kunmap(*ppage);
+               put_page(*ppage);
+       }
+       *stripe_offset = min_idx;
+       *entp = min_ent;
+       *ppage = min_page;
+       return rc;
+}
+
+/**
+ * Build dir entry page from a striped directory
+ *
+ * This function gets one entry by @offset from a striped directory. It will
+ * read entries from all of stripes, and choose one closest to the required
+ * offset(&offset). A few notes
+ * 1. skip . and .. for non-zero stripes, because there can only have one .
+ * and .. in a directory.
+ * 2. op_data will be shared by all of stripes, instead of allocating new
+ * one, so need to restore before reusing.
+ * 3. release the entry page if that is not being chosen.
+ *
+ * \param[in] exp      obd export refer to LMV
+ * \param[in] op_data  hold those MD parameters of read_entry
+ * \param[in] cb_op    ldlm callback being used in enqueue in mdc_read_entry
+ * \param[out] ldp     the entry being read
+ * \param[out] ppage   the page holding the entry. Note: because the entry
+ *                     will be accessed in upper layer, so we need hold the
+ *                     page until the usages of entry is finished, see
+ *                     ll_dir_entry_next.
+ *
+ * retval              =0 if get entry successfully
+ *                     <0 cannot get entry
+ */
+static int lmv_read_striped_page(struct obd_export *exp,
+                                struct md_op_data *op_data,
+                                struct md_callback *cb_op,
+                                __u64 offset, struct page **ppage)
+{
+       struct inode *master_inode = op_data->op_data;
+       struct lu_fid master_fid = op_data->op_fid1;
+       struct obd_device *obd = exp->exp_obd;
+       __u64 hash_offset = offset;
+       struct page *min_ent_page = NULL;
+       struct page *ent_page = NULL;
+       struct lu_dirent *min_ent = NULL;
+       struct lu_dirent *last_ent;
+       struct lu_dirent *ent;
+       struct lu_dirpage *dp;
+       size_t left_bytes;
+       int ent_idx = 0;
+       void *area;
+       int rc;
+
+       rc = lmv_check_connect(obd);
+       if (rc)
+               return rc;
+
+       /*
+        * Allocate a page and read entries from all of stripes and fill
+        * the page by hash order
+        */
+       ent_page = alloc_page(GFP_KERNEL);
+       if (!ent_page)
+               return -ENOMEM;
+
+       /* Initialize the entry page */
+       dp = kmap(ent_page);
+       memset(dp, 0, sizeof(*dp));
+       dp->ldp_hash_start = cpu_to_le64(offset);
+       dp->ldp_flags |= LDF_COLLIDE;
+
+       area = dp + 1;
+       left_bytes = PAGE_SIZE - sizeof(*dp);
+       ent = area;
+       last_ent = ent;
+       do {
+               __u16 ent_size;
+
+               /* Find the minum entry from all sub-stripes */
+               rc = lmv_get_min_striped_entry(exp, op_data, cb_op, hash_offset,
+                                              &ent_idx, &min_ent,
+                                              &min_ent_page);
+               if (rc)
+                       goto out;
+
+               /*
+                * If it can not get minum entry, it means it already reaches
+                * the end of this directory
+                */
+               if (!min_ent) {
+                       last_ent->lde_reclen = 0;
+                       hash_offset = MDS_DIR_END_OFF;
+                       goto out;
+               }
+
+               ent_size = le16_to_cpu(min_ent->lde_reclen);
+
+               /*
+                * the last entry lde_reclen is 0, but it might not
+                * the end of this entry of this temporay entry
+                */
+               if (!ent_size)
+                       ent_size = lu_dirent_calc_size(
+                                       le16_to_cpu(min_ent->lde_namelen),
+                                       le32_to_cpu(min_ent->lde_attrs));
+               if (ent_size > left_bytes) {
+                       last_ent->lde_reclen = cpu_to_le16(0);
+                       hash_offset = le64_to_cpu(min_ent->lde_hash);
+                       goto out;
+               }
+
+               memcpy(ent, min_ent, ent_size);
+
+               /*
+                * Replace . with master FID and Replace .. with the parent FID
+                * of master object
+                */
+               if (!strncmp(ent->lde_name, ".",
+                            le16_to_cpu(ent->lde_namelen)) &&
+                   le16_to_cpu(ent->lde_namelen) == 1)
+                       fid_cpu_to_le(&ent->lde_fid, &master_fid);
+               else if (!strncmp(ent->lde_name, "..",
+                                 le16_to_cpu(ent->lde_namelen)) &&
+                        le16_to_cpu(ent->lde_namelen) == 2)
+                       fid_cpu_to_le(&ent->lde_fid, &op_data->op_fid3);
+
+               left_bytes -= ent_size;
+               ent->lde_reclen = cpu_to_le16(ent_size);
+               last_ent = ent;
+               ent = (void *)ent + ent_size;
+               hash_offset = le64_to_cpu(min_ent->lde_hash);
+               if (hash_offset == MDS_DIR_END_OFF) {
+                       last_ent->lde_reclen = 0;
+                       break;
+               }
+       } while (1);
+out:
+       if (min_ent_page) {
+               kunmap(min_ent_page);
+               put_page(min_ent_page);
+       }
+
+       if (unlikely(rc)) {
+               __free_page(ent_page);
+               ent_page = NULL;
+       } else {
+               if (ent == area)
+                       dp->ldp_flags |= LDF_EMPTY;
+               dp->ldp_flags = cpu_to_le32(dp->ldp_flags);
+               dp->ldp_hash_end = cpu_to_le64(hash_offset);
+       }
+
+       /*
+        * We do not want to allocate md_op_data during each
+        * dir entry reading, so op_data will be shared by every stripe,
+        * then we need to restore it back to original value before
+        * return to the upper layer
+        */
+       op_data->op_fid1 = master_fid;
+       op_data->op_fid2 = master_fid;
+       op_data->op_data = master_inode;
+
+       *ppage = ent_page;
+
+       return rc;
+}
+
+int lmv_read_page(struct obd_export *exp, struct md_op_data *op_data,
+                 struct md_callback *cb_op, __u64 offset,
+                 struct page **ppage)
+{
+       struct lmv_stripe_md *lsm = op_data->op_mea1;
+       struct obd_device *obd = exp->exp_obd;
+       struct lmv_obd *lmv = &obd->u.lmv;
+       struct lmv_tgt_desc *tgt;
+       int rc;
+
+       rc = lmv_check_connect(obd);
+       if (rc)
+               return rc;
+
+       if (unlikely(lsm)) {
+               rc = lmv_read_striped_page(exp, op_data, cb_op, offset, ppage);
+               return rc;
+       }
+
+       tgt = lmv_find_target(lmv, &op_data->op_fid1);
+       if (IS_ERR(tgt))
+               return PTR_ERR(tgt);
+
+       rc = md_read_page(tgt->ltd_exp, op_data, cb_op, offset, ppage);
+
+       return rc;
+}
+
 /**
  * Unlink a file/directory
  *
@@ -3268,6 +3596,7 @@ static struct md_ops lmv_md_ops = {
        .setxattr               = lmv_setxattr,
        .sync                   = lmv_sync,
        .readpage               = lmv_readpage,
+       .read_page              = lmv_read_page,
        .unlink                 = lmv_unlink,
        .init_ea_size           = lmv_init_ea_size,
        .cancel_unused          = lmv_cancel_unused,