X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Flov%2Flov_object.c;h=cc05543033726d070050569b66fa175709e7e832;hb=916b5f7f4672c1070e27a2fe0bfae371b0a729d6;hp=c852ab5c1152232d83d5b5dc42c02fb3d3cf72c2;hpb=f6995cf04407dff15d6ca79ca44cfa97dc6eb014;p=fs%2Flustre-release.git

diff --git a/lustre/lov/lov_object.c b/lustre/lov/lov_object.c
index c852ab5..cc05543 100644
--- a/lustre/lov/lov_object.c
+++ b/lustre/lov/lov_object.c
@@ -27,7 +27,7 @@
  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2011, 2014, Intel Corporation.
+ * Copyright (c) 2011, 2015, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -43,6 +43,11 @@
 
 #include "lov_cl_internal.h"
 
+static inline struct lov_device *lov_object_dev(struct lov_object *obj)
+{
+	return lu2lov_dev(obj->lo_cl.co_lu.lo_dev);
+}
+
 /** \addtogroup lov
  *  @{
  */
@@ -54,10 +59,10 @@
  */
 
 struct lov_layout_operations {
-        int (*llo_init)(const struct lu_env *env, struct lov_device *dev,
-                        struct lov_object *lov,
-                        const struct cl_object_conf *conf,
-                        union lov_layout_state *state);
+	int (*llo_init)(const struct lu_env *env, struct lov_device *dev,
+			struct lov_object *lov, struct lov_stripe_md *lsm,
+			const struct cl_object_conf *conf,
+			union lov_layout_state *state);
 	int (*llo_delete)(const struct lu_env *env, struct lov_object *lov,
                            union lov_layout_state *state);
         void (*llo_fini)(const struct lu_env *env, struct lov_object *lov,
@@ -75,8 +80,6 @@ struct lov_layout_operations {
                             struct cl_object *obj, struct cl_io *io);
         int  (*llo_getattr)(const struct lu_env *env, struct cl_object *obj,
                             struct cl_attr *attr);
-	int  (*llo_find_cbdata)(const struct lu_env *env, struct cl_object *obj,
-				ldlm_iterator_t iter, void *data);
 };
 
 static int lov_layout_wait(const struct lu_env *env, struct lov_object *lov);
@@ -102,12 +105,12 @@ static void lov_install_empty(const struct lu_env *env,
          */
 }
 
-static int lov_init_empty(const struct lu_env *env,
-                          struct lov_device *dev, struct lov_object *lov,
-                          const struct cl_object_conf *conf,
-                          union  lov_layout_state *state)
+static int lov_init_empty(const struct lu_env *env, struct lov_device *dev,
+			  struct lov_object *lov, struct lov_stripe_md *lsm,
+			  const struct cl_object_conf *conf,
+			  union lov_layout_state *state)
 {
-        return 0;
+	return 0;
 }
 
 static void lov_install_raid0(const struct lu_env *env,
@@ -192,8 +195,8 @@ static int lov_init_sub(const struct lu_env *env, struct lov_object *lov,
 		}
 
 		LU_OBJECT_DEBUG(mask, env, &stripe->co_lu,
-				"stripe %d is already owned.\n", idx);
-		LU_OBJECT_DEBUG(mask, env, old_obj, "owned.\n");
+				"stripe %d is already owned.", idx);
+		LU_OBJECT_DEBUG(mask, env, old_obj, "owned.");
 		LU_OBJECT_HEADER(mask, env, lov2lu(lov), "try to own.\n");
 		cl_object_put(env, stripe);
 	}
@@ -216,10 +219,10 @@ static int lov_page_slice_fixup(struct lov_object *lov,
 	return cl_object_header(stripe)->coh_page_bufsize;
 }
 
-static int lov_init_raid0(const struct lu_env *env,
-                          struct lov_device *dev, struct lov_object *lov,
-                          const struct cl_object_conf *conf,
-                          union  lov_layout_state *state)
+static int lov_init_raid0(const struct lu_env *env, struct lov_device *dev,
+			  struct lov_object *lov, struct lov_stripe_md *lsm,
+			  const struct cl_object_conf *conf,
+			  union lov_layout_state *state)
 {
         int result;
         int i;
@@ -227,7 +230,6 @@ static int lov_init_raid0(const struct lu_env *env,
         struct cl_object        *stripe;
         struct lov_thread_info  *lti     = lov_env_info(env);
         struct cl_object_conf   *subconf = &lti->lti_stripe_conf;
-        struct lov_stripe_md    *lsm     = conf->u.coc_md->lsm;
         struct lu_fid           *ofid    = &lti->lti_fid;
         struct lov_layout_raid0 *r0      = &state->raid0;
 
@@ -241,7 +243,7 @@ static int lov_init_raid0(const struct lu_env *env,
 
 	LASSERT(lov->lo_lsm == NULL);
 	lov->lo_lsm = lsm_addref(lsm);
-	r0->lo_nr  = lsm->lsm_stripe_count;
+	r0->lo_nr = lsm->lsm_stripe_count;
 	LASSERT(r0->lo_nr <= lov_targets_nr(dev));
 
 	lov->lo_layout_invalid = true;
@@ -269,6 +271,12 @@ static int lov_init_raid0(const struct lu_env *env,
 			if (result != 0)
 				GOTO(out, result);
 
+			if (dev->ld_target[ost_idx] == NULL) {
+				CERROR("%s: OST %04x is not initialized\n",
+				       lov2obd(dev->ld_lov)->obd_name, ost_idx);
+				GOTO(out, result = -EIO);
+			}
+
 			subdev = lovsub2cl_dev(dev->ld_target[ost_idx]);
 			subconf->u.coc_oinfo = oinfo;
 			LASSERTF(subdev != NULL, "not init ost %d\n", ost_idx);
@@ -302,12 +310,11 @@ out:
 }
 
 static int lov_init_released(const struct lu_env *env,
-			struct lov_device *dev, struct lov_object *lov,
-			const struct cl_object_conf *conf,
-			union  lov_layout_state *state)
+			     struct lov_device *dev, struct lov_object *lov,
+			     struct lov_stripe_md *lsm,
+			     const struct cl_object_conf *conf,
+			     union lov_layout_state *state)
 {
-	struct lov_stripe_md *lsm = conf->u.coc_md->lsm;
-
 	LASSERT(lsm != NULL);
 	LASSERT(lsm_is_released(lsm));
 	LASSERT(lov->lo_lsm == NULL);
@@ -590,37 +597,6 @@ static int lov_attr_get_raid0(const struct lu_env *env, struct cl_object *obj,
 	RETURN(result);
 }
 
-static int lov_find_cbdata_empty(const struct lu_env *env,
-				 struct cl_object *obj, ldlm_iterator_t iter,
-				 void *data)
-{
-	return 0;
-}
-
-static int lov_find_cbdata_raid0(const struct lu_env *env,
-				 struct cl_object *obj, ldlm_iterator_t iter,
-				 void *data)
-{
-	struct lov_object	*lov = cl2lov(obj);
-	struct lov_layout_raid0	*r0 = lov_r0(lov);
-	struct cl_object	*subobj;
-	int			i;
-	int			rc = 0;
-
-	for (i = 0; i < r0->lo_nr; ++i) {
-		if (r0->lo_sub[i] == NULL)
-			continue;
-
-		subobj = lovsub2cl(r0->lo_sub[i]);
-
-		rc = cl_object_find_cbdata(env, subobj, iter, data);
-		if (rc != 0)
-			break;
-	}
-
-	return rc;
-}
-
 const static struct lov_layout_operations lov_dispatch[] = {
         [LLT_EMPTY] = {
                 .llo_init      = lov_init_empty,
@@ -632,7 +608,6 @@ const static struct lov_layout_operations lov_dispatch[] = {
                 .llo_lock_init = lov_lock_init_empty,
                 .llo_io_init   = lov_io_init_empty,
 		.llo_getattr   = lov_attr_get_empty,
-		.llo_find_cbdata = lov_find_cbdata_empty
         },
         [LLT_RAID0] = {
                 .llo_init      = lov_init_raid0,
@@ -644,7 +619,6 @@ const static struct lov_layout_operations lov_dispatch[] = {
                 .llo_lock_init = lov_lock_init_raid0,
                 .llo_io_init   = lov_io_init_raid0,
 		.llo_getattr   = lov_attr_get_raid0,
-		.llo_find_cbdata = lov_find_cbdata_raid0
 	},
         [LLT_RELEASED] = {
                 .llo_init      = lov_init_released,
@@ -656,7 +630,6 @@ const static struct lov_layout_operations lov_dispatch[] = {
                 .llo_lock_init = lov_lock_init_empty,
                 .llo_io_init   = lov_io_init_released,
 		.llo_getattr   = lov_attr_get_empty,
-		.llo_find_cbdata = lov_find_cbdata_empty
         }
 };
 
@@ -687,12 +660,16 @@ static enum lov_layout_type lov_type(struct lov_stripe_md *lsm)
 
 static inline void lov_conf_freeze(struct lov_object *lov)
 {
+	CDEBUG(D_INODE, "To take share lov(%p) owner %p/%p\n",
+		lov, lov->lo_owner, current);
 	if (lov->lo_owner != current)
 		down_read(&lov->lo_type_guard);
 }
 
 static inline void lov_conf_thaw(struct lov_object *lov)
 {
+	CDEBUG(D_INODE, "To release share lov(%p) owner %p/%p\n",
+		lov, lov->lo_owner, current);
 	if (lov->lo_owner != current)
 		up_read(&lov->lo_type_guard);
 }
@@ -735,10 +712,14 @@ static void lov_conf_lock(struct lov_object *lov)
 	down_write(&lov->lo_type_guard);
 	LASSERT(lov->lo_owner == NULL);
 	lov->lo_owner = current;
+	CDEBUG(D_INODE, "Took exclusive lov(%p) owner %p\n",
+		lov, lov->lo_owner);
 }
 
 static void lov_conf_unlock(struct lov_object *lov)
 {
+	CDEBUG(D_INODE, "To release exclusive lov(%p) owner %p\n",
+		lov, lov->lo_owner);
 	lov->lo_owner = NULL;
 	up_write(&lov->lo_type_guard);
 }
@@ -760,32 +741,26 @@ static int lov_layout_wait(const struct lu_env *env, struct lov_object *lov)
 }
 
 static int lov_layout_change(const struct lu_env *unused,
-			     struct lov_object *lov,
+			     struct lov_object *lov, struct lov_stripe_md *lsm,
 			     const struct cl_object_conf *conf)
 {
-	int result;
-	enum lov_layout_type llt = LLT_EMPTY;
+	enum lov_layout_type llt = lov_type(lsm);
 	union lov_layout_state *state = &lov->u;
 	const struct lov_layout_operations *old_ops;
 	const struct lov_layout_operations *new_ops;
-
-	void *cookie;
+	struct lov_device *lov_dev = lov_object_dev(lov);
 	struct lu_env *env;
-	int refcheck;
+	__u16 refcheck;
+	int rc;
 	ENTRY;
 
 	LASSERT(0 <= lov->lo_type && lov->lo_type < ARRAY_SIZE(lov_dispatch));
 
-	if (conf->u.coc_md != NULL)
-		llt = lov_type(conf->u.coc_md->lsm);
-	LASSERT(0 <= llt && llt < ARRAY_SIZE(lov_dispatch));
-
-	cookie = cl_env_reenter();
 	env = cl_env_get(&refcheck);
-	if (IS_ERR(env)) {
-		cl_env_reexit(cookie);
+	if (IS_ERR(env))
 		RETURN(PTR_ERR(env));
-	}
+
+	LASSERT(0 <= llt && llt < ARRAY_SIZE(lov_dispatch));
 
 	CDEBUG(D_INODE, DFID" from %s to %s\n",
 	       PFID(lu_object_fid(lov2lu(lov))),
@@ -794,38 +769,45 @@ static int lov_layout_change(const struct lu_env *unused,
 	old_ops = &lov_dispatch[lov->lo_type];
 	new_ops = &lov_dispatch[llt];
 
-	result = cl_object_prune(env, &lov->lo_cl);
-	if (result != 0)
-		GOTO(out, result);
+	rc = cl_object_prune(env, &lov->lo_cl);
+	if (rc != 0)
+		GOTO(out, rc);
 
-	result = old_ops->llo_delete(env, lov, &lov->u);
-	if (result == 0) {
-		old_ops->llo_fini(env, lov, &lov->u);
+	rc = old_ops->llo_delete(env, lov, &lov->u);
+	if (rc != 0)
+		GOTO(out, rc);
 
-		LASSERT(atomic_read(&lov->lo_active_ios) == 0);
+	old_ops->llo_fini(env, lov, &lov->u);
 
-		lov->lo_type = LLT_EMPTY;
-		/* page bufsize fixup */
-		cl_object_header(&lov->lo_cl)->coh_page_bufsize -=
-			lov_page_slice_fixup(lov, NULL);
+	LASSERT(atomic_read(&lov->lo_active_ios) == 0);
 
-		result = new_ops->llo_init(env,
-					lu2lov_dev(lov->lo_cl.co_lu.lo_dev),
-					lov, conf, state);
-		if (result == 0) {
-			new_ops->llo_install(env, lov, state);
-			lov->lo_type = llt;
-		} else {
-			new_ops->llo_delete(env, lov, state);
-			new_ops->llo_fini(env, lov, state);
-			/* this file becomes an EMPTY file. */
-		}
+	CDEBUG(D_INODE, DFID "Apply new layout lov %p, type %d\n",
+	       PFID(lu_object_fid(lov2lu(lov))), lov, llt);
+
+	lov->lo_type = LLT_EMPTY;
+
+	/* page bufsize fixup */
+	cl_object_header(&lov->lo_cl)->coh_page_bufsize -=
+		lov_page_slice_fixup(lov, NULL);
+
+	rc = new_ops->llo_init(env, lov_dev, lov, lsm, conf, state);
+	if (rc != 0) {
+		struct obd_device *obd = lov2obd(lov_dev->ld_lov);
+
+		CERROR("%s: cannot apply new layout on "DFID" : rc = %d\n",
+		       obd->obd_name, PFID(lu_object_fid(lov2lu(lov))), rc);
+		new_ops->llo_delete(env, lov, state);
+		new_ops->llo_fini(env, lov, state);
+		/* this file becomes an EMPTY file. */
+		GOTO(out, rc);
 	}
 
+	new_ops->llo_install(env, lov, state);
+	lov->lo_type = llt;
+
 out:
 	cl_env_put(env, &refcheck);
-	cl_env_reexit(cookie);
-	RETURN(result);
+	RETURN(rc);
 }
 
 /*****************************************************************************
@@ -834,29 +816,44 @@ out:
  *
  */
 int lov_object_init(const struct lu_env *env, struct lu_object *obj,
-                    const struct lu_object_conf *conf)
+		    const struct lu_object_conf *conf)
 {
-        struct lov_device            *dev   = lu2lov_dev(obj->lo_dev);
-        struct lov_object            *lov   = lu2lov(obj);
-        const struct cl_object_conf  *cconf = lu2cl_conf(conf);
-        union  lov_layout_state      *set   = &lov->u;
-        const struct lov_layout_operations *ops;
-        int result;
+	struct lov_object            *lov   = lu2lov(obj);
+	struct lov_device            *dev   = lov_object_dev(lov);
+	const struct cl_object_conf  *cconf = lu2cl_conf(conf);
+	union lov_layout_state	     *set   = &lov->u;
+	const struct lov_layout_operations *ops;
+	struct lov_stripe_md *lsm = NULL;
+	int rc;
+	ENTRY;
 
-        ENTRY;
 	init_rwsem(&lov->lo_type_guard);
 	atomic_set(&lov->lo_active_ios, 0);
 	init_waitqueue_head(&lov->lo_waitq);
-
 	cl_object_page_init(lu2cl(obj), sizeof(struct lov_page));
 
-        /* no locking is necessary, as object is being created */
-	lov->lo_type = lov_type(cconf->u.coc_md->lsm);
-        ops = &lov_dispatch[lov->lo_type];
-        result = ops->llo_init(env, dev, lov, cconf, set);
-        if (result == 0)
-                ops->llo_install(env, lov, set);
-        RETURN(result);
+	lov->lo_type = LLT_EMPTY;
+	if (cconf->u.coc_layout.lb_buf != NULL) {
+		lsm = lov_unpackmd(dev->ld_lov,
+				   cconf->u.coc_layout.lb_buf,
+				   cconf->u.coc_layout.lb_len);
+		if (IS_ERR(lsm))
+			RETURN(PTR_ERR(lsm));
+	}
+
+	/* no locking is necessary, as object is being created */
+	lov->lo_type = lov_type(lsm);
+	ops = &lov_dispatch[lov->lo_type];
+	rc = ops->llo_init(env, dev, lov, lsm, cconf, set);
+	if (rc != 0)
+		GOTO(out_lsm, rc);
+
+	ops->llo_install(env, lov, set);
+
+out_lsm:
+	lov_lsm_put(lsm);
+
+	RETURN(rc);
 }
 
 static int lov_conf_set(const struct lu_env *env, struct cl_object *obj,
@@ -867,6 +864,15 @@ static int lov_conf_set(const struct lu_env *env, struct cl_object *obj,
 	int			 result = 0;
 	ENTRY;
 
+	if (conf->coc_opc == OBJECT_CONF_SET &&
+	    conf->u.coc_layout.lb_buf != NULL) {
+		lsm = lov_unpackmd(lov_object_dev(lov)->ld_lov,
+				   conf->u.coc_layout.lb_buf,
+				   conf->u.coc_layout.lb_len);
+		if (IS_ERR(lsm))
+			RETURN(PTR_ERR(lsm));
+	}
+
 	lov_conf_lock(lov);
 	if (conf->coc_opc == OBJECT_CONF_INVALIDATE) {
 		lov->lo_layout_invalid = true;
@@ -885,8 +891,6 @@ static int lov_conf_set(const struct lu_env *env, struct cl_object *obj,
 
 	LASSERT(conf->coc_opc == OBJECT_CONF_SET);
 
-	if (conf->u.coc_md != NULL)
-		lsm = conf->u.coc_md->lsm;
 	if ((lsm == NULL && lov->lo_lsm == NULL) ||
 	    ((lsm != NULL && lov->lo_lsm != NULL) &&
 	     (lov->lo_lsm->lsm_layout_gen == lsm->lsm_layout_gen) &&
@@ -902,12 +906,13 @@ static int lov_conf_set(const struct lu_env *env, struct cl_object *obj,
 		GOTO(out, result = -EBUSY);
 	}
 
-	result = lov_layout_change(env, lov, conf);
+	result = lov_layout_change(env, lov, lsm, conf);
 	lov->lo_layout_invalid = result != 0;
 	EXIT;
 
 out:
 	lov_conf_unlock(lov);
+	lov_lsm_put(lsm);
 	CDEBUG(D_INODE, DFID" lo_layout_invalid=%d\n",
 	       PFID(lu_object_fid(lov2lu(lov))), lov->lo_layout_invalid);
 	RETURN(result);
@@ -954,6 +959,11 @@ int lov_io_init(const struct lu_env *env, struct cl_object *obj,
 		struct cl_io *io)
 {
 	CL_IO_SLICE_CLEAN(lov_env_io(env), lis_cl);
+
+	CDEBUG(D_INODE, DFID "io %p type %d ignore/verify layout %d/%d\n",
+	       PFID(lu_object_fid(&obj->co_lu)), io, io->ci_type,
+	       io->ci_ignore_layout, io->ci_verify_layout);
+
 	return LOV_2DISPATCH_MAYLOCK(cl2lov(obj), llo_io_init,
 				     !io->ci_ignore_layout, env, obj, io);
 }
@@ -1043,7 +1053,7 @@ static int fiemap_calc_last_stripe(struct lov_stripe_md *lsm,
  * \param current_extent [in]	where to start copying in the extent array
  */
 static void fiemap_prepare_and_copy_exts(struct fiemap *fiemap,
-					 struct ll_fiemap_extent *lcl_fm_ext,
+					 struct fiemap_extent *lcl_fm_ext,
 					 int ost_index, unsigned int ext_count,
 					 int current_extent)
 {
@@ -1057,7 +1067,7 @@ static void fiemap_prepare_and_copy_exts(struct fiemap *fiemap,
 
 	/* Copy fm_extent's from fm_local to return buffer */
 	to = (char *)fiemap + fiemap_count_to_size(current_extent);
-	memcpy(to, lcl_fm_ext, ext_count * sizeof(struct ll_fiemap_extent));
+	memcpy(to, lcl_fm_ext, ext_count * sizeof(struct fiemap_extent));
 }
 
 #define FIEMAP_BUFFER_SIZE 4096
@@ -1150,7 +1160,7 @@ static int lov_object_fiemap(const struct lu_env *env, struct cl_object *obj,
 	struct cl_object	*subobj = NULL;
 	struct lov_obd		*lov = lu2lov_dev(obj->co_lu.lo_dev)->ld_lov;
 	struct fiemap		*fm_local = NULL;
-	struct ll_fiemap_extent	*lcl_fm_ext;
+	struct fiemap_extent	*lcl_fm_ext;
 	loff_t			fm_start;
 	loff_t			fm_end;
 	loff_t			fm_length;
@@ -1183,10 +1193,10 @@ static int lov_object_fiemap(const struct lu_env *env, struct cl_object *obj,
 	 */
 	if (lsm->lsm_stripe_count > 1 && !(fiemap->fm_flags &
 					   FIEMAP_FLAG_DEVICE_ORDER))
-		GOTO(out, rc = -ENOTSUPP);
+		GOTO(out_lsm, rc = -ENOTSUPP);
 
 	if (lsm_is_released(lsm)) {
-		if (fiemap->fm_start < fmkey->oa.o_size) {
+		if (fiemap->fm_start < fmkey->lfik_oa.o_size) {
 			/**
 			 * released file, return a minimal FIEMAP if
 			 * request fits in file-size.
@@ -1194,16 +1204,17 @@ static int lov_object_fiemap(const struct lu_env *env, struct cl_object *obj,
 			fiemap->fm_mapped_extents = 1;
 			fiemap->fm_extents[0].fe_logical = fiemap->fm_start;
 			if (fiemap->fm_start + fiemap->fm_length <
-			    fmkey->oa.o_size)
+			    fmkey->lfik_oa.o_size)
 				fiemap->fm_extents[0].fe_length =
 					fiemap->fm_length;
 			else
 				fiemap->fm_extents[0].fe_length =
-					fmkey->oa.o_size - fiemap->fm_start;
+					fmkey->lfik_oa.o_size -
+					fiemap->fm_start;
 			fiemap->fm_extents[0].fe_flags |=
 				FIEMAP_EXTENT_UNKNOWN | FIEMAP_EXTENT_LAST;
 		}
-		GOTO(out, rc = 0);
+		GOTO(out_lsm, rc = 0);
 	}
 
 	if (fiemap_count_to_size(fiemap->fm_extent_count) < buffer_size)
@@ -1211,7 +1222,7 @@ static int lov_object_fiemap(const struct lu_env *env, struct cl_object *obj,
 
 	OBD_ALLOC_LARGE(fm_local, buffer_size);
 	if (fm_local == NULL)
-		GOTO(out, rc = -ENOMEM);
+		GOTO(out_lsm, rc = -ENOMEM);
 	lcl_fm_ext = &fm_local->fm_extents[0];
 	count_local = fiemap_size_to_count(buffer_size);
 
@@ -1219,18 +1230,18 @@ static int lov_object_fiemap(const struct lu_env *env, struct cl_object *obj,
 	fm_length = fiemap->fm_length;
 	/* Calculate start stripe, last stripe and length of mapping */
 	start_stripe = lov_stripe_number(lsm, fm_start);
-	fm_end = (fm_length == ~0ULL) ? fmkey->oa.o_size :
+	fm_end = (fm_length == ~0ULL) ? fmkey->lfik_oa.o_size :
 					fm_start + fm_length - 1;
 	/* If fm_length != ~0ULL but fm_start_fm_length-1 exceeds file size */
-	if (fm_end > fmkey->oa.o_size)
-		fm_end = fmkey->oa.o_size;
+	if (fm_end > fmkey->lfik_oa.o_size)
+		fm_end = fmkey->lfik_oa.o_size;
 
 	last_stripe = fiemap_calc_last_stripe(lsm, fm_start, fm_end,
 					      start_stripe, &stripe_count);
 	fm_end_offset = fiemap_calc_fm_end_offset(fiemap, lsm, fm_start, fm_end,
 						  &start_stripe);
 	if (fm_end_offset == -EINVAL)
-		GOTO(out, rc = -EINVAL);
+		GOTO(out_fm_local, rc = -EINVAL);
 
 	/**
 	 * Requested extent count exceeds the fiemap buffer size, shrink our
@@ -1260,7 +1271,7 @@ static int lov_object_fiemap(const struct lu_env *env, struct cl_object *obj,
 			continue;
 
 		if (lov_oinfo_is_dummy(lsm->lsm_oinfo[cur_stripe]))
-			GOTO(out, rc = -EIO);
+			GOTO(out_fm_local, rc = -EIO);
 
 		/* If this is a continuation FIEMAP call and we are on
 		 * starting stripe then lun_start needs to be set to
@@ -1289,7 +1300,7 @@ static int lov_object_fiemap(const struct lu_env *env, struct cl_object *obj,
 		subobj = lov_find_subobj(env, cl2lov(obj), lsm,
 					     cur_stripe);
 		if (IS_ERR(subobj))
-			GOTO(out, rc = PTR_ERR(subobj));
+			GOTO(out_fm_local, rc = PTR_ERR(subobj));
 		/* If the output buffer is very large and the objects have many
 		 * extents we may need to loop on a single OST repeatedly */
 		ost_eof = false;
@@ -1332,7 +1343,8 @@ static int lov_object_fiemap(const struct lu_env *env, struct cl_object *obj,
 
 			fm_local->fm_start = lun_start;
 			fm_local->fm_flags &= ~FIEMAP_FLAG_DEVICE_ORDER;
-			memcpy(&fmkey->fiemap, fm_local, sizeof(*fm_local));
+			memcpy(&fmkey->lfik_fiemap, fm_local,
+			       sizeof(*fm_local));
 			*buflen = fiemap_count_to_size(
 						fm_local->fm_extent_count);
 
@@ -1383,7 +1395,7 @@ inactive_tgt:
 			if (lov_stripe_size(lsm,
 					lcl_fm_ext[ext_count - 1].fe_logical +
 					lcl_fm_ext[ext_count - 1].fe_length,
-					cur_stripe) >= fmkey->oa.o_size)
+					cur_stripe) >= fmkey->lfik_oa.o_size)
 				ost_eof = true;
 
 			fiemap_prepare_and_copy_exts(fiemap, lcl_fm_ext,
@@ -1421,139 +1433,15 @@ skip_last_device_calc:
 obj_put:
 	if (subobj != NULL)
 		cl_object_put(env, subobj);
-out:
-	if (fm_local != NULL)
-		OBD_FREE_LARGE(fm_local, buffer_size);
+out_fm_local:
+	OBD_FREE_LARGE(fm_local, buffer_size);
 
+out_lsm:
 	lov_lsm_put(lsm);
 
 	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(lsm);
-	RETURN(rc);
-}
-
 static int lov_object_getstripe(const struct lu_env *env, struct cl_object *obj,
 				struct lov_user_md __user *lum)
 {
@@ -1584,14 +1472,12 @@ static int lov_object_layout_get(const struct lu_env *env,
 	if (lsm == NULL) {
 		cl->cl_size = 0;
 		cl->cl_layout_gen = CL_LAYOUT_GEN_EMPTY;
-		cl->cl_is_released = false;
 
 		RETURN(0);
 	}
 
 	cl->cl_size = lov_mds_md_size(lsm->lsm_stripe_count, lsm->lsm_magic);
 	cl->cl_layout_gen = lsm->lsm_layout_gen;
-	cl->cl_is_released = lsm_is_released(lsm);
 
 	rc = lov_lsm_pack(lsm, buf->lb_buf, buf->lb_len);
 	lov_lsm_put(lsm);
@@ -1599,17 +1485,20 @@ static int lov_object_layout_get(const struct lu_env *env,
 	RETURN(rc < 0 ? rc : 0);
 }
 
-static int lov_object_find_cbdata(const struct lu_env *env,
-				  struct cl_object *obj, ldlm_iterator_t iter,
-				  void *data)
+static loff_t lov_object_maxbytes(struct cl_object *obj)
 {
-	int rc;
-	ENTRY;
+	struct lov_object *lov = cl2lov(obj);
+	struct lov_stripe_md *lsm = lov_lsm_addref(lov);
+	loff_t maxbytes;
 
-	/* call cl_object_find_cbdata for sub obj */
-	rc = LOV_2DISPATCH_NOLOCK(cl2lov(obj), llo_find_cbdata,
-				  env, obj, iter, data);
-	RETURN(rc);
+	if (lsm == NULL)
+		return LLONG_MAX;
+
+	maxbytes = lsm->lsm_maxbytes;
+
+	lov_lsm_put(lsm);
+
+	return maxbytes;
 }
 
 static const struct cl_object_operations lov_ops = {
@@ -1621,9 +1510,8 @@ static const struct cl_object_operations lov_ops = {
 	.coo_conf_set     = lov_conf_set,
 	.coo_getstripe    = lov_object_getstripe,
 	.coo_layout_get   = lov_object_layout_get,
-	.coo_find_cbdata  = lov_object_find_cbdata,
+	.coo_maxbytes     = lov_object_maxbytes,
 	.coo_fiemap       = lov_object_fiemap,
-	.coo_data_version = lov_object_data_version,
 };
 
 static const struct lu_object_operations lov_lu_obj_ops = {