X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fobdclass%2Fdt_object.c;h=07f2c7c992755cca169622c4866472f618c1bd32;hb=8b0b0892da03dc5de5cd4025b9d31c9217b1e6f1;hp=460cbebb02441955b7433353f32730b264f4899d;hpb=8e3d8b71fb4ec61f93d6fdcdaf8afe2f25fdda10;p=fs%2Flustre-release.git

diff --git a/lustre/obdclass/dt_object.c b/lustre/obdclass/dt_object.c
index 460cbeb..07f2c7c 100644
--- a/lustre/obdclass/dt_object.c
+++ b/lustre/obdclass/dt_object.c
@@ -27,7 +27,7 @@
  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2011, 2013, Intel Corporation.
+ * Copyright (c) 2011, 2014, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -60,7 +60,6 @@ struct lu_context_key dt_key = {
         .lct_init = dt_global_key_init,
         .lct_fini = dt_global_key_fini
 };
-EXPORT_SYMBOL(dt_key);
 
 /* no lock is necessary to protect the list, because call-backs
  * are added during system startup. Please refer to "struct dt_device".
@@ -80,56 +79,77 @@ EXPORT_SYMBOL(dt_txn_callback_del);
 int dt_txn_hook_start(const struct lu_env *env,
                       struct dt_device *dev, struct thandle *th)
 {
-        int rc = 0;
-        struct dt_txn_callback *cb;
+	int rc = 0;
+	struct dt_txn_callback *cb;
 
-        if (th->th_local)
-                return 0;
+	if (th->th_local)
+		return 0;
 
 	list_for_each_entry(cb, &dev->dd_txn_callbacks, dtc_linkage) {
-                if (cb->dtc_txn_start == NULL ||
-                    !(cb->dtc_tag & env->le_ctx.lc_tags))
-                        continue;
-                rc = cb->dtc_txn_start(env, th, cb->dtc_cookie);
-                if (rc < 0)
-                        break;
-        }
-        return rc;
+		struct thandle *dtc_th = th;
+
+		if (cb->dtc_txn_start == NULL ||
+		    !(cb->dtc_tag & env->le_ctx.lc_tags))
+			continue;
+
+		/* Usually dt_txn_hook_start is called from bottom device,
+		 * and if the thandle has th_top, then we need use top
+		 * thandle for the callback in the top thandle layer */
+		if (th->th_top != NULL)
+			dtc_th = th->th_top;
+
+		rc = cb->dtc_txn_start(env, dtc_th, cb->dtc_cookie);
+		if (rc < 0)
+			break;
+	}
+	return rc;
 }
 EXPORT_SYMBOL(dt_txn_hook_start);
 
-int dt_txn_hook_stop(const struct lu_env *env, struct thandle *txn)
+int dt_txn_hook_stop(const struct lu_env *env, struct thandle *th)
 {
-        struct dt_device       *dev = txn->th_dev;
-        struct dt_txn_callback *cb;
-        int                     rc = 0;
+	struct dt_device       *dev = th->th_dev;
+	struct dt_txn_callback *cb;
+	int                     rc = 0;
 
-        if (txn->th_local)
-                return 0;
+	if (th->th_local)
+		return 0;
 
 	list_for_each_entry(cb, &dev->dd_txn_callbacks, dtc_linkage) {
-                if (cb->dtc_txn_stop == NULL ||
-                    !(cb->dtc_tag & env->le_ctx.lc_tags))
-                        continue;
-                rc = cb->dtc_txn_stop(env, txn, cb->dtc_cookie);
-                if (rc < 0)
-                        break;
-        }
-        return rc;
+		struct thandle *dtc_th = th;
+
+		if (cb->dtc_txn_stop == NULL ||
+		    !(cb->dtc_tag & env->le_ctx.lc_tags))
+			continue;
+
+		/* Usually dt_txn_hook_stop is called from bottom device,
+		 * and if the thandle has th_top, then we need use top
+		 * thandle for the callback in the top thandle layer */
+		if (th->th_top != NULL)
+			dtc_th = th->th_top;
+
+		rc = cb->dtc_txn_stop(env, dtc_th, cb->dtc_cookie);
+		if (rc < 0)
+			break;
+	}
+	return rc;
 }
 EXPORT_SYMBOL(dt_txn_hook_stop);
 
-void dt_txn_hook_commit(struct thandle *txn)
+void dt_txn_hook_commit(struct thandle *th)
 {
 	struct dt_txn_callback *cb;
 
-	if (txn->th_local)
+	if (th->th_local)
 		return;
 
-	list_for_each_entry(cb, &txn->th_dev->dd_txn_callbacks,
+	list_for_each_entry(cb, &th->th_dev->dd_txn_callbacks,
 			    dtc_linkage) {
+		/* Right now, the bottom device (OSD) will use this hook
+		 * commit to notify OSP, so we do not check and replace
+		 * the thandle to top thandle now */
 		if (cb->dtc_txn_commit)
-			cb->dtc_txn_commit(txn, cb->dtc_cookie);
+			cb->dtc_txn_commit(th, cb->dtc_cookie);
 	}
 }
 EXPORT_SYMBOL(dt_txn_hook_commit);
@@ -204,10 +224,10 @@ EXPORT_SYMBOL(dt_mode_to_dft);
 int dt_lookup_dir(const struct lu_env *env, struct dt_object *dir,
                   const char *name, struct lu_fid *fid)
 {
-        if (dt_try_as_dir(env, dir))
-                return dt_lookup(env, dir, (struct dt_rec *)fid,
-                                 (const struct dt_key *)name, BYPASS_CAPA);
-        return -ENOTDIR;
+	if (dt_try_as_dir(env, dir))
+		return dt_lookup(env, dir, (struct dt_rec *)fid,
+				 (const struct dt_key *)name);
+	return -ENOTDIR;
 }
 EXPORT_SYMBOL(dt_lookup_dir);
 
@@ -238,7 +258,7 @@ struct dt_object *dt_locate_at(const struct lu_env *env,
 EXPORT_SYMBOL(dt_locate_at);
 
 /**
- * find a object named \a entry in given \a dfh->dfh_o directory.
+ * find an object named \a entry in given \a dfh->dfh_o directory.
  */
 static int dt_find_entry(const struct lu_env *env, const char *entry, void *data)
 {
@@ -320,7 +340,6 @@ dt_store_resolve(const struct lu_env *env, struct dt_device *dt,
         }
         return obj;
 }
-EXPORT_SYMBOL(dt_store_resolve);
 
 static struct dt_object *dt_reg_open(const struct lu_env *env,
                                      struct dt_device *dt,
@@ -365,7 +384,6 @@ struct dt_object *dt_store_open(const struct lu_env *env,
         }
         return file;
 }
-EXPORT_SYMBOL(dt_store_open);
 
 struct dt_object *dt_find_or_create(const struct lu_env *env,
                                     struct dt_device *dt,
@@ -451,8 +469,8 @@ void dt_global_fini(void)
 int dt_read(const struct lu_env *env, struct dt_object *dt,
             struct lu_buf *buf, loff_t *pos)
 {
-        LASSERTF(dt != NULL, "dt is NULL when we want to read record\n");
-        return dt->do_body_ops->dbo_read(env, dt, buf, pos, BYPASS_CAPA);
+	LASSERTF(dt != NULL, "dt is NULL when we want to read record\n");
+	return dt->do_body_ops->dbo_read(env, dt, buf, pos);
 }
 EXPORT_SYMBOL(dt_read);
 
@@ -472,35 +490,31 @@ EXPORT_SYMBOL(dt_read);
 int dt_record_read(const struct lu_env *env, struct dt_object *dt,
                    struct lu_buf *buf, loff_t *pos)
 {
-        int rc;
+	ssize_t size;
 
-        LASSERTF(dt != NULL, "dt is NULL when we want to read record\n");
+	LASSERTF(dt != NULL, "dt is NULL when we want to read record\n");
 
-        rc = dt->do_body_ops->dbo_read(env, dt, buf, pos, BYPASS_CAPA);
-
-        if (rc == buf->lb_len)
-                rc = 0;
-        else if (rc >= 0)
-                rc = -EFAULT;
-        return rc;
+	size = dt->do_body_ops->dbo_read(env, dt, buf, pos);
+	if (size < 0)
+		return size;
+	return (size == (ssize_t)buf->lb_len) ? 0 : -EFAULT;
 }
 EXPORT_SYMBOL(dt_record_read);
 
 int dt_record_write(const struct lu_env *env, struct dt_object *dt,
                     const struct lu_buf *buf, loff_t *pos, struct thandle *th)
 {
-        int rc;
+	ssize_t size;
 
-        LASSERTF(dt != NULL, "dt is NULL when we want to write record\n");
-        LASSERT(th != NULL);
-        LASSERT(dt->do_body_ops);
-        LASSERT(dt->do_body_ops->dbo_write);
-        rc = dt->do_body_ops->dbo_write(env, dt, buf, pos, th, BYPASS_CAPA, 1);
-        if (rc == buf->lb_len)
-                rc = 0;
-        else if (rc >= 0)
-                rc = -EFAULT;
-        return rc;
+	LASSERTF(dt != NULL, "dt is NULL when we want to write record\n");
+	LASSERT(th != NULL);
+	LASSERT(dt->do_body_ops);
+	LASSERT(dt->do_body_ops->dbo_write);
+
+	size = dt->do_body_ops->dbo_write(env, dt, buf, pos, th, 1);
+	if (size < 0)
+		return size;
+	return (size == (ssize_t)buf->lb_len) ? 0 : -EFAULT;
 }
 EXPORT_SYMBOL(dt_record_write);
 
@@ -529,7 +543,7 @@ void dt_version_set(const struct lu_env *env, struct dt_object *o,
         vbuf.lb_buf = &version;
         vbuf.lb_len = sizeof(version);
 
-        rc = dt_xattr_set(env, o, &vbuf, xname, 0, th, BYPASS_CAPA);
+	rc = dt_xattr_set(env, o, &vbuf, xname, 0, th);
         if (rc < 0)
                 CDEBUG(D_INODE, "Can't set version, rc %d\n", rc);
         return;
@@ -546,7 +560,7 @@ dt_obj_version_t dt_version_get(const struct lu_env *env, struct dt_object *o)
         LASSERT(o);
         vbuf.lb_buf = &version;
         vbuf.lb_len = sizeof(version);
-        rc = dt_xattr_get(env, o, &vbuf, xname, BYPASS_CAPA);
+	rc = dt_xattr_get(env, o, &vbuf, xname);
         if (rc != sizeof(version)) {
                 CDEBUG(D_INODE, "Can't get version, rc %d\n", rc);
                 version = 0;
@@ -666,15 +680,19 @@ static inline const struct dt_index_features *dt_index_feat_select(__u64 seq,
  * \param arg  - is a pointer to the idx_info structure
  */
 static int dt_index_page_build(const struct lu_env *env, union lu_page *lp,
-			       int nob, const struct dt_it_ops *iops,
+			       size_t nob, const struct dt_it_ops *iops,
 			       struct dt_it *it, __u32 attr, void *arg)
 {
 	struct idx_info		*ii = (struct idx_info *)arg;
 	struct lu_idxpage	*lip = &lp->lp_idx;
 	char			*entry;
-	int			 rc, size;
+	size_t			 size;
+	int			 rc;
 	ENTRY;
 
+	if (nob < LIP_HDR_SIZE)
+		return -EINVAL;
+
 	/* initialize the header of the new container */
 	memset(lip, 0, LIP_HDR_SIZE);
 	lip->lip_magic = LIP_MAGIC;
@@ -786,7 +804,7 @@ int dt_index_walk(const struct lu_env *env, struct dt_object *obj,
 {
 	struct dt_it		*it;
 	const struct dt_it_ops	*iops;
-	unsigned int		 pageidx, nob, nlupgs = 0;
+	size_t			 pageidx, nob, nlupgs = 0;
 	int			 rc;
 	ENTRY;
 
@@ -794,13 +812,13 @@ int dt_index_walk(const struct lu_env *env, struct dt_object *obj,
 	LASSERT(obj->do_index_ops != NULL);
 
 	nob = rdpg->rp_count;
-	if (nob <= 0)
+	if (nob == 0)
 		RETURN(-EFAULT);
 
 	/* Iterate through index and fill containers from @rdpg */
 	iops = &obj->do_index_ops->dio_it;
 	LASSERT(iops != NULL);
-	it = iops->init(env, obj, rdpg->rp_attrs, BYPASS_CAPA);
+	it = iops->init(env, obj, rdpg->rp_attrs);
 	if (IS_ERR(it))
 		RETURN(PTR_ERR(it));
 
@@ -841,7 +859,7 @@ int dt_index_walk(const struct lu_env *env, struct dt_object *obj,
 
 		/* fill lu pages */
 		for (i = 0; i < LU_PAGE_COUNT; i++, lp++, nob -= LU_PAGE_SIZE) {
-			rc = filler(env, lp, min_t(int, nob, LU_PAGE_SIZE),
+			rc = filler(env, lp, min_t(size_t, nob, LU_PAGE_SIZE),
 				    iops, it, rdpg->rp_attrs, arg);
 			if (rc < 0)
 				break;
@@ -859,7 +877,7 @@ out:
 	iops->fini(env, it);
 
 	if (rc >= 0)
-		rc = min_t(unsigned int, nlupgs * LU_PAGE_SIZE, rdpg->rp_count);
+		rc = min_t(size_t, nlupgs * LU_PAGE_SIZE, rdpg->rp_count);
 
 	RETURN(rc);
 }
@@ -954,7 +972,7 @@ out:
 }
 EXPORT_SYMBOL(dt_index_read);
 
-#ifdef LPROCFS
+#ifdef CONFIG_PROC_FS
 int lprocfs_dt_blksize_seq_show(struct seq_file *m, void *v)
 {
 	struct dt_device *dt = m->private;
@@ -1048,4 +1066,4 @@ int lprocfs_dt_filesfree_seq_show(struct seq_file *m, void *v)
 }
 EXPORT_SYMBOL(lprocfs_dt_filesfree_seq_show);
 
-#endif /* LPROCFS */
+#endif /* CONFIG_PROC_FS */