X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fosd-ldiskfs%2Fosd_handler.c;h=d9f0a40683c3c7612c56ad6f4e23b8e3adfb23f1;hb=e0702769f267dd009a6287bbc9da2760079a101d;hp=523cfdf31f6299f97686e772d7fc9ac051c92c2a;hpb=0bc85700a2e41f369173e92bbe06b27ed1145618;p=fs%2Flustre-release.git

diff --git a/lustre/osd-ldiskfs/osd_handler.c b/lustre/osd-ldiskfs/osd_handler.c
index 523cfdf..d9f0a40 100644
--- a/lustre/osd-ldiskfs/osd_handler.c
+++ b/lustre/osd-ldiskfs/osd_handler.c
@@ -317,11 +317,12 @@ static int osd_check_lma(const struct lu_env *env, struct osd_object *obj)
 	CLASSERT(LMA_OLD_SIZE >= sizeof(*lma));
 	rc = __osd_xattr_get(inode, dentry, XATTR_NAME_LMA,
 			     info->oti_mdt_attrs_old, LMA_OLD_SIZE);
-	if (rc == -ENODATA) {
+	if (rc == -ENODATA && !fid_is_igif(lu_object_fid(&obj->oo_dt.do_lu)) &&
+	    osd_obj2dev(obj)->od_check_ff) {
 		fid = &lma->lma_self_fid;
 		rc = osd_get_idif(info, inode, dentry, fid);
 		if (rc > 0)
-			rc = 0;
+			RETURN(0);
 	}
 
 	if (unlikely(rc == -ENODATA))
@@ -392,7 +393,8 @@ static int osd_fid_lookup(const struct lu_env *env, struct osd_object *obj,
 		RETURN(-ENOENT);
 
 	/* Search order: 1. per-thread cache. */
-	if (lu_fid_eq(fid, &oic->oic_fid)) {
+	if (lu_fid_eq(fid, &oic->oic_fid) &&
+	    likely(oic->oic_dev == dev)) {
 		id = &oic->oic_lid;
 		goto iget;
 	}
@@ -1418,48 +1420,50 @@ static int capa_is_sane(const struct lu_env *env,
 }
 
 int osd_object_auth(const struct lu_env *env, struct dt_object *dt,
-                    struct lustre_capa *capa, __u64 opc)
+		    struct lustre_capa *capa, __u64 opc)
 {
-        const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
-        struct osd_device *dev = osd_dev(dt->do_lu.lo_dev);
-        struct md_capainfo *ci;
-        int rc;
+	const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
+	struct osd_device *osd = osd_dev(dt->do_lu.lo_dev);
+	struct lu_capainfo *lci;
+	int rc;
 
-        if (!dev->od_fl_capa)
-                return 0;
+	if (!osd->od_fl_capa)
+		return 0;
 
-        if (capa == BYPASS_CAPA)
-                return 0;
+	if (capa == BYPASS_CAPA)
+		return 0;
 
-        ci = md_capainfo(env);
-        if (unlikely(!ci))
-                return 0;
+	lci = lu_capainfo_get(env);
+	if (unlikely(lci == NULL))
+		return 0;
 
-        if (ci->mc_auth == LC_ID_NONE)
-                return 0;
+	if (lci->lci_auth == LC_ID_NONE)
+		return 0;
 
-        if (!capa) {
-                CERROR("no capability is provided for fid "DFID"\n", PFID(fid));
-                return -EACCES;
-        }
+	if (capa == NULL) {
+		CERROR("%s: no capability provided for FID "DFID": rc = %d\n",
+		       osd_name(osd), PFID(fid), -EACCES);
+		return -EACCES;
+	}
 
-        if (!lu_fid_eq(fid, &capa->lc_fid)) {
-                DEBUG_CAPA(D_ERROR, capa, "fid "DFID" mismatch with",
-                           PFID(fid));
-                return -EACCES;
-        }
+	if (!lu_fid_eq(fid, &capa->lc_fid)) {
+		DEBUG_CAPA(D_ERROR, capa, "fid "DFID" mismatch with",
+			   PFID(fid));
+		return -EACCES;
+	}
 
-        if (!capa_opc_supported(capa, opc)) {
-                DEBUG_CAPA(D_ERROR, capa, "opc "LPX64" not supported by", opc);
-                return -EACCES;
-        }
+	if (!capa_opc_supported(capa, opc)) {
+		DEBUG_CAPA(D_ERROR, capa, "opc "LPX64" not supported by", opc);
+		return -EACCES;
+	}
 
-        if ((rc = capa_is_sane(env, dev, capa, dev->od_capa_keys))) {
-                DEBUG_CAPA(D_ERROR, capa, "insane (rc %d)", rc);
-                return -EACCES;
-        }
+	rc = capa_is_sane(env, osd, capa, osd->od_capa_keys);
+	if (rc != 0) {
+		DEBUG_CAPA(D_ERROR, capa, "insane: rc = %d", rc);
+		return -EACCES;
+	}
 
-        return 0;
+	return 0;
 }
 
 static struct timespec *osd_inode_time(const struct lu_env *env,
@@ -1795,9 +1799,8 @@ struct dentry *osd_child_dentry_get(const struct lu_env *env,
 }
 
 static int osd_mkfile(struct osd_thread_info *info, struct osd_object *obj,
-                      cfs_umode_t mode,
-                      struct dt_allocation_hint *hint,
-                      struct thandle *th)
+		      umode_t mode, struct dt_allocation_hint *hint,
+		      struct thandle *th)
 {
         int result;
         struct osd_device  *osd = osd_obj2dev(obj);
@@ -1930,16 +1933,16 @@ static int osd_mksym(struct osd_thread_info *info, struct osd_object *obj,
 }
 
 static int osd_mknod(struct osd_thread_info *info, struct osd_object *obj,
-                     struct lu_attr *attr,
-                     struct dt_allocation_hint *hint,
-                     struct dt_object_format *dof,
-                     struct thandle *th)
+		     struct lu_attr *attr,
+		     struct dt_allocation_hint *hint,
+		     struct dt_object_format *dof,
+		     struct thandle *th)
 {
-        cfs_umode_t mode = attr->la_mode & (S_IFMT | S_IALLUGO | S_ISVTX);
-        int result;
+	umode_t mode = attr->la_mode & (S_IFMT | S_IALLUGO | S_ISVTX);
+	int result;
 
-        LINVRNT(osd_invariant(obj));
-        LASSERT(obj->oo_inode == NULL);
+	LINVRNT(osd_invariant(obj));
+	LASSERT(obj->oo_inode == NULL);
         LASSERT(S_ISCHR(mode) || S_ISBLK(mode) ||
                 S_ISFIFO(mode) || S_ISSOCK(mode));
 
@@ -1994,7 +1997,7 @@ static osd_obj_type_f osd_create_type_f(enum dt_format_type type)
 
 static void osd_ah_init(const struct lu_env *env, struct dt_allocation_hint *ah,
 			struct dt_object *parent, struct dt_object *child,
-			cfs_umode_t child_mode)
+			umode_t child_mode)
 {
         LASSERT(ah);
 
@@ -2094,18 +2097,18 @@ static int __osd_oi_insert(const struct lu_env *env, struct osd_object *obj,
 }
 
 int osd_fld_lookup(const struct lu_env *env, struct osd_device *osd,
-		   const struct lu_fid *fid, struct lu_seq_range *range)
+		   obd_seq seq, struct lu_seq_range *range)
 {
 	struct seq_server_site	*ss = osd_seq_site(osd);
 	int			rc;
 
-	if (fid_is_idif(fid)) {
+	if (fid_seq_is_idif(seq)) {
 		fld_range_set_ost(range);
-		range->lsr_index = fid_idif_ost_idx(fid);
+		range->lsr_index = idif_ost_idx(seq);
 		return 0;
 	}
 
-	if (!fid_seq_in_fldb(fid_seq(fid))) {
+	if (!fid_seq_in_fldb(seq)) {
 		fld_range_set_mdt(range);
 		if (ss != NULL)
 			/* FIXME: If ss is NULL, it suppose not get lsr_index
@@ -2116,10 +2119,10 @@ int osd_fld_lookup(const struct lu_env *env, struct osd_device *osd,
 
 	LASSERT(ss != NULL);
 	fld_range_set_any(range);
-	rc = fld_server_lookup(env, ss->ss_server_fld, fid_seq(fid), range);
+	rc = fld_server_lookup(env, ss->ss_server_fld, seq, range);
 	if (rc != 0) {
-		CERROR("%s: cannot find FLD range for "DFID": rc = %d\n",
-		       osd_name(osd), PFID(fid), rc);
+		CERROR("%s: cannot find FLD range for "LPX64": rc = %d\n",
+		       osd_name(osd), seq, rc);
 	}
 	return rc;
 }
@@ -2179,7 +2182,7 @@ static int osd_declare_object_create(const struct lu_env *env,
 	if (fid_is_norm(lu_object_fid(&dt->do_lu)) &&
 		!fid_is_last_id(lu_object_fid(&dt->do_lu)))
 		osd_fld_lookup(env, osd_dt_dev(handle->th_dev),
-			       lu_object_fid(&dt->do_lu), range);
+			       fid_seq(lu_object_fid(&dt->do_lu)), range);
 
 
 	RETURN(rc);
@@ -2877,90 +2880,89 @@ static int osd_xattr_del(const struct lu_env *env, struct dt_object *dt,
 }
 
 static struct obd_capa *osd_capa_get(const struct lu_env *env,
-                                     struct dt_object *dt,
-                                     struct lustre_capa *old,
-                                     __u64 opc)
+				     struct dt_object *dt,
+				     struct lustre_capa *old, __u64 opc)
 {
-        struct osd_thread_info *info = osd_oti_get(env);
-        const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
-        struct osd_object *obj = osd_dt_obj(dt);
-        struct osd_device *dev = osd_obj2dev(obj);
-        struct lustre_capa_key *key = &info->oti_capa_key;
-        struct lustre_capa *capa = &info->oti_capa;
-        struct obd_capa *oc;
-        struct md_capainfo *ci;
-        int rc;
-        ENTRY;
+	struct osd_thread_info *info = osd_oti_get(env);
+	const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
+	struct osd_object *obj = osd_dt_obj(dt);
+	struct osd_device *osd = osd_obj2dev(obj);
+	struct lustre_capa_key *key = &info->oti_capa_key;
+	struct lustre_capa *capa = &info->oti_capa;
+	struct obd_capa *oc;
+	struct lu_capainfo *lci;
+	int rc;
+	ENTRY;
 
-        if (!dev->od_fl_capa)
-                RETURN(ERR_PTR(-ENOENT));
+	if (!osd->od_fl_capa)
+		RETURN(ERR_PTR(-ENOENT));
 
 	LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
-        LINVRNT(osd_invariant(obj));
+	LINVRNT(osd_invariant(obj));
 
-        /* renewal sanity check */
-        if (old && osd_object_auth(env, dt, old, opc))
-                RETURN(ERR_PTR(-EACCES));
-
-        ci = md_capainfo(env);
-        if (unlikely(!ci))
-                RETURN(ERR_PTR(-ENOENT));
-
-        switch (ci->mc_auth) {
-        case LC_ID_NONE:
-                RETURN(NULL);
-        case LC_ID_PLAIN:
-                capa->lc_uid = obj->oo_inode->i_uid;
-                capa->lc_gid = obj->oo_inode->i_gid;
-                capa->lc_flags = LC_ID_PLAIN;
-                break;
-        case LC_ID_CONVERT: {
-                __u32 d[4], s[4];
-
-                s[0] = obj->oo_inode->i_uid;
-                cfs_get_random_bytes(&(s[1]), sizeof(__u32));
-                s[2] = obj->oo_inode->i_gid;
-                cfs_get_random_bytes(&(s[3]), sizeof(__u32));
-                rc = capa_encrypt_id(d, s, key->lk_key, CAPA_HMAC_KEY_MAX_LEN);
-                if (unlikely(rc))
-                        RETURN(ERR_PTR(rc));
-
-                capa->lc_uid   = ((__u64)d[1] << 32) | d[0];
-                capa->lc_gid   = ((__u64)d[3] << 32) | d[2];
-                capa->lc_flags = LC_ID_CONVERT;
-                break;
-        }
-        default:
-                RETURN(ERR_PTR(-EINVAL));
+	/* renewal sanity check */
+	if (old && osd_object_auth(env, dt, old, opc))
+		RETURN(ERR_PTR(-EACCES));
+
+	lci = lu_capainfo_get(env);
+	if (unlikely(lci == NULL))
+		RETURN(ERR_PTR(-ENOENT));
+
+	switch (lci->lci_auth) {
+	case LC_ID_NONE:
+		RETURN(NULL);
+	case LC_ID_PLAIN:
+		capa->lc_uid = obj->oo_inode->i_uid;
+		capa->lc_gid = obj->oo_inode->i_gid;
+		capa->lc_flags = LC_ID_PLAIN;
+		break;
+	case LC_ID_CONVERT: {
+		__u32 d[4], s[4];
+
+		s[0] = obj->oo_inode->i_uid;
+		cfs_get_random_bytes(&(s[1]), sizeof(__u32));
+		s[2] = obj->oo_inode->i_gid;
+		cfs_get_random_bytes(&(s[3]), sizeof(__u32));
+		rc = capa_encrypt_id(d, s, key->lk_key, CAPA_HMAC_KEY_MAX_LEN);
+		if (unlikely(rc))
+			RETURN(ERR_PTR(rc));
+
+		capa->lc_uid   = ((__u64)d[1] << 32) | d[0];
+		capa->lc_gid   = ((__u64)d[3] << 32) | d[2];
+		capa->lc_flags = LC_ID_CONVERT;
+		break;
         }
+	default:
+		RETURN(ERR_PTR(-EINVAL));
+	}
 
-        capa->lc_fid = *fid;
-        capa->lc_opc = opc;
-        capa->lc_flags |= dev->od_capa_alg << 24;
-        capa->lc_timeout = dev->od_capa_timeout;
-        capa->lc_expiry = 0;
+	capa->lc_fid = *fid;
+	capa->lc_opc = opc;
+	capa->lc_flags |= osd->od_capa_alg << 24;
+	capa->lc_timeout = osd->od_capa_timeout;
+	capa->lc_expiry = 0;
 
-        oc = capa_lookup(dev->od_capa_hash, capa, 1);
-        if (oc) {
-                LASSERT(!capa_is_expired(oc));
-                RETURN(oc);
-        }
+	oc = capa_lookup(osd->od_capa_hash, capa, 1);
+	if (oc) {
+		LASSERT(!capa_is_expired(oc));
+		RETURN(oc);
+	}
 
 	spin_lock(&capa_lock);
-	*key = dev->od_capa_keys[1];
+	*key = osd->od_capa_keys[1];
 	spin_unlock(&capa_lock);
 
-        capa->lc_keyid = key->lk_keyid;
-        capa->lc_expiry = cfs_time_current_sec() + dev->od_capa_timeout;
+	capa->lc_keyid = key->lk_keyid;
+	capa->lc_expiry = cfs_time_current_sec() + osd->od_capa_timeout;
 
-        rc = capa_hmac(capa->lc_hmac, capa, key->lk_key);
-        if (rc) {
-                DEBUG_CAPA(D_ERROR, capa, "HMAC failed: %d for", rc);
-                RETURN(ERR_PTR(rc));
-        }
+	rc = capa_hmac(capa->lc_hmac, capa, key->lk_key);
+	if (rc) {
+		DEBUG_CAPA(D_ERROR, capa, "HMAC failed: %d for", rc);
+		RETURN(ERR_PTR(rc));
+	}
 
-        oc = capa_add(dev->od_capa_hash, capa);
-        RETURN(oc);
+	oc = capa_add(osd->od_capa_hash, capa);
+	RETURN(oc);
 }
 
 static int osd_object_sync(const struct lu_env *env, struct dt_object *dt)
@@ -3321,26 +3323,45 @@ static inline int osd_get_fid_from_dentry(struct ldiskfs_dir_entry_2 *de,
 	return rc;
 }
 
-static int osd_remote_fid(const struct lu_env *env, struct osd_device *osd,
-			  struct lu_fid *fid)
+static int osd_mdt_seq_exists(const struct lu_env *env,
+			      struct osd_device *osd, obd_seq seq)
 {
 	struct lu_seq_range	*range = &osd_oti_get(env)->oti_seq_range;
 	struct seq_server_site	*ss = osd_seq_site(osd);
 	int			rc;
 	ENTRY;
 
-	/* Those FID seqs, which are not in FLDB, must be local seq */
-	if (unlikely(!fid_seq_in_fldb(fid_seq(fid)) || ss == NULL))
-		RETURN(0);
+	if (ss == NULL)
+		RETURN(1);
 
-	rc = osd_fld_lookup(env, osd, fid, range);
+	/* XXX: currently, each MDT only store avaible sequence on disk, and no
+	 * allocated sequences information on disk, so we have to lookup FLDB,
+	 * but it probably makes more sense also store allocated sequence
+	 * locally, so we do not need do remote FLDB lookup in OSD */
+	rc = osd_fld_lookup(env, osd, seq, range);
 	if (rc != 0) {
-		CERROR("%s: Can not lookup fld for "DFID"\n",
-		       osd_name(osd), PFID(fid));
-		RETURN(rc);
+		CERROR("%s: Can not lookup fld for "LPX64"\n",
+		       osd_name(osd), seq);
+		RETURN(0);
 	}
 
-	RETURN(ss->ss_node_id != range->lsr_index);
+	RETURN(ss->ss_node_id == range->lsr_index);
+}
+
+static int osd_remote_fid(const struct lu_env *env, struct osd_device *osd,
+			  struct lu_fid *fid)
+{
+	ENTRY;
+
+	/* FID seqs not in FLDB, must be local seq */
+	if (unlikely(!fid_seq_in_fldb(fid_seq(fid))))
+		RETURN(0);
+
+	/* Currently only check this for FID on MDT */
+	if (osd_mdt_seq_exists(env, osd, fid_seq(fid)))
+		RETURN(0);
+
+	RETURN(1);
 }
 
 /**
@@ -3879,6 +3900,7 @@ int osd_add_oi_cache(struct osd_thread_info *info, struct osd_device *osd,
 	       id->oii_ino, id->oii_gen, info);
 	info->oti_cache.oic_lid = *id;
 	info->oti_cache.oic_fid = *fid;
+	info->oti_cache.oic_dev = osd;
 
 	return 0;
 }
@@ -3941,7 +3963,7 @@ static int osd_ea_lookup_rec(const struct lu_env *env, struct osd_object *obj,
 			rc = osd_ea_fid_get(env, obj, ino, fid, id);
 		else
 			osd_id_gen(id, ino, OSD_OII_NOGEN);
-		if (rc != 0 || osd_remote_fid(env, dev, fid)) {
+		if (rc != 0) {
 			fid_zero(&oic->oic_fid);
 			GOTO(out, rc);
 		}
@@ -5537,6 +5559,9 @@ static int osd_device_init0(const struct lu_env *env,
 		GOTO(out_mnt, rc);
 	}
 
+	if (server_name_is_ost(o->od_svname))
+		o->od_is_ost = 1;
+
 	rc = osd_obj_map_init(env, o);
 	if (rc != 0)
 		GOTO(out_mnt, rc);