X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fosp%2Fosp_object.c;h=2898da6f257cd7529be5f3fbfc3a476cd671375c;hp=c9841518da4dbbbd6f530d768b307ad73e7e6c36;hb=de8572645d287d17c409b99dabdf176822d91486;hpb=3a36f39a86d6b9984edbeb08dbd74bebc3b579ee

diff --git a/lustre/osp/osp_object.c b/lustre/osp/osp_object.c
index c984151..2898da6 100644
--- a/lustre/osp/osp_object.c
+++ b/lustre/osp/osp_object.c
@@ -575,7 +575,7 @@ int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
 		GOTO(out, rc);
 	}
 
-	rc = osp_remote_sync(env, osp, update, &req, false);
+	rc = osp_remote_sync(env, osp, update, &req);
 	if (rc != 0) {
 		if (rc == -ENOENT) {
 			osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
@@ -612,15 +612,29 @@ out:
 	return rc;
 }
 
-static int __osp_attr_set(const struct lu_env *env, struct dt_object *dt,
-			  const struct lu_attr *attr, struct thandle *th)
+/**
+ * Implement OSP layer dt_object_operations::do_declare_attr_set() interface.
+ *
+ * If the transaction is not remote one, then declare the credits that will
+ * be used for the subsequent llog record for the object's attributes.
+ *
+ * \param[in] env	pointer to the thread context
+ * \param[in] dt	pointer to the OSP layer dt_object
+ * \param[in] attr	pointer to the attribute to be set
+ * \param[in] th	pointer to the transaction handler
+ *
+ * \retval		0 for success
+ * \retval		negative error number on failure
+ */
+static int osp_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
+				const struct lu_attr *attr, struct thandle *th)
 {
 	struct osp_device	*d = lu2osp_dev(dt->do_lu.lo_dev);
 	struct osp_object	*o = dt2osp_obj(dt);
-	struct lu_attr		*la;
-	int			 rc = 0;
-	ENTRY;
+	int			rc;
 
+	if (is_only_remote_trans(th))
+		return osp_md_declare_attr_set(env, dt, attr, th);
 	/*
 	 * Usually we don't allow server stack to manipulate size
 	 * but there is a special case when striping is created
@@ -649,71 +663,17 @@ static int __osp_attr_set(const struct lu_env *env, struct dt_object *dt,
 		LASSERT(!dt_object_exists(dt));
 		osp_object_assign_fid(env, d, o);
 		rc = osp_object_truncate(env, dt, attr->la_size);
-		if (rc)
+		if (rc != 0)
 			RETURN(rc);
 	}
 
 	if (!(attr->la_valid & (LA_UID | LA_GID)))
 		RETURN(0);
 
-	if (!is_only_remote_trans(th)) {
-		/*
-		 * track all UID/GID changes via llog
-		 */
-		rc = osp_sync_declare_add(env, o, MDS_SETATTR64_REC, th);
-	} else {
-		/* It is for OST-object attr_set directly without updating
-		 * local MDT-object attribute. It is usually used by LFSCK. */
-		rc = __osp_md_attr_set(env, dt, attr, th);
-	}
-
-	if (rc != 0 || o->opo_ooa == NULL)
-		RETURN(rc);
-
-	/* Update the OSP object attributes cache. */
-	la = &o->opo_ooa->ooa_attr;
-	spin_lock(&o->opo_lock);
-	if (attr->la_valid & LA_UID) {
-		la->la_uid = attr->la_uid;
-		la->la_valid |= LA_UID;
-	}
-
-	if (attr->la_valid & LA_GID) {
-		la->la_gid = attr->la_gid;
-		la->la_valid |= LA_GID;
-	}
-	spin_unlock(&o->opo_lock);
-
-	RETURN(0);
-}
-
-/**
- * Implement OSP layer dt_object_operations::do_declare_attr_set() interface.
- *
- * If the transaction is not remote one, then declare the credits that will
- * be used for the subsequent llog record for the object's attributes.
- *
- * \param[in] env	pointer to the thread context
- * \param[in] dt	pointer to the OSP layer dt_object
- * \param[in] attr	pointer to the attribute to be set
- * \param[in] th	pointer to the transaction handler
- *
- * \retval		0 for success
- * \retval		negative error number on failure
- */
-static int osp_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
-				const struct lu_attr *attr, struct thandle *th)
-{
-	int rc = 0;
-
-	if (!is_only_remote_trans(th)) {
-		rc = __osp_attr_set(env, dt, attr, th);
-
-		CDEBUG(D_INFO, "declare set attr "DFID": rc = %d\n",
-		       PFID(&dt->do_lu.lo_header->loh_fid), rc);
-	}
+	/* track all UID/GID changes via llog */
+	rc = osp_sync_declare_add(env, o, MDS_SETATTR64_REC, th);
 
-	return rc;
+	return 0;
 }
 
 /**
@@ -744,24 +704,39 @@ static int osp_attr_set(const struct lu_env *env, struct dt_object *dt,
 	int			 rc = 0;
 	ENTRY;
 
-	if (is_only_remote_trans(th)) {
-		rc = __osp_attr_set(env, dt, attr, th);
+	/* we're interested in uid/gid changes only */
+	if (!(attr->la_valid & (LA_UID | LA_GID)))
+		RETURN(0);
+
+	if (!is_only_remote_trans(th)) {
+		rc = osp_sync_add(env, o, MDS_SETATTR64_REC, th, attr);
+		/* XXX: send new uid/gid to OST ASAP? */
+	} else {
+		struct lu_attr	*la;
 
+		/* It is for OST-object attr_set directly without updating
+		 * local MDT-object attribute. It is usually used by LFSCK. */
+		rc = osp_md_attr_set(env, dt, attr, th);
 		CDEBUG(D_INFO, "(1) set attr "DFID": rc = %d\n",
 		       PFID(&dt->do_lu.lo_header->loh_fid), rc);
 
-		RETURN(rc);
-	}
-
-	/* we're interested in uid/gid changes only */
-	if (!(attr->la_valid & (LA_UID | LA_GID)))
-		RETURN(0);
+		if (rc != 0 || o->opo_ooa == NULL)
+			RETURN(rc);
 
-	rc = osp_sync_add(env, o, MDS_SETATTR64_REC, th, attr);
-	/* XXX: send new uid/gid to OST ASAP? */
+		/* Update the OSP object attributes cache. */
+		la = &o->opo_ooa->ooa_attr;
+		spin_lock(&o->opo_lock);
+		if (attr->la_valid & LA_UID) {
+			la->la_uid = attr->la_uid;
+			la->la_valid |= LA_UID;
+		}
 
-	CDEBUG(D_INFO, "(2) set attr "DFID": rc = %d\n",
-	       PFID(&dt->do_lu.lo_header->loh_fid), rc);
+		if (attr->la_valid & LA_GID) {
+			la->la_gid = attr->la_gid;
+			la->la_valid |= LA_GID;
+		}
+		spin_unlock(&o->opo_lock);
+	}
 
 	RETURN(rc);
 }
@@ -996,7 +971,7 @@ unlock:
 		GOTO(out, rc);
 	}
 
-	rc = osp_remote_sync(env, osp, update, &req, false);
+	rc = osp_remote_sync(env, osp, update, &req);
 	if (rc != 0) {
 		if (rc == -ENOENT) {
 			dt->do_lu.lo_header->loh_attr &= ~LOHA_EXISTS;
@@ -1110,68 +1085,6 @@ out:
 	return rc;
 }
 
-static int __osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
-			   const struct lu_buf *buf, const char *name,
-			   int flag, struct thandle *th)
-{
-	struct osp_object	*o = dt2osp_obj(dt);
-	struct dt_update_request *update;
-	struct osp_xattr_entry	*oxe;
-	int			rc;
-	ENTRY;
-
-	LASSERT(buf->lb_len > 0 && buf->lb_buf != NULL);
-	update = thandle_to_dt_update_request(th);
-	LASSERT(update != NULL);
-
-	rc = out_xattr_set_pack(env, &update->dur_buf,
-				lu_object_fid(&dt->do_lu),
-				buf, name, flag, update->dur_batchid);
-	if (rc != 0 || o->opo_ooa == NULL)
-		RETURN(rc);
-
-	oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
-	if (oxe == NULL) {
-		CWARN("%s: cannot cache xattr '%s' of "DFID"\n",
-		      dt->do_lu.lo_dev->ld_obd->obd_name,
-		      name, PFID(lu_object_fid(&dt->do_lu)));
-
-		RETURN(0);
-	}
-
-	if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) {
-		struct osp_xattr_entry *old = oxe;
-		struct osp_xattr_entry *tmp;
-
-		tmp = osp_oac_xattr_replace(o, &old, buf->lb_len);
-		osp_oac_xattr_put(oxe);
-		oxe = tmp;
-		if (tmp == NULL) {
-			CWARN("%s: cannot update cached xattr '%s' of "DFID"\n",
-			      dt->do_lu.lo_dev->ld_obd->obd_name,
-			      name, PFID(lu_object_fid(&dt->do_lu)));
-			spin_lock(&o->opo_lock);
-			old->oxe_ready = 0;
-			spin_unlock(&o->opo_lock);
-
-			RETURN(0);
-		}
-
-		/* Drop the ref for entry on list. */
-		osp_oac_xattr_put(old);
-	}
-
-	spin_lock(&o->opo_lock);
-	oxe->oxe_vallen = buf->lb_len;
-	memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
-	oxe->oxe_exist = 1;
-	oxe->oxe_ready = 1;
-	spin_unlock(&o->opo_lock);
-	osp_oac_xattr_put(oxe);
-
-	RETURN(0);
-}
-
 /**
  * Implement OSP layer dt_object_operations::do_declare_xattr_set() interface.
  *
@@ -1199,16 +1112,7 @@ int osp_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
 			  const struct lu_buf *buf, const char *name,
 			  int flag, struct thandle *th)
 {
-	int rc = 0;
-
-	if (!is_only_remote_trans(th)) {
-		rc = __osp_xattr_set(env, dt, buf, name, flag, th);
-
-		CDEBUG(D_INFO, "declare xattr %s set object "DFID": rc = %d\n",
-		       name, PFID(&dt->do_lu.lo_header->loh_fid), rc);
-	}
-
-	return rc;
+	return osp_trans_update_request_create(th);
 }
 
 /**
@@ -1216,11 +1120,10 @@ int osp_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
  *
  * Set extended attribute to the specified MDT/OST object.
  *
- * If it is remote transaction, it will add an OUT_XATTR_SET sub-request into
- * the OUT RPC that will be flushed when the transaction stop. And if the OSP
- * attributes cache is initialized, then check whether the name extended
- * attribute entry exists in the cache or not. If yes, replace it; otherwise,
- * add the extended attribute to the cache.
+ * Add an OUT_XATTR_SET sub-request into the OUT RPC that will be flushed in
+ * the transaction stop. And if the OSP attributes cache is initialized, then
+ * check whether the name extended attribute entry exists in the cache or not.
+ * If yes, replace it; otherwise, add the extended attribute to the cache.
  *
  * \param[in] env	pointer to the thread context
  * \param[in] dt	pointer to the OSP layer dt_object
@@ -1237,44 +1140,65 @@ int osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
 		  const struct lu_buf *buf, const char *name, int fl,
 		  struct thandle *th)
 {
-	int rc = 0;
-
-	if (is_only_remote_trans(th)) {
-		rc = __osp_xattr_set(env, dt, buf, name, fl, th);
-
-		CDEBUG(D_INFO, "xattr %s set object "DFID": rc = %d\n",
-		       name, PFID(&dt->do_lu.lo_header->loh_fid), rc);
-	}
-
-	return rc;
-}
-
-static int __osp_xattr_del(const struct lu_env *env, struct dt_object *dt,
-			   const char *name, struct thandle *th)
-{
+	struct osp_object	*o = dt2osp_obj(dt);
 	struct dt_update_request *update;
-	const struct lu_fid	 *fid;
-	struct osp_object	 *o	= dt2osp_obj(dt);
-	struct osp_xattr_entry	 *oxe;
-	int			  rc;
+	struct osp_xattr_entry	*oxe;
+	int			rc;
+	ENTRY;
 
+	LASSERT(buf->lb_len > 0 && buf->lb_buf != NULL);
 	update = thandle_to_dt_update_request(th);
 	LASSERT(update != NULL);
 
-	fid = lu_object_fid(&dt->do_lu);
-
-	rc = out_xattr_del_pack(env, &update->dur_buf, fid, name,
-				update->dur_batchid);
+	CDEBUG(D_INODE, DFID" set xattr '%s' with size %zd\n",
+	       PFID(lu_object_fid(&dt->do_lu)), name, buf->lb_len);
 
+	rc = out_xattr_set_pack(env, &update->dur_buf,
+				lu_object_fid(&dt->do_lu),
+				buf, name, fl, update->dur_batchid);
 	if (rc != 0 || o->opo_ooa == NULL)
 		return rc;
 
-	oxe = osp_oac_xattr_find(o, name, true);
-	if (oxe != NULL)
-		/* Drop the ref for entry on list. */
+	oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
+	if (oxe == NULL) {
+		CWARN("%s: cannot cache xattr '%s' of "DFID"\n",
+		      dt->do_lu.lo_dev->ld_obd->obd_name,
+		      name, PFID(lu_object_fid(&dt->do_lu)));
+
+		RETURN(0);
+	}
+
+	if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) {
+		struct osp_xattr_entry *old = oxe;
+		struct osp_xattr_entry *tmp;
+
+		tmp = osp_oac_xattr_replace(o, &old, buf->lb_len);
 		osp_oac_xattr_put(oxe);
+		oxe = tmp;
+		if (tmp == NULL) {
+			CWARN("%s: cannot update cached xattr '%s' of "DFID"\n",
+			      dt->do_lu.lo_dev->ld_obd->obd_name,
+			      name, PFID(lu_object_fid(&dt->do_lu)));
+			spin_lock(&o->opo_lock);
+			old->oxe_ready = 0;
+			spin_unlock(&o->opo_lock);
 
-	return 0;
+			RETURN(0);
+		}
+
+		/* Drop the ref for entry on list. */
+		osp_oac_xattr_put(old);
+	}
+
+	spin_lock(&o->opo_lock);
+	oxe->oxe_vallen = buf->lb_len;
+	memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
+	oxe->oxe_exist = 1;
+	oxe->oxe_ready = 1;
+	spin_unlock(&o->opo_lock);
+	osp_oac_xattr_put(oxe);
+
+	RETURN(0);
 }
 
 /**
@@ -1299,16 +1223,7 @@ static int __osp_xattr_del(const struct lu_env *env, struct dt_object *dt,
 int osp_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
 			  const char *name, struct thandle *th)
 {
-	int rc = 0;
-
-	if (!is_only_remote_trans(th)) {
-		rc = __osp_xattr_del(env, dt, name, th);
-
-		CDEBUG(D_INFO, "declare xattr %s del object "DFID": rc = %d\n",
-		       name, PFID(&dt->do_lu.lo_header->loh_fid), rc);
-	}
-
-	return rc;
+	return osp_trans_update_request_create(th);
 }
 
 /**
@@ -1332,16 +1247,26 @@ int osp_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
 int osp_xattr_del(const struct lu_env *env, struct dt_object *dt,
 		  const char *name, struct thandle *th)
 {
-	int rc = 0;
+	struct dt_update_request *update;
+	const struct lu_fid	 *fid = lu_object_fid(&dt->do_lu);
+	struct osp_object	 *o = dt2osp_obj(dt);
+	struct osp_xattr_entry	 *oxe;
+	int			  rc;
 
-	if (is_only_remote_trans(th)) {
-		rc = __osp_xattr_del(env, dt, name, th);
+	update = thandle_to_dt_update_request(th);
+	LASSERT(update != NULL);
 
-		CDEBUG(D_INFO, "xattr %s del object "DFID": rc = %d\n",
-		       name, PFID(&dt->do_lu.lo_header->loh_fid), rc);
-	}
+	rc = out_xattr_del_pack(env, &update->dur_buf, fid, name,
+				update->dur_batchid);
+	if (rc != 0 || o->opo_ooa == NULL)
+		return rc;
 
-	return rc;
+	oxe = osp_oac_xattr_find(o, name, true);
+	if (oxe != NULL)
+		/* Drop the ref for entry on list. */
+		osp_oac_xattr_put(oxe);
+
+	return 0;
 }
 
 /**
@@ -1593,13 +1518,12 @@ int osp_declare_object_destroy(const struct lu_env *env,
 			       struct dt_object *dt, struct thandle *th)
 {
 	struct osp_object	*o = dt2osp_obj(dt);
+	struct osp_device	*osp = lu2osp_dev(dt->do_lu.lo_dev);
 	int			 rc = 0;
 
 	ENTRY;
 
-	/*
-	 * track objects to be destroyed via llog
-	 */
+	LASSERT(!osp->opd_connect_mdt);
 	rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
 
 	RETURN(rc);
@@ -1622,20 +1546,22 @@ int osp_declare_object_destroy(const struct lu_env *env,
  * \retval		0 for success
  * \retval		negative error number on failure
  */
-int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
-		       struct thandle *th)
+static int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
+			      struct thandle *th)
 {
 	struct osp_object	*o = dt2osp_obj(dt);
+	struct osp_device	*osp = lu2osp_dev(dt->do_lu.lo_dev);
 	int			 rc = 0;
 
 	ENTRY;
-
 	o->opo_non_exist = 1;
-	/*
-	 * once transaction is committed put proper command on
-	 * the queue going to our OST
-	 */
+
+	LASSERT(!osp->opd_connect_mdt);
+	/* once transaction is committed put proper command on
+	 * the queue going to our OST. */
 	rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
+	if (rc < 0)
+		RETURN(rc);
 
 	/* not needed in cache any more */
 	set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);