X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fosd-zfs%2Fosd_xattr.c;h=2900fb6a63c505fe0fa0b64f93beb082cec4d3f8;hb=93fe562c5dd3829939e3bc0533918f66b19776a4;hp=c7ea065f2df76c70e3fd7b98e1642c834e6badbd;hpb=da94c5388a4e4344e86b837ff35c2c693569fc77;p=fs%2Flustre-release.git

diff --git a/lustre/osd-zfs/osd_xattr.c b/lustre/osd-zfs/osd_xattr.c
index c7ea065..2900fb6 100644
--- a/lustre/osd-zfs/osd_xattr.c
+++ b/lustre/osd-zfs/osd_xattr.c
@@ -28,7 +28,7 @@
  * Use is subject to license terms.
  */
 /*
- * Copyright (c) 2011, 2012 Whamcloud, Inc.
+ * Copyright (c) 2012, 2013, Intel Corporation.
  * Use is subject to license terms.
  */
 /*
@@ -42,14 +42,10 @@
  * Author: Mike Pershin <tappro@whamcloud.com>
  */
 
-#ifndef EXPORT_SYMTAB
-# define EXPORT_SYMTAB
-#endif
 #define DEBUG_SUBSYSTEM S_OSD
 
 #include <lustre_ver.h>
 #include <libcfs/libcfs.h>
-#include <lustre_fsfilt.h>
 #include <obd_support.h>
 #include <lustre_net.h>
 #include <obd.h>
@@ -84,28 +80,23 @@
  *
  * No locking is done here.
  */
-int __osd_xattr_cache(const struct lu_env *env, struct osd_object *obj)
+int __osd_xattr_load(udmu_objset_t *uos, uint64_t dnode, nvlist_t **sa_xattr)
 {
-	struct osd_device *osd = osd_obj2dev(obj);
-	udmu_objset_t     *uos = &osd->od_objset;
-	sa_handle_t       *sa_hdl;
-	char              *buf;
-	int                size;
-	int                rc;
+	sa_handle_t *sa_hdl;
+	char	    *buf;
+	int	     rc, size;
 
-	LASSERT(obj->oo_sa_xattr == NULL);
-	LASSERT(obj->oo_db != NULL);
+	if (unlikely(dnode == ZFS_NO_OBJECT))
+		return -ENOENT;
 
-	rc = -sa_handle_get(uos->os, obj->oo_db->db_object, NULL,
-			SA_HDL_PRIVATE, &sa_hdl);
+	rc = -sa_handle_get(uos->os, dnode, NULL, SA_HDL_PRIVATE, &sa_hdl);
 	if (rc)
 		return rc;
 
 	rc = -sa_size(sa_hdl, SA_ZPL_DXATTR(uos), &size);
 	if (rc) {
 		if (rc == -ENOENT)
-			rc = -nvlist_alloc(&obj->oo_sa_xattr,
-					NV_UNIQUE_NAME, KM_SLEEP);
+			rc = -nvlist_alloc(sa_xattr, NV_UNIQUE_NAME, KM_SLEEP);
 		goto out_sa;
 	}
 
@@ -116,7 +107,7 @@ int __osd_xattr_cache(const struct lu_env *env, struct osd_object *obj)
 	}
 	rc = -sa_lookup(sa_hdl, SA_ZPL_DXATTR(uos), buf, size);
 	if (rc == 0)
-		rc = -nvlist_unpack(buf, size, &obj->oo_sa_xattr, KM_SLEEP);
+		rc = -nvlist_unpack(buf, size, sa_xattr, KM_SLEEP);
 	sa_spill_free(buf);
 out_sa:
 	sa_handle_destroy(sa_hdl);
@@ -124,6 +115,16 @@ out_sa:
 	return rc;
 }
 
+static inline int __osd_xattr_cache(const struct lu_env *env,
+				    struct osd_object *obj)
+{
+	LASSERT(obj->oo_sa_xattr == NULL);
+	LASSERT(obj->oo_db != NULL);
+
+	return __osd_xattr_load(&osd_obj2dev(obj)->od_objset,
+				obj->oo_db->db_object, &obj->oo_sa_xattr);
+}
+
 int __osd_sa_xattr_get(const struct lu_env *env, struct osd_object *obj,
 		const struct lu_buf *buf, const char *name, int *sizep)
 {
@@ -156,28 +157,21 @@ int __osd_sa_xattr_get(const struct lu_env *env, struct osd_object *obj,
 	return 0;
 }
 
-int __osd_xattr_get(const struct lu_env *env, struct osd_object *obj,
-		struct lu_buf *buf, const char *name, int *sizep)
+int __osd_xattr_get_large(const struct lu_env *env, udmu_objset_t *uos,
+			  uint64_t xattr, struct lu_buf *buf,
+			  const char *name, int *sizep)
 {
-	struct osd_device *osd = osd_obj2dev(obj);
-	udmu_objset_t     *uos = &osd->od_objset;
-	uint64_t           xa_data_obj;
-	dmu_buf_t         *xa_data_db;
-	sa_handle_t       *sa_hdl = NULL;
-	uint64_t           size;
-	int                rc;
-
-	/* check SA_ZPL_DXATTR first then fallback to directory xattr */
-	rc = __osd_sa_xattr_get(env, obj, buf, name, sizep);
-	if (rc != -ENOENT)
-		return rc;
+	dmu_buf_t	*xa_data_db;
+	sa_handle_t	*sa_hdl = NULL;
+	uint64_t	 xa_data_obj, size;
+	int		 rc;
 
 	/* are there any extended attributes? */
-	if (obj->oo_xattr == ZFS_NO_OBJECT)
+	if (xattr == ZFS_NO_OBJECT)
 		return -ENOENT;
 
 	/* Lookup the object number containing the xattr data */
-	rc = -zap_lookup(uos->os, obj->oo_xattr, name, sizeof(uint64_t), 1,
+	rc = -zap_lookup(uos->os, xattr, name, sizeof(uint64_t), 1,
 			&xa_data_obj);
 	if (rc)
 		return rc;
@@ -219,6 +213,23 @@ out:
 	sa_handle_destroy(sa_hdl);
 out_rele:
 	dmu_buf_rele(xa_data_db, FTAG);
+
+	return rc;
+}
+
+int __osd_xattr_get(const struct lu_env *env, struct osd_object *obj,
+		struct lu_buf *buf, const char *name, int *sizep)
+{
+	int rc;
+
+	/* check SA_ZPL_DXATTR first then fallback to directory xattr */
+	rc = __osd_sa_xattr_get(env, obj, buf, name, sizep);
+	if (rc != -ENOENT)
+		return rc;
+
+	rc = __osd_xattr_get_large(env, &osd_obj2dev(obj)->od_objset,
+				   obj->oo_xattr, buf, name, sizep);
+
 	return rc;
 }
 
@@ -234,9 +245,9 @@ int osd_xattr_get(const struct lu_env *env, struct dt_object *dt,
 	LASSERT(osd_invariant(obj));
 	LASSERT(dt_object_exists(dt));
 
-	cfs_down(&obj->oo_guard);
+	down(&obj->oo_guard);
 	rc = __osd_xattr_get(env, obj, buf, name, &size);
-	cfs_up(&obj->oo_guard);
+	up(&obj->oo_guard);
 
 	if (rc == -ENOENT)
 		rc = -ENODATA;
@@ -321,9 +332,9 @@ int osd_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
 	LASSERT(handle != NULL);
 	oh = container_of0(handle, struct osd_thandle, ot_super);
 
-	cfs_down(&obj->oo_guard);
+	down(&obj->oo_guard);
 	__osd_xattr_declare_set(env, obj, buf->lb_len, name, oh);
-	cfs_up(&obj->oo_guard);
+	up(&obj->oo_guard);
 
 	RETURN(0);
 }
@@ -426,6 +437,20 @@ int __osd_sa_xattr_set(const struct lu_env *env, struct osd_object *obj,
 		return rc;
 	}
 
+	/* Ensure xattr doesn't exist in ZAP */
+	if (obj->oo_xattr != ZFS_NO_OBJECT) {
+		udmu_objset_t     *uos = &osd_obj2dev(obj)->od_objset;
+		uint64_t           xa_data_obj;
+		rc = -zap_lookup(uos->os, obj->oo_xattr,
+				 name, 8, 1, &xa_data_obj);
+		if (rc == 0) {
+			rc = __osd_object_free(uos, xa_data_obj, oh->ot_tx);
+			if (rc == 0)
+				zap_remove(uos->os, obj->oo_xattr,
+					   name, oh->ot_tx);
+		}
+	}
+
 	rc = -nvlist_add_byte_array(obj->oo_sa_xattr, name,
 				    (uchar_t *)buf->lb_buf, buf->lb_len);
 	if (rc)
@@ -435,7 +460,7 @@ int __osd_sa_xattr_set(const struct lu_env *env, struct osd_object *obj,
 	return rc;
 }
 
-static int
+int
 __osd_xattr_set(const struct lu_env *env, struct osd_object *obj,
 		const struct lu_buf *buf, const char *name, int fl,
 		struct osd_thandle *oh)
@@ -457,7 +482,8 @@ __osd_xattr_set(const struct lu_env *env, struct osd_object *obj,
 
 		la->la_valid = LA_MODE;
 		la->la_mode = S_IFDIR | S_IRUGO | S_IWUSR | S_IXUGO;
-		rc = __osd_zap_create(env, uos, &xa_zap_db, tx, la, FTAG, 0);
+		rc = __osd_zap_create(env, uos, &xa_zap_db, tx, la,
+				      obj->oo_db->db_object, FTAG, 0);
 		if (rc)
 			return rc;
 
@@ -512,7 +538,8 @@ __osd_xattr_set(const struct lu_env *env, struct osd_object *obj,
 
 		la->la_valid = LA_MODE;
 		la->la_mode = S_IFREG | S_IRUGO | S_IWUSR;
-		rc = __osd_object_create(env, uos, &xa_data_db, tx, la, FTAG);
+		rc = __osd_object_create(env, uos, &xa_data_db, tx, la,
+					 obj->oo_xattr, FTAG);
 		if (rc)
 			goto out;
 		xa_data_obj = xa_data_db->db_object;
@@ -549,8 +576,8 @@ out:
 }
 
 int osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
-		const struct lu_buf *buf, const char *name, int fl,
-		struct thandle *handle, struct lustre_capa *capa)
+		  const struct lu_buf *buf, const char *name, int fl,
+		  struct thandle *handle, struct lustre_capa *capa)
 {
 	struct osd_object  *obj = osd_dt_obj(dt);
 	struct osd_thandle *oh;
@@ -564,14 +591,11 @@ int osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
 
 	oh = container_of0(handle, struct osd_thandle, ot_super);
 
-	cfs_down(&obj->oo_guard);
+	down(&obj->oo_guard);
 	CDEBUG(D_INODE, "Setting xattr %s with size %d\n",
 		name, (int)buf->lb_len);
-	rc = __osd_sa_xattr_set(env, obj, buf, name, fl, oh);
-	/* place xattr in dnode if SA is full */
-	if (rc == -EFBIG)
-		rc = __osd_xattr_set(env, obj, buf, name, fl, oh);
-	cfs_up(&obj->oo_guard);
+	rc = osd_xattr_set_internal(env, obj, buf, name, fl, oh, capa);
+	up(&obj->oo_guard);
 
 	RETURN(rc);
 }
@@ -628,9 +652,9 @@ int osd_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
 	LASSERT(oh->ot_tx != NULL);
 	LASSERT(obj->oo_db != NULL);
 
-	cfs_down(&obj->oo_guard);
+	down(&obj->oo_guard);
 	__osd_xattr_declare_del(env, obj, name, oh);
-	cfs_up(&obj->oo_guard);
+	up(&obj->oo_guard);
 
 	RETURN(0);
 }
@@ -666,7 +690,7 @@ int __osd_xattr_del(const struct lu_env *env, struct osd_object *obj,
 		return rc;
 
 	if (obj->oo_xattr == ZFS_NO_OBJECT)
-		return rc;
+		return 0;
 
 	rc = -zap_lookup(uos->os, obj->oo_xattr, name, sizeof(uint64_t), 1,
 			&xa_data_obj);
@@ -703,9 +727,9 @@ int osd_xattr_del(const struct lu_env *env, struct dt_object *dt,
 	oh = container_of0(handle, struct osd_thandle, ot_super);
 	LASSERT(oh->ot_tx != NULL);
 
-	cfs_down(&obj->oo_guard);
+	down(&obj->oo_guard);
 	rc = __osd_xattr_del(env, obj, name, oh);
-	cfs_up(&obj->oo_guard);
+	up(&obj->oo_guard);
 
 	RETURN(rc);
 }
@@ -728,14 +752,16 @@ osd_sa_xattr_list(const struct lu_env *env, struct osd_object *obj,
 
 	while ((nvp = nvlist_next_nvpair(obj->oo_sa_xattr, nvp)) != NULL) {
 		len = strlen(nvpair_name(nvp));
-		if (len >= remain)
-			return -ERANGE;
-
-		memcpy(lb->lb_buf, nvpair_name(nvp), len);
-		lb->lb_buf += len;
-		*((char *)lb->lb_buf) = '\0';
-		lb->lb_buf++;
-		remain -= len + 1;
+		if (lb->lb_buf != NULL) {
+			if (len + 1 > remain)
+				return -ERANGE;
+
+			memcpy(lb->lb_buf, nvpair_name(nvp), len);
+			lb->lb_buf += len;
+			*((char *)lb->lb_buf) = '\0';
+			lb->lb_buf++;
+			remain -= len + 1;
+		}
 		counted += len + 1;
 	}
 	return counted;
@@ -756,7 +782,7 @@ int osd_xattr_list(const struct lu_env *env, struct dt_object *dt,
 	LASSERT(osd_invariant(obj));
 	LASSERT(dt_object_exists(dt));
 
-	cfs_down(&obj->oo_guard);
+	down(&obj->oo_guard);
 
 	rc = osd_sa_xattr_list(env, obj, lb);
 	if (rc < 0)
@@ -775,26 +801,30 @@ int osd_xattr_list(const struct lu_env *env, struct dt_object *dt,
 	while ((rc = -udmu_zap_cursor_retrieve_key(env, zc, oti->oti_key,
 						MAXNAMELEN)) == 0) {
 		rc = strlen(oti->oti_key);
-		if (rc >= remain)
-			GOTO(out_fini, rc = -ERANGE);
-
-		memcpy(lb->lb_buf, oti->oti_key, rc);
-		lb->lb_buf += rc;
-		*((char *)lb->lb_buf) = '\0';
-		lb->lb_buf++;
-		remain -= rc + 1;
+		if (lb->lb_buf != NULL) {
+			if (rc + 1 > remain)
+				RETURN(-ERANGE);
+
+			memcpy(lb->lb_buf, oti->oti_key, rc);
+			lb->lb_buf += rc;
+			*((char *)lb->lb_buf) = '\0';
+			lb->lb_buf++;
+			remain -= rc + 1;
+		}
 		counted += rc + 1;
 
 		zap_cursor_advance(zc);
 	}
-	if (rc < 0)
+	if (rc == -ENOENT) /* no more kes in the index */
+		rc = 0;
+	else if (unlikely(rc < 0))
 		GOTO(out_fini, rc);
 	rc = counted;
 
 out_fini:
 	udmu_zap_cursor_fini(zc);
 out:
-	cfs_up(&obj->oo_guard);
+	up(&obj->oo_guard);
 	RETURN(rc);
 
 }