From: Fan Yong <fan.yong@intel.com>
Date: Thu, 22 Sep 2016 08:54:55 +0000 (+0800)
Subject: LU-8840 osp: handle EA cache properly
X-Git-Tag: 2.9.52~24
X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=commitdiff_plain;h=555d02f47401340182b47b3245a657b52fc3e68a;hp=5801dad47e4727a44b6343a3f7f875d7992f29a3;ds=sidebyside

LU-8840 osp: handle EA cache properly

For success case, dt_xattr_get() should return the EA size
instead of zero. If such EA does not exist, return -ENODATA.

More code cleanup for OSP EA cache to avoid potential reference
leak, buffer overflow, and so on.

Signed-off-by: Fan Yong <fan.yong@intel.com>
Change-Id: I352b99b1ed08f1b15bdb8da2bf28689ae2d61c23
Reviewed-on: https://review.whamcloud.com/23782
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Lai Siyao <lai.siyao@intel.com>
Reviewed-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>
---

diff --git a/lustre/lfsck/lfsck_namespace.c b/lustre/lfsck/lfsck_namespace.c
index 4384bb8..1cb3963 100644
--- a/lustre/lfsck/lfsck_namespace.c
+++ b/lustre/lfsck/lfsck_namespace.c
@@ -2688,7 +2688,8 @@ again:
 		 * When the LFSCK runs again, if the dangling name is still
 		 * there, the LFSCK should move the orphan directory object
 		 * back to the normal namespace. */
-		if (!lpf && !lu_fid_eq(pfid, &tfid) && once) {
+		if (!lpf && !fid_is_zero(pfid) &&
+		    !lu_fid_eq(pfid, &tfid) && once) {
 			linkea_next_entry(ldata);
 			continue;
 		}
@@ -2739,7 +2740,7 @@ again:
 			 * directory contains the specified child, but such
 			 * parent does not match the dotdot name entry, then
 			 * trust the linkEA. */
-			if (!lu_fid_eq(pfid, pfid2)) {
+			if (!fid_is_zero(pfid) && !lu_fid_eq(pfid, pfid2)) {
 				*type = LNIT_UNMATCHED_PAIRS;
 				rc = lfsck_namespace_repair_unmatched_pairs(env,
 						com, child, pfid2, cname);
diff --git a/lustre/osp/osp_object.c b/lustre/osp/osp_object.c
index 1fcbac7..0781eb9 100644
--- a/lustre/osp/osp_object.c
+++ b/lustre/osp/osp_object.c
@@ -126,11 +126,28 @@ static inline __u32 osp_dev2node(struct osp_device *osp)
 	return osp->opd_storage->dd_lu_dev.ld_site->ld_seq_site->ss_node_id;
 }
 
+static inline const char *osp_dto2name(struct osp_object *obj)
+{
+	return obj->opo_obj.do_lu.lo_dev->ld_obd->obd_name;
+}
+
 static inline bool is_ost_obj(struct lu_object *lo)
 {
 	return !lu2osp_dev(lo->lo_dev)->opd_connect_mdt;
 }
 
+static inline void __osp_oac_xattr_assignment(struct osp_object *obj,
+					      struct osp_xattr_entry *oxe,
+					      const struct lu_buf *buf)
+{
+	if (buf->lb_len > 0)
+		memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
+
+	oxe->oxe_vallen = buf->lb_len;
+	oxe->oxe_exist = 1;
+	oxe->oxe_ready = 1;
+}
+
 /**
  * Assign FID to the OST object.
  *
@@ -155,6 +172,24 @@ static void osp_object_assign_fid(const struct lu_env *env,
 	lu_object_assign_fid(env, &o->opo_obj.do_lu, &osi->osi_fid);
 }
 
+#define OXE_DEFAULT_LEN	16
+
+/**
+ * Release reference from the OSP object extended attribute entry.
+ *
+ * If it is the last reference, then free the entry.
+ *
+ * \param[in] oxe	pointer to the OSP object extended attribute entry.
+ */
+static inline void osp_oac_xattr_put(struct osp_xattr_entry *oxe)
+{
+	if (atomic_dec_and_test(&oxe->oxe_ref)) {
+		LASSERT(list_empty(&oxe->oxe_list));
+
+		OBD_FREE(oxe, oxe->oxe_buflen);
+	}
+}
+
 /**
  * Find the named extended attribute in the OSP object attributes cache.
  *
@@ -235,9 +270,10 @@ static struct osp_xattr_entry *
 osp_oac_xattr_find_or_add(struct osp_object *obj, const char *name, size_t len)
 {
 	struct osp_xattr_entry *oxe;
-	struct osp_xattr_entry *tmp	= NULL;
-	size_t			namelen = strlen(name);
-	size_t			size	= sizeof(*oxe) + namelen + 1 + len;
+	struct osp_xattr_entry *tmp = NULL;
+	size_t namelen = strlen(name);
+	size_t size = sizeof(*oxe) + namelen + 1 +
+		      (len ? len : OXE_DEFAULT_LEN);
 
 	oxe = osp_oac_xattr_find(obj, name, false);
 	if (oxe != NULL)
@@ -272,67 +308,87 @@ osp_oac_xattr_find_or_add(struct osp_object *obj, const char *name, size_t len)
 }
 
 /**
- * Add the given extended attribute to the OSP object attributes cache.
+ * Assign the cached OST-object's EA with the given value.
  *
- * If there is an old extended attributed entry with the same name,
- * remove it from the cache and return it via the parameter \a poxe.
+ * If the current EA entry in cache has not enough space to hold the new
+ * value, remove it, create a new one, then assign with the given value.
  *
  * \param[in] obj	pointer to the OSP object
- * \param[in,out] poxe	double pointer to the OSP object extended attribute
- *			entry: the new extended attribute entry is transferred
- *			via such pointer target, and if old the extended
- *			attribute entry exists, then it will be returned back
- *			via such pointer target.
- * \param[in] len	the length of the (new) extended attribute value
- *
- * \retval		pointer to the new extended attribute entry
- * \retval		NULL for failure cases.
+ * \param[in] oxe	pointer to the cached EA entry to be assigned
+ * \param[in] buf	pointer to the buffer with new EA value
+ *
+ * \retval		pointer to the new created EA entry in cache if
+ *			current entry is not big enough; otherwise, the
+ *			input 'oxe' will be returned.
  */
 static struct osp_xattr_entry *
-osp_oac_xattr_replace(struct osp_object *obj,
-		      struct osp_xattr_entry **poxe, size_t len)
+osp_oac_xattr_assignment(struct osp_object *obj, struct osp_xattr_entry *oxe,
+			 const struct lu_buf *buf)
 {
-	struct osp_xattr_entry *oxe;
-	size_t			namelen = (*poxe)->oxe_namelen;
-	size_t			size	= sizeof(*oxe) + namelen + 1 + len;
+	struct osp_xattr_entry *new = NULL;
+	struct osp_xattr_entry *old = NULL;
+	int namelen = oxe->oxe_namelen;
+	bool unlink_only = false;
 
-	OBD_ALLOC(oxe, size);
-	if (unlikely(oxe == NULL))
-		return NULL;
-
-	INIT_LIST_HEAD(&oxe->oxe_list);
-	oxe->oxe_buflen = size;
-	oxe->oxe_namelen = namelen;
-	memcpy(oxe->oxe_buf, (*poxe)->oxe_buf, namelen);
-	oxe->oxe_value = oxe->oxe_buf + namelen + 1;
-	/* One ref is for the caller, the other is for the entry on the list. */
-	atomic_set(&oxe->oxe_ref, 2);
+	if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) {
+		size_t size = sizeof(*oxe) + namelen + 1 + buf->lb_len;
+
+		OBD_ALLOC(new, size);
+		if (likely(new)) {
+			INIT_LIST_HEAD(&new->oxe_list);
+			new->oxe_buflen = size;
+			new->oxe_namelen = namelen;
+			memcpy(new->oxe_buf, oxe->oxe_buf, namelen);
+			new->oxe_value = new->oxe_buf + namelen + 1;
+			/* One ref is for the caller,
+			 * the other is for the entry on the list. */
+			atomic_set(&new->oxe_ref, 2);
+			__osp_oac_xattr_assignment(obj, new, buf);
+		} else {
+			unlink_only = true;
+			CWARN("%s: cannot update cached xattr %.*s of "DFID"\n",
+			      osp_dto2name(obj), namelen, oxe->oxe_buf,
+			      PFID(lu_object_fid(&obj->opo_obj.do_lu)));
+		}
+	}
 
 	spin_lock(&obj->opo_lock);
-	*poxe = osp_oac_xattr_find_locked(obj, oxe->oxe_buf, namelen);
-	LASSERT(*poxe != NULL);
+	old = osp_oac_xattr_find_locked(obj, oxe->oxe_buf, namelen);
+	if (likely(old)) {
+		if (new) {
+			/* Unlink the 'old'. */
+			list_del_init(&old->oxe_list);
 
-	list_del_init(&(*poxe)->oxe_list);
-	list_add_tail(&oxe->oxe_list, &obj->opo_xattr_list);
-	spin_unlock(&obj->opo_lock);
+			/* Drop the ref for 'old' on list. */
+			osp_oac_xattr_put(old);
 
-	return oxe;
-}
+			/* Insert 'new' into list. */
+			list_add_tail(&new->oxe_list, &obj->opo_xattr_list);
 
-/**
- * Release reference from the OSP object extended attribute entry.
- *
- * If it is the last reference, then free the entry.
- *
- * \param[in] oxe	pointer to the OSP object extended attribute entry.
- */
-static inline void osp_oac_xattr_put(struct osp_xattr_entry *oxe)
-{
-	if (atomic_dec_and_test(&oxe->oxe_ref)) {
-		LASSERT(list_empty(&oxe->oxe_list));
+			/* Drop the ref for current using. */
+			osp_oac_xattr_put(oxe);
+			oxe = new;
+		} else if (unlink_only) {
+			/* Unlink the 'old'. */
+			list_del_init(&old->oxe_list);
 
-		OBD_FREE(oxe, oxe->oxe_buflen);
+			/* Drop the ref for 'old' on list. */
+			osp_oac_xattr_put(old);
+		} else {
+			__osp_oac_xattr_assignment(obj, oxe, buf);
+		}
+	} else if (new) {
+		/* Someone unlinked the 'old' by race, need NOT to assign
+		 * for unlinked 'oxe', just add the 'new' one. */
+		list_add_tail(&new->oxe_list, &obj->opo_xattr_list);
+
+		/* Drop the ref for current using. */
+		osp_oac_xattr_put(oxe);
+		oxe = new;
 	}
+	spin_unlock(&obj->opo_lock);
+
+	return oxe;
 }
 
 /**
@@ -712,13 +768,18 @@ static int osp_xattr_get_interpterer(const struct lu_env *env,
 	struct osp_xattr_entry	*oxe  = data;
 	struct lu_buf		*rbuf = &osp_env_info(env)->osi_lb2;
 
-	if (rc == 0) {
+	if (!rc) {
 		size_t len = sizeof(*oxe) + oxe->oxe_namelen + 1;
 
 		rc = object_update_result_data_get(reply, rbuf, index);
 		if (rc < 0 || rbuf->lb_len > (oxe->oxe_buflen - len)) {
 			spin_lock(&obj->opo_lock);
-			oxe->oxe_ready = 0;
+			if (unlikely(rc == -ENODATA)) {
+				oxe->oxe_exist = 0;
+				oxe->oxe_ready = 1;
+			} else {
+				oxe->oxe_ready = 0;
+			}
 			spin_unlock(&obj->opo_lock);
 			osp_oac_xattr_put(oxe);
 
@@ -726,10 +787,7 @@ static int osp_xattr_get_interpterer(const struct lu_env *env,
 		}
 
 		spin_lock(&obj->opo_lock);
-		oxe->oxe_vallen = rbuf->lb_len;
-		memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
-		oxe->oxe_exist = 1;
-		oxe->oxe_ready = 1;
+		__osp_oac_xattr_assignment(obj, oxe, rbuf);
 		spin_unlock(&obj->opo_lock);
 	} else if (rc == -ENOENT || rc == -ENODATA) {
 		spin_lock(&obj->opo_lock);
@@ -777,16 +835,14 @@ static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
 	LASSERT(buf != NULL);
 	LASSERT(name != NULL);
 
-	namelen = strlen(name);
-
-	/* If only for xattr size, return directly. */
 	if (unlikely(buf->lb_len == 0))
-		return 0;
+		return -EINVAL;
 
 	oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
 	if (oxe == NULL)
 		return -ENOMEM;
 
+	namelen = strlen(name);
 	mutex_lock(&osp->opd_async_requests_mutex);
 	rc = osp_insert_async_request(env, OUT_XATTR_GET, obj, 1,
 				      &namelen, (const void **)&name,
@@ -853,8 +909,8 @@ int osp_xattr_get(const struct lu_env *env, struct dt_object *dt,
 	struct ptlrpc_request	*req	= NULL;
 	struct object_update_reply *reply;
 	struct osp_xattr_entry	*oxe	= NULL;
-	const char		*dname  = dt->do_lu.lo_dev->ld_obd->obd_name;
-	int			 rc	= 0;
+	const char *dname = osp_dto2name(obj);
+	int rc = 0;
 	ENTRY;
 
 	LASSERT(buf != NULL);
@@ -957,15 +1013,15 @@ unlock:
 	if (rc < 0)
 		GOTO(out, rc);
 
-	if (buf->lb_buf == NULL)
-		GOTO(out, rc);
-
-	if (unlikely(buf->lb_len < rbuf->lb_len))
-		GOTO(out, rc = -ERANGE);
+	/* The peer should have converted '0' as '-ENODATA'. */
+	if (unlikely(!rc))
+		GOTO(out, rc = -EINVAL);
 
-	memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len);
+	/* For detecting EA size. */
+	if (!buf->lb_buf)
+		GOTO(out, rc);
 
-	if (oxe == NULL) {
+	if (!oxe) {
 		oxe = osp_oac_xattr_find_or_add(obj, name, rbuf->lb_len);
 		if (oxe == NULL) {
 			CWARN("%s: Fail to add xattr (%s) to "
@@ -976,38 +1032,18 @@ unlock:
 		}
 	}
 
-	if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < rbuf->lb_len) {
-		struct osp_xattr_entry *old = oxe;
-		struct osp_xattr_entry *tmp;
-
-		tmp = osp_oac_xattr_replace(obj, &old, rbuf->lb_len);
-		osp_oac_xattr_put(oxe);
-		oxe = tmp;
-		if (tmp == NULL) {
-			CWARN("%s: Fail to update xattr (%s) to "
-			      "cache for "DFID": rc = %d\n",
-			      dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
-			spin_lock(&obj->opo_lock);
-			old->oxe_ready = 0;
-			spin_unlock(&obj->opo_lock);
-
-			GOTO(out, rc);
-		}
-
-		/* Drop the ref for entry on list. */
-		osp_oac_xattr_put(old);
-	}
-
-	spin_lock(&obj->opo_lock);
-	oxe->oxe_vallen = rbuf->lb_len;
-	memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
-	oxe->oxe_exist = 1;
-	oxe->oxe_ready = 1;
-	spin_unlock(&obj->opo_lock);
+	oxe = osp_oac_xattr_assignment(obj, oxe, rbuf);
 
 	GOTO(out, rc);
 
 out:
+	if (rc > 0 && buf->lb_buf) {
+		if (unlikely(buf->lb_len < rbuf->lb_len))
+			rc = -ERANGE;
+		else
+			memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len);
+	}
+
 	if (req != NULL)
 		ptlrpc_req_finished(req);
 
@@ -1105,41 +1141,14 @@ int osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
 	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)));
+		      osp_dto2name(o), 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);
+	oxe = osp_oac_xattr_assignment(o, oxe, buf);
+	if (oxe)
 		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);
 }
diff --git a/lustre/target/out_handler.c b/lustre/target/out_handler.c
index 1cf8c21..7781ff3 100644
--- a/lustre/target/out_handler.c
+++ b/lustre/target/out_handler.c
@@ -277,19 +277,22 @@ static int out_xattr_get(struct tgt_session_info *tsi)
 	lbuf->lb_len = (int)tti->tti_u.update.tti_update->ou_result_size;
 	lbuf->lb_buf = update_result->our_data;
 	if (lbuf->lb_len == 0)
-		lbuf->lb_buf = 0;
+		lbuf->lb_buf = NULL;
 	dt_read_lock(env, obj, MOR_TGT_CHILD);
 	rc = dt_xattr_get(env, obj, lbuf, name);
 	dt_read_unlock(env, obj);
-	if (rc < 0)
+	if (rc <= 0) {
 		lbuf->lb_len = 0;
-	CDEBUG(D_INFO, "%s: "DFID" get xattr %s len %d\n",
-	       tgt_name(tsi->tsi_tgt), PFID(lu_object_fid(&obj->do_lu)),
-	       name, (int)lbuf->lb_len);
+		if (unlikely(!rc))
+			rc = -ENODATA;
+	} else if (lbuf->lb_buf) {
+		lbuf->lb_len = rc;
+	}
 
-	GOTO(out, rc);
+	CDEBUG(D_INFO, "%s: "DFID" get xattr %s len %d: rc = %d\n",
+	       tgt_name(tsi->tsi_tgt), PFID(lu_object_fid(&obj->do_lu)),
+	       name, (int)lbuf->lb_len, rc);
 
-out:
 	object_update_result_insert(reply, lbuf->lb_buf, lbuf->lb_len, idx, rc);
 	RETURN(0);
 }