staging/lustre/ost: check pre 2.4 echo client in obdo validation

Because old echo client still uses o_id/o_seq for objid,
but new echo client will uses FID for the objid. Add
OBD_CONNECT_FID for 2.4 echo client, so 2.4 OST will
convert o_id/o_seq to FID if the request from old echo
client.

Add local flag OBD_FL_OSTID for o_flags to indicate
OST does not support FID yet, then echo client will
still send o_id/o_seq to OST.

cleanup ost_validate_obdo

[picked client part for upstream kernel submission]
Intel-bug-id: https://jira.hpdd.intel.com/browse/LU-3187
Lustre-change: http://review.whamcloud.com/6287Signed-off-by: wang di <di.wang@intel.com>
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>
Reviewed-by: Mike Pershin <mike.pershin@intel.com>
Signed-off-by: Peng Tao <tao.peng@emc.com>
Signed-off-by: Andreas Dilger <andreas.dilger@intel.com>
Signed-off-by: Greg Kroah-Hartman <gregkh@linuxfoundation.org>

drivers/staging/lustre/lustre/include/lustre/lustre_idl.h

@@ -3216,13 +3216,25 @@ struct obdo {
 #define o_cksum   o_nlink
 #define o_grant_used o_data_version
 
-static inline void lustre_set_wire_obdo(struct obdo *wobdo, struct obdo *lobdo)
+static inline void lustre_set_wire_obdo(struct obd_connect_data *ocd,
+					struct obdo *wobdo, struct obdo *lobdo)
 {
 	memcpy(wobdo, lobdo, sizeof(*lobdo));
 	wobdo->o_flags &= ~OBD_FL_LOCAL_MASK;
+	if (ocd == NULL)
+		return;
+
+	if (unlikely(!(ocd->ocd_connect_flags & OBD_CONNECT_FID)) &&
+	    fid_seq_is_echo(fid_seq(&lobdo->o_oi.oi_fid))) {
+		/* Currently OBD_FL_OSTID will only be used when 2.4 echo
+		 * client communicate with pre-2.4 server */
+		wobdo->o_oi.oi.oi_id = fid_oid(&lobdo->o_oi.oi_fid);
+		wobdo->o_oi.oi.oi_seq = fid_seq(&lobdo->o_oi.oi_fid);
+	}
 }
 
-static inline void lustre_get_wire_obdo(struct obdo *lobdo, struct obdo *wobdo)
+static inline void lustre_get_wire_obdo(struct obd_connect_data *ocd,
+					struct obdo *lobdo, struct obdo *wobdo)
 {
 	obd_flag local_flags = 0;
 
@@ -3233,9 +3245,19 @@ static inline void lustre_get_wire_obdo(struct obdo *lobdo, struct obdo *wobdo)
 
 	memcpy(lobdo, wobdo, sizeof(*lobdo));
 	if (local_flags != 0) {
-		 lobdo->o_valid |= OBD_MD_FLFLAGS;
-		 lobdo->o_flags &= ~OBD_FL_LOCAL_MASK;
-		 lobdo->o_flags |= local_flags;
+		lobdo->o_valid |= OBD_MD_FLFLAGS;
+		lobdo->o_flags &= ~OBD_FL_LOCAL_MASK;
+		lobdo->o_flags |= local_flags;
+	}
+	if (ocd == NULL)
+		return;
+
+	if (unlikely(!(ocd->ocd_connect_flags & OBD_CONNECT_FID)) &&
+	    fid_seq_is_echo(wobdo->o_oi.oi.oi_seq)) {
+		/* see above */
+		lobdo->o_oi.oi_fid.f_seq = wobdo->o_oi.oi.oi_seq;
+		lobdo->o_oi.oi_fid.f_oid = wobdo->o_oi.oi.oi_id;
+		lobdo->o_oi.oi_fid.f_ver = 0;
 	}
 }
 

drivers/staging/lustre/lustre/obdecho/echo_client.c

@@ -3036,7 +3036,8 @@ static int echo_client_setup(const struct lu_env *env,
 	ocd->ocd_connect_flags = OBD_CONNECT_VERSION | OBD_CONNECT_REQPORTAL |
 				 OBD_CONNECT_BRW_SIZE |
 				 OBD_CONNECT_GRANT | OBD_CONNECT_FULL20 |
-				 OBD_CONNECT_64BITHASH | OBD_CONNECT_LVB_TYPE;
+				 OBD_CONNECT_64BITHASH | OBD_CONNECT_LVB_TYPE |
+				 OBD_CONNECT_FID;
 	ocd->ocd_brw_size = DT_MAX_BRW_SIZE;
 	ocd->ocd_version = LUSTRE_VERSION_CODE;
 	ocd->ocd_group = FID_SEQ_ECHO;

drivers/staging/lustre/lustre/osc/osc_request.c

@@ -181,7 +181,8 @@ static inline void osc_pack_req_body(struct ptlrpc_request *req,
 	body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 	LASSERT(body);
 
-	lustre_set_wire_obdo(&body->oa, oinfo->oi_oa);
+	lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa,
+			     oinfo->oi_oa);
 	osc_pack_capa(req, body, oinfo->oi_capa);
 }
 
@@ -209,7 +210,8 @@ static int osc_getattr_interpret(const struct lu_env *env,
 	body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 	if (body) {
 		CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
-		lustre_get_wire_obdo(aa->aa_oi->oi_oa, &body->oa);
+		lustre_get_wire_obdo(&req->rq_import->imp_connect_data,
+				     aa->aa_oi->oi_oa, &body->oa);
 
 		/* This should really be sent by the OST */
 		aa->aa_oi->oi_oa->o_blksize = DT_MAX_BRW_SIZE;
@@ -288,7 +290,8 @@ static int osc_getattr(const struct lu_env *env, struct obd_export *exp,
 		GOTO(out, rc = -EPROTO);
 
 	CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
-	lustre_get_wire_obdo(oinfo->oi_oa, &body->oa);
+	lustre_get_wire_obdo(&req->rq_import->imp_connect_data, oinfo->oi_oa,
+			     &body->oa);
 
 	oinfo->oi_oa->o_blksize = cli_brw_size(exp->exp_obd);
 	oinfo->oi_oa->o_valid |= OBD_MD_FLBLKSZ;
@@ -332,7 +335,8 @@ static int osc_setattr(const struct lu_env *env, struct obd_export *exp,
 	if (body == NULL)
 		GOTO(out, rc = -EPROTO);
 
-	lustre_get_wire_obdo(oinfo->oi_oa, &body->oa);
+	lustre_get_wire_obdo(&req->rq_import->imp_connect_data, oinfo->oi_oa,
+			     &body->oa);
 
 	EXIT;
 out:
@@ -354,7 +358,8 @@ static int osc_setattr_interpret(const struct lu_env *env,
 	if (body == NULL)
 		GOTO(out, rc = -EPROTO);
 
-	lustre_get_wire_obdo(sa->sa_oa, &body->oa);
+	lustre_get_wire_obdo(&req->rq_import->imp_connect_data, sa->sa_oa,
+			     &body->oa);
 out:
 	rc = sa->sa_upcall(sa->sa_cookie, rc);
 	RETURN(rc);
@@ -450,7 +455,8 @@ int osc_real_create(struct obd_export *exp, struct obdo *oa,
 
 	body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 	LASSERT(body);
-	lustre_set_wire_obdo(&body->oa, oa);
+
+	lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
 
 	ptlrpc_request_set_replen(req);
 
@@ -470,7 +476,8 @@ int osc_real_create(struct obd_export *exp, struct obdo *oa,
 	if (body == NULL)
 		GOTO(out_req, rc = -EPROTO);
 
-	lustre_get_wire_obdo(oa, &body->oa);
+	CDEBUG(D_INFO, "oa flags %x\n", oa->o_flags);
+	lustre_get_wire_obdo(&req->rq_import->imp_connect_data, oa, &body->oa);
 
 	oa->o_blksize = cli_brw_size(exp->exp_obd);
 	oa->o_valid |= OBD_MD_FLBLKSZ;
@@ -527,7 +534,8 @@ int osc_punch_base(struct obd_export *exp, struct obd_info *oinfo,
 
 	body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 	LASSERT(body);
-	lustre_set_wire_obdo(&body->oa, oinfo->oi_oa);
+	lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa,
+			     oinfo->oi_oa);
 	osc_pack_capa(req, body, oinfo->oi_capa);
 
 	ptlrpc_request_set_replen(req);
@@ -604,7 +612,8 @@ int osc_sync_base(struct obd_export *exp, struct obd_info *oinfo,
 	/* overload the size and blocks fields in the oa with start/end */
 	body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 	LASSERT(body);
-	lustre_set_wire_obdo(&body->oa, oinfo->oi_oa);
+	lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa,
+			     oinfo->oi_oa);
 	osc_pack_capa(req, body, oinfo->oi_capa);
 
 	ptlrpc_request_set_replen(req);
@@ -782,7 +791,7 @@ static int osc_destroy(const struct lu_env *env, struct obd_export *exp,
 		oa->o_lcookie = *oti->oti_logcookies;
 	body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 	LASSERT(body);
-	lustre_set_wire_obdo(&body->oa, oa);
+	lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
 
 	osc_pack_capa(req, body, (struct obd_capa *)capa);
 	ptlrpc_request_set_replen(req);
@@ -1302,7 +1311,7 @@ static int osc_brw_prep_request(int cmd, struct client_obd *cli,struct obdo *oa,
 	niobuf = req_capsule_client_get(pill, &RMF_NIOBUF_REMOTE);
 	LASSERT(body != NULL && ioobj != NULL && niobuf != NULL);
 
-	lustre_set_wire_obdo(&body->oa, oa);
+	lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
 
 	obdo_to_ioobj(oa, ioobj);
 	ioobj->ioo_bufcnt = niocount;
@@ -1629,7 +1638,8 @@ static int osc_brw_fini_request(struct ptlrpc_request *req, int rc)
 	}
 out:
 	if (rc >= 0)
-		lustre_get_wire_obdo(aa->aa_oa, &body->oa);
+		lustre_get_wire_obdo(&req->rq_import->imp_connect_data,
+				     aa->aa_oa, &body->oa);
 
 	RETURN(rc);
 }