LU-5823 clio: add coo_obd_info_get and coo_data_version

[fs/lustre-release.git] / lustre / lov / lov_object.c

diff --git a/lustre/lov/lov_object.c b/lustre/lov/lov_object.c
index 8feeee2..afe3b9f 100644 (file)
--- a/lustre/lov/lov_object.c
+++ b/lustre/lov/lov_object.c
@@ -1445,6 +1445,130 @@ out:
        RETURN(rc);
 }
 
+static int lov_dispatch_obd_info_get(const struct lu_env *env,
+                                    struct cl_object *obj,
+                                    struct obd_info *oinfo,
+                                    struct ptlrpc_request_set *set)
+{
+       struct cl_object        *subobj = NULL;
+       struct lov_obd          *lov = lu2lov_dev(obj->co_lu.lo_dev)->ld_lov;
+       struct lov_request_set  *lovset;
+       struct list_head        *pos;
+       struct lov_request      *req;
+       int                     rc;
+       int                     rc2;
+       ENTRY;
+
+       rc = lov_prep_getattr_set(lov2obd(lov)->obd_self_export, oinfo,
+                                 &lovset);
+       if (rc != 0)
+               RETURN(rc);
+
+       CDEBUG(D_INFO, "objid "DOSTID": %ux%u byte stripes.\n",
+              POSTID(&oinfo->oi_md->lsm_oi),
+              oinfo->oi_md->lsm_stripe_count,
+              oinfo->oi_md->lsm_stripe_size);
+
+       list_for_each(pos, &lovset->set_list) {
+               req = list_entry(pos, struct lov_request, rq_link);
+
+               CDEBUG(D_INFO, "objid "DOSTID"[%d] has subobj "DOSTID" at idx"
+                      "%u\n", POSTID(&oinfo->oi_oa->o_oi), req->rq_stripe,
+                      POSTID(&req->rq_oi.oi_oa->o_oi), req->rq_idx);
+               subobj = lov_find_subobj(env, cl2lov(obj), oinfo->oi_md,
+                                        req->rq_stripe);
+               if (IS_ERR(subobj))
+                       GOTO(errout, rc = PTR_ERR(subobj));
+
+               rc = cl_object_obd_info_get(env, subobj, &req->rq_oi, set);
+               cl_object_put(env, subobj);
+               if (rc != 0) {
+                       CERROR("%s: getattr objid "DOSTID" subobj"
+                              DOSTID" on OST idx %d: rc = %d\n",
+                              lov2obd(lov)->obd_name,
+                              POSTID(&oinfo->oi_oa->o_oi),
+                              POSTID(&req->rq_oi.oi_oa->o_oi),
+                              req->rq_idx, rc);
+                       GOTO(errout, rc);
+               }
+       }
+
+       if (!list_empty(&set->set_requests)) {
+               LASSERT(rc == 0);
+               LASSERT(set->set_interpret == NULL);
+               set->set_interpret = lov_getattr_interpret;
+               set->set_arg = lovset;
+               GOTO(out, rc);
+       }
+errout:
+       if (rc)
+               atomic_set(&lovset->set_completes, 0);
+       rc2 = lov_fini_getattr_set(lovset);
+       rc = rc != 0 ? rc : rc2;
+out:
+       RETURN(rc);
+}
+
+static int lov_object_data_version(const struct lu_env *env,
+                                  struct cl_object *obj, __u64 *data_version,
+                                  int flags)
+{
+       struct ptlrpc_request_set       *set;
+       struct obd_info                 oinfo =  { { { 0 } } };
+       struct obdo                     *obdo = NULL;
+       struct lov_stripe_md            *lsm;
+       int                             rc;
+       ENTRY;
+
+       lsm = lov_lsm_addref(cl2lov(obj));
+       if (!lsm_has_objects(lsm)) {
+               /* If no stripe, we consider version is 0. */
+               *data_version = 0;
+               GOTO(out, rc = 0);
+       }
+
+       OBD_ALLOC_PTR(obdo);
+       if (obdo == NULL)
+               GOTO(out, rc = -ENOMEM);
+
+       oinfo.oi_md = lsm;
+       oinfo.oi_oa = obdo;
+       obdo->o_oi = lsm->lsm_oi;
+       obdo->o_mode = S_IFREG;
+       obdo->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP | OBD_MD_FLTYPE;
+       if (flags & (LL_DV_RD_FLUSH | LL_DV_WR_FLUSH)) {
+               obdo->o_valid |= OBD_MD_FLFLAGS;
+               obdo->o_flags |= OBD_FL_SRVLOCK;
+               if (flags & LL_DV_WR_FLUSH)
+                       obdo->o_flags |= OBD_FL_FLUSH;
+       }
+
+       set = ptlrpc_prep_set();
+       if (set == NULL)
+               GOTO(out_obdo, rc = -ENOMEM);
+
+       rc = lov_dispatch_obd_info_get(env, obj, &oinfo, set);
+       if (rc == 0)
+               rc = ptlrpc_set_wait(set);
+       ptlrpc_set_destroy(set);
+       if (rc == 0) {
+               oinfo.oi_oa->o_valid &= OBD_MD_FLDATAVERSION | OBD_MD_FLFLAGS;
+               if (flags & LL_DV_WR_FLUSH &&
+                   !(oinfo.oi_oa->o_valid & OBD_MD_FLFLAGS &&
+                     oinfo.oi_oa->o_flags & OBD_FL_FLUSH))
+                       rc = -EOPNOTSUPP;
+               else if (!(obdo->o_valid & OBD_MD_FLDATAVERSION))
+                       rc = -EOPNOTSUPP;
+               else
+                       *data_version = obdo->o_data_version;
+       }
+out_obdo:
+       OBD_FREE_PTR(obdo);
+out:
+       lov_lsm_put(obj, lsm);
+       RETURN(rc);
+}
+
 static int lov_object_getstripe(const struct lu_env *env, struct cl_object *obj,
                                struct lov_user_md __user *lum)
 {
@@ -1476,15 +1600,16 @@ static int lov_object_find_cbdata(const struct lu_env *env,
 }
 
 static const struct cl_object_operations lov_ops = {
-       .coo_page_init   = lov_page_init,
-       .coo_lock_init   = lov_lock_init,
-       .coo_io_init     = lov_io_init,
-       .coo_attr_get    = lov_attr_get,
-       .coo_attr_update = lov_attr_update,
-       .coo_conf_set    = lov_conf_set,
-       .coo_getstripe   = lov_object_getstripe,
-       .coo_find_cbdata = lov_object_find_cbdata,
-       .coo_fiemap      = lov_object_fiemap,
+       .coo_page_init    = lov_page_init,
+       .coo_lock_init    = lov_lock_init,
+       .coo_io_init      = lov_io_init,
+       .coo_attr_get     = lov_attr_get,
+       .coo_attr_update  = lov_attr_update,
+       .coo_conf_set     = lov_conf_set,
+       .coo_getstripe    = lov_object_getstripe,
+       .coo_find_cbdata  = lov_object_find_cbdata,
+       .coo_fiemap       = lov_object_fiemap,
+       .coo_data_version = lov_object_data_version,
 };
 
 static const struct lu_object_operations lov_lu_obj_ops = {