X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Flod%2Flod_dev.c;h=9ab7972b52860e5413efb3bcdb94a8e0d4e3de48;hb=3fc039a99f41725798d157a422174005a49200c8;hp=13e24cfb3a5c1ad99e8fd6ee15b05e0588967f3a;hpb=4112a290df2763d53760ef6a96ee2453a41f1856;p=fs%2Flustre-release.git

diff --git a/lustre/lod/lod_dev.c b/lustre/lod/lod_dev.c
index 13e24cf..9ab7972 100644
--- a/lustre/lod/lod_dev.c
+++ b/lustre/lod/lod_dev.c
@@ -23,7 +23,7 @@
  * Copyright  2009 Sun Microsystems, Inc. All rights reserved
  * Use is subject to license terms.
  *
- * Copyright (c) 2012, Intel Corporation.
+ * Copyright (c) 2012, 2013, Intel Corporation.
  *
  */
 /*
@@ -38,12 +38,10 @@
  * Author: Mikhail Pershin <mike.pershin@intel.com>
  */
 
-#ifndef EXPORT_SYMTAB
-# define EXPORT_SYMTAB
-#endif
 #define DEBUG_SUBSYSTEM S_MDS
 
 #include <obd_class.h>
+#include <md_object.h>
 #include <lustre_fid.h>
 #include <lustre_param.h>
 #include <lustre_update.h>
@@ -56,38 +54,40 @@
  * \param lod LOD to be lookup at.
  * \param fid FID of object to find MDT/OST.
  * \param tgt MDT/OST index to return.
- * \param flags indidcate the FID is on MDS or OST.
+ * \param type indidcate the FID is on MDS or OST.
  **/
 int lod_fld_lookup(const struct lu_env *env, struct lod_device *lod,
-		   const struct lu_fid *fid, __u32 *tgt, int flags)
+		   const struct lu_fid *fid, __u32 *tgt, int *type)
 {
-	struct lu_seq_range	range;
+	struct lu_seq_range	range = { 0 };
 	struct lu_server_fld	*server_fld;
 	int rc = 0;
 	ENTRY;
 
 	LASSERTF(fid_is_sane(fid), "Invalid FID "DFID"\n", PFID(fid));
+
 	if (fid_is_idif(fid)) {
 		*tgt = fid_idif_ost_idx(fid);
+		*type = LU_SEQ_RANGE_OST;
 		RETURN(rc);
 	}
 
-	if (!lod->lod_initialized || !fid_is_norm(fid)) {
+	if (!lod->lod_initialized || (!fid_seq_in_fldb(fid_seq(fid)))) {
 		LASSERT(lu_site2seq(lod2lu_dev(lod)->ld_site) != NULL);
+
 		*tgt = lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_node_id;
+		*type = LU_SEQ_RANGE_MDT;
 		RETURN(rc);
 	}
 
 	server_fld = lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_server_fld;
-	range.lsr_flags = flags;
+	fld_range_set_type(&range, *type);
 	rc = fld_server_lookup(env, server_fld, fid_seq(fid), &range);
-	if (rc) {
-		CERROR("%s: Can't find tgt by seq "LPX64", rc %d\n",
-		       lod2obd(lod)->obd_name, fid_seq(fid), rc);
+	if (rc)
 		RETURN(rc);
-	}
 
 	*tgt = range.lsr_index;
+	*type = range.lsr_flags;
 
 	CDEBUG(D_INFO, "LOD: got tgt %x for sequence: "
 	       LPX64"\n", *tgt, fid_seq(fid));
@@ -96,12 +96,10 @@ int lod_fld_lookup(const struct lu_env *env, struct lod_device *lod,
 }
 
 extern struct lu_object_operations lod_lu_obj_ops;
-extern struct lu_object_operations lod_lu_robj_ops;
 extern struct dt_object_operations lod_obj_ops;
-extern struct dt_lock_operations   lod_lock_ops;
 
 /* Slab for OSD object allocation */
-cfs_mem_cache_t *lod_object_kmem;
+struct kmem_cache *lod_object_kmem;
 
 static struct lu_kmem_descr lod_caches[] = {
 	{
@@ -123,30 +121,17 @@ struct lu_object *lod_object_alloc(const struct lu_env *env,
 {
 	struct lod_object	*lod_obj;
 	struct lu_object	*lu_obj;
-	const struct lu_fid	*fid = &hdr->loh_fid;
-	mdsno_t			mds;
-	int			rc = 0;
 	ENTRY;
 
-	OBD_SLAB_ALLOC_PTR_GFP(lod_obj, lod_object_kmem, CFS_ALLOC_IO);
+	OBD_SLAB_ALLOC_PTR_GFP(lod_obj, lod_object_kmem, GFP_NOFS);
 	if (lod_obj == NULL)
 		RETURN(ERR_PTR(-ENOMEM));
 
-	rc = lod_fld_lookup(env, lu2lod_dev(dev), fid, &mds, LU_SEQ_RANGE_MDT);
-	if (rc) {
-		OBD_SLAB_FREE_PTR(lod_obj, lod_object_kmem);
-		RETURN(ERR_PTR(rc));
-	}
-
-	lod_obj->ldo_mds_num = mds;
 	lu_obj = lod2lu_obj(lod_obj);
 	dt_object_init(&lod_obj->ldo_obj, NULL, dev);
 	lod_obj->ldo_obj.do_ops = &lod_obj_ops;
-	lod_obj->ldo_obj.do_lock_ops = &lod_lock_ops;
-	if (likely(mds == lu_site2seq(dev->ld_site)->ss_node_id))
-		lu_obj->lo_ops = &lod_lu_obj_ops;
-	else
-		lu_obj->lo_ops = &lod_lu_robj_ops;
+	lu_obj->lo_ops = &lod_lu_obj_ops;
+
 	RETURN(lu_obj);
 }
 
@@ -183,7 +168,7 @@ static int lod_cleanup_desc_tgts(const struct lu_env *env,
 	return rc;
 }
 
-static int lodname2mdt_index(char *lodname, int *index)
+static int lodname2mdt_index(char *lodname, long *index)
 {
 	char *ptr, *tmp;
 
@@ -217,98 +202,6 @@ static int lodname2mdt_index(char *lodname, int *index)
 	return 0;
 }
 
-/*
- * Init client sequence manager which is used by local MDS to talk to sequence
- * controller on remote node.
- */
-static int lod_seq_init_cli(const struct lu_env *env,
-			    struct lod_device *lod,
-			    char *tgtuuid, int index)
-{
-	struct seq_server_site	*ss;
-	struct obd_device       *osp;
-	int			rc;
-	char			*prefix;
-	struct obd_uuid		obd_uuid;
-	ENTRY;
-
-	ss = lu_site2seq(lod2lu_dev(lod)->ld_site);
-	LASSERT(ss != NULL);
-
-	/* check if this is adding the first MDC and controller is not yet
-	 * initialized. */
-	if (index != 0 || ss->ss_client_seq)
-		RETURN(0);
-
-	obd_str2uuid(&obd_uuid, tgtuuid);
-	osp = class_find_client_obd(&obd_uuid, LUSTRE_OSP_NAME,
-				   &lod->lod_dt_dev.dd_lu_dev.ld_obd->obd_uuid);
-	if (osp == NULL) {
-		CERROR("%s: can't find %s device\n",
-			lod->lod_dt_dev.dd_lu_dev.ld_obd->obd_name,
-			tgtuuid);
-		RETURN(-EINVAL);
-	}
-
-	if (!osp->obd_set_up) {
-		CERROR("target %s not set up\n", osp->obd_name);
-		rc = -EINVAL;
-	}
-
-	LASSERT(ss->ss_control_exp);
-	OBD_ALLOC_PTR(ss->ss_client_seq);
-	if (ss->ss_client_seq == NULL)
-		RETURN(-ENOMEM);
-
-	OBD_ALLOC(prefix, MAX_OBD_NAME + 5);
-	if (!prefix) {
-		OBD_FREE_PTR(ss->ss_client_seq);
-		ss->ss_client_seq = NULL;
-		RETURN(-ENOMEM);
-	}
-
-	snprintf(prefix, MAX_OBD_NAME + 5, "ctl-%s", osp->obd_name);
-	rc = seq_client_init(ss->ss_client_seq, ss->ss_control_exp,
-			     LUSTRE_SEQ_METADATA, prefix, NULL);
-	OBD_FREE(prefix, MAX_OBD_NAME + 5);
-	if (rc) {
-		OBD_FREE_PTR(ss->ss_client_seq);
-		ss->ss_client_seq = NULL;
-		RETURN(rc);
-	}
-
-	LASSERT(ss->ss_server_seq != NULL);
-	rc = seq_server_set_cli(ss->ss_server_seq, ss->ss_client_seq,
-				env);
-
-	RETURN(rc);
-}
-
-static void lod_seq_fini_cli(struct lod_device *lod)
-{
-	struct seq_server_site *ss;
-
-	ENTRY;
-
-	ss = lu_site2seq(lod2lu_dev(lod)->ld_site);
-	if (ss == NULL) {
-		EXIT;
-		return;
-	}
-
-	if (ss->ss_server_seq)
-		seq_server_set_cli(ss->ss_server_seq,
-			   NULL, NULL);
-
-	if (ss->ss_control_exp) {
-		class_export_put(ss->ss_control_exp);
-		ss->ss_control_exp = NULL;
-	}
-
-	EXIT;
-	return;
-}
-
 /**
  * Procss config log on LOD
  * \param env environment info
@@ -367,10 +260,13 @@ static int lod_process_config(const struct lu_env *env,
 			if (mdt == NULL) {
 				mdt_index = 0;
 			} else {
+				long long_index;
 				rc = lodname2mdt_index(
-					lustre_cfg_string(lcfg, 0), &mdt_index);
+					lustre_cfg_string(lcfg, 0),
+					&long_index);
 				if (rc != 0)
 					GOTO(out, rc);
+				mdt_index = long_index;
 			}
 			rc = lod_add_device(env, lod, arg1, index, gen,
 					    mdt_index, LUSTRE_OSC_NAME, 1);
@@ -378,9 +274,6 @@ static int lod_process_config(const struct lu_env *env,
 			mdt_index = index;
 			rc = lod_add_device(env, lod, arg1, index, gen,
 					    mdt_index, LUSTRE_MDC_NAME, 1);
-			if (rc == 0)
-				rc = lod_seq_init_cli(env, lod, arg1,
-						      mdt_index);
 		} else if (lcfg->lcfg_command == LCFG_LOV_ADD_INA) {
 			/*FIXME: Add mdt_index for LCFG_LOV_ADD_INA*/
 			mdt_index = 0;
@@ -389,19 +282,17 @@ static int lod_process_config(const struct lu_env *env,
 		} else {
 			rc = lod_del_device(env, lod,
 					    &lod->lod_ost_descs,
-					    arg1, index, gen);
+					    arg1, index, gen, true);
 		}
 
 		break;
 	}
 
 	case LCFG_PARAM: {
-		struct lprocfs_static_vars  v = { 0 };
-		struct obd_device	  *obd = lod2obd(lod);
-
-		lprocfs_lod_init_vars(&v);
+		struct obd_device *obd = lod2obd(lod);
 
-		rc = class_process_proc_param(PARAM_LOV, v.obd_vars, lcfg, obd);
+		rc = class_process_proc_seq_param(PARAM_LOV, obd->obd_vars,
+						  lcfg, obd);
 		if (rc > 0)
 			rc = 0;
 		GOTO(out, rc);
@@ -411,9 +302,6 @@ static int lod_process_config(const struct lu_env *env,
 		lu_dev_del_linkage(dev->ld_site, dev);
 		lod_cleanup_desc_tgts(env, lod, &lod->lod_mdt_descs, lcfg);
 		lod_cleanup_desc_tgts(env, lod, &lod->lod_ost_descs, lcfg);
-
-		lod_seq_fini_cli(lod);
-
 		if (lcfg->lcfg_command == LCFG_PRE_CLEANUP)
 			break;
 		/*
@@ -520,71 +408,54 @@ static struct thandle *lod_trans_create(const struct lu_env *env,
 	if (IS_ERR(th))
 		return th;
 
-	CFS_INIT_LIST_HEAD(&th->th_remote_update_list);
 	return th;
 }
 
-static int lod_remote_sync(const struct lu_env *env, struct dt_device *dev,
-			   struct thandle *th)
-{
-	struct update_request *update;
-	int    rc = 0;
-	ENTRY;
-
-	if (cfs_list_empty(&th->th_remote_update_list))
-		RETURN(0);
-
-	cfs_list_for_each_entry(update, &th->th_remote_update_list,
-				ur_list) {
-		/* In DNE phase I, there should be only one OSP
-		 * here, so we will do send/receive one by one,
-		 * instead of sending them parallel, will fix this
-		 * in Phase II */
-		th->th_current_request = update;
-		rc = dt_trans_start(env, update->ur_dt, th);
-		if (rc != 0) {
-			/* FIXME how to revert the partial results
-			 * once error happened? Resolved by 2 Phase commit */
-			update->ur_rc = rc;
-			break;
-		}
-	}
-
-	RETURN(rc);
-}
-
 static int lod_trans_start(const struct lu_env *env, struct dt_device *dev,
 			   struct thandle *th)
 {
 	struct lod_device *lod = dt2lod_dev((struct dt_device *) dev);
-	int rc;
+	int rc = 0;
 
-	rc = lod_remote_sync(env, dev, th);
-	if (rc)
-		return rc;
+	if (unlikely(th->th_update != NULL)) {
+		struct thandle_update *tu = th->th_update;
+		struct dt_update_request *update;
 
+		list_for_each_entry(update, &tu->tu_remote_update_list,
+				    dur_list) {
+			LASSERT(update->dur_dt != NULL);
+			rc = dt_trans_start(env, update->dur_dt, th);
+			if (rc != 0)
+				return rc;
+		}
+	}
 	return dt_trans_start(env, lod->lod_child, th);
 }
 
-static int lod_trans_stop(const struct lu_env *env, struct thandle *th)
+static int lod_trans_stop(const struct lu_env *env, struct dt_device *dt,
+			  struct thandle *th)
 {
-	struct update_request *update;
-	struct update_request *tmp;
-	int rc = 0;
-	int rc2 = 0;
+	struct thandle_update		*tu = th->th_update;
+	struct dt_update_request	*update;
+	struct dt_update_request	*tmp;
+	int				rc2 = 0;
+	int				rc;
+	ENTRY;
 
-	cfs_list_for_each_entry_safe(update, tmp,
-				     &th->th_remote_update_list,
-				     ur_list) {
-		th->th_current_request = update;
-		rc2 = dt_trans_stop(env, update->ur_dt, th);
+	rc = dt_trans_stop(env, th->th_dev, th);
+	if (likely(tu == NULL))
+		RETURN(rc);
+
+	list_for_each_entry_safe(update, tmp,
+				 &tu->tu_remote_update_list,
+				 dur_list) {
+		/* update will be freed inside dt_trans_stop */
+		rc2 = dt_trans_stop(env, update->dur_dt, th);
 		if (unlikely(rc2 != 0 && rc == 0))
 			rc = rc2;
 	}
 
-	rc2 = dt_trans_stop(env, th->th_dev, th);
-
-	return rc2 != 0 ? rc2 : rc;
+	RETURN(rc);
 }
 
 static void lod_conf_get(const struct lu_env *env,
@@ -810,7 +681,7 @@ static struct lu_device *lod_device_free(const struct lu_env *env,
 	struct lu_device  *next = &lod->lod_child->dd_lu_dev;
 	ENTRY;
 
-	LASSERT(cfs_atomic_read(&lu->ld_ref) == 0);
+	LASSERT(atomic_read(&lu->ld_ref) == 0);
 	dt_device_fini(&lod->lod_dt_dev);
 	OBD_FREE_PTR(lod);
 	RETURN(next);
@@ -852,12 +723,12 @@ static struct lu_device *lod_device_fini(const struct lu_env *env,
 
 	lod_procfs_fini(lod);
 
-	rc = lod_fini_tgt(lod, &lod->lod_ost_descs);
+	rc = lod_fini_tgt(env, lod, &lod->lod_ost_descs, true);
 	if (rc)
 		CERROR("%s:can not fini ost descs %d\n",
 			lod2obd(lod)->obd_name, rc);
 
-	rc = lod_fini_tgt(lod, &lod->lod_mdt_descs);
+	rc = lod_fini_tgt(env, lod, &lod->lod_mdt_descs, false);
 	if (rc)
 		CERROR("%s:can not fini mdt descs %d\n",
 			lod2obd(lod)->obd_name, rc);
@@ -943,6 +814,7 @@ static void lod_key_fini(const struct lu_context *ctx,
 		info->lti_ea_store = NULL;
 		info->lti_ea_store_size = 0;
 	}
+	lu_buf_free(&info->lti_linkea_buf);
 	OBD_FREE_PTR(info);
 }
 
@@ -971,25 +843,37 @@ static struct lu_device_type lod_device_type = {
 	.ldt_ctx_tags = LCT_MD_THREAD,
 };
 
-static int lod_obd_health_check(const struct lu_env *env,
-		struct obd_device *obd)
+static int lod_obd_get_info(const struct lu_env *env, struct obd_export *exp,
+			    __u32 keylen, void *key, __u32 *vallen, void *val,
+			    struct lov_stripe_md *lsm)
 {
-	struct lod_device   *d = lu2lod_dev(obd->obd_lu_dev);
-	struct lod_ost_desc *ost;
-	int                  i, rc = 1;
-	ENTRY;
+	int rc = -EINVAL;
 
-	LASSERT(d);
-	lod_getref(&d->lod_ost_descs);
-	lod_foreach_ost(d, i) {
-		ost = OST_TGT(d, i);
-		LASSERT(ost && ost->ltd_ost);
-		rc = obd_health_check(env, ost->ltd_exp->exp_obd);
-		/* one healthy device is enough */
-		if (rc == 0)
-			break;
+	if (KEY_IS(KEY_OSP_CONNECTED)) {
+		struct obd_device	*obd = exp->exp_obd;
+		struct lod_device	*d;
+		struct lod_ost_desc	*ost;
+		int			i, rc = 1;
+
+		if (!obd->obd_set_up || obd->obd_stopping)
+			RETURN(-EAGAIN);
+
+		d = lu2lod_dev(obd->obd_lu_dev);
+		lod_getref(&d->lod_ost_descs);
+		lod_foreach_ost(d, i) {
+			ost = OST_TGT(d, i);
+			LASSERT(ost && ost->ltd_ost);
+
+			rc = obd_get_info(env, ost->ltd_exp, keylen, key,
+					  vallen, val, lsm);
+			/* one healthy device is enough */
+			if (rc == 0)
+				break;
+		}
+		lod_putref(d, &d->lod_ost_descs);
+		RETURN(rc);
 	}
-	lod_putref(d, &d->lod_ost_descs);
+
 	RETURN(rc);
 }
 
@@ -997,7 +881,7 @@ static struct obd_ops lod_obd_device_ops = {
 	.o_owner        = THIS_MODULE,
 	.o_connect      = lod_obd_connect,
 	.o_disconnect   = lod_obd_disconnect,
-	.o_health_check = lod_obd_health_check,
+	.o_get_info     = lod_obd_get_info,
 	.o_pool_new     = lod_pool_new,
 	.o_pool_rem     = lod_pool_remove,
 	.o_pool_add     = lod_pool_add,
@@ -1006,17 +890,17 @@ static struct obd_ops lod_obd_device_ops = {
 
 static int __init lod_mod_init(void)
 {
-	struct lprocfs_static_vars  lvars = { 0 };
-	cfs_proc_dir_entry_t       *lov_proc_dir;
-	int			    rc;
+	struct obd_type	*type;
+	int rc;
 
 	rc = lu_kmem_init(lod_caches);
 	if (rc)
 		return rc;
 
-	lprocfs_lod_init_vars(&lvars);
-
-	rc = class_register_type(&lod_obd_device_ops, NULL, lvars.module_vars,
+	rc = class_register_type(&lod_obd_device_ops, NULL, true, NULL,
+#ifndef HAVE_ONLY_PROCFS_SEQ
+				 NULL,
+#endif
 				 LUSTRE_LOD_NAME, &lod_device_type);
 	if (rc) {
 		lu_kmem_fini(lod_caches);
@@ -1024,23 +908,23 @@ static int __init lod_mod_init(void)
 	}
 
 	/* create "lov" entry in procfs for compatibility purposes */
-	lov_proc_dir = lprocfs_srch(proc_lustre_root, "lov");
-	if (lov_proc_dir == NULL) {
-		lov_proc_dir = lprocfs_register("lov", proc_lustre_root,
-						NULL, NULL);
-		if (IS_ERR(lov_proc_dir))
-			CERROR("lod: can't create compat entry \"lov\": %d\n",
-			       (int)PTR_ERR(lov_proc_dir));
-	}
+	type = class_search_type(LUSTRE_LOV_NAME);
+	if (type != NULL && type->typ_procroot != NULL)
+		return rc;
 
+	type = class_search_type(LUSTRE_LOD_NAME);
+	type->typ_procsym = lprocfs_seq_register("lov", proc_lustre_root,
+						 NULL, NULL);
+	if (IS_ERR(type->typ_procsym)) {
+		CERROR("lod: can't create compat entry \"lov\": %d\n",
+		       (int)PTR_ERR(type->typ_procsym));
+		type->typ_procsym = NULL;
+	}
 	return rc;
 }
 
 static void __exit lod_mod_exit(void)
 {
-
-	lprocfs_try_remove_proc_entry("lov", proc_lustre_root);
-
 	class_unregister_type(LUSTRE_LOD_NAME);
 	lu_kmem_fini(lod_caches);
 }