X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Ftarget%2Ftgt_handler.c;h=c3213bae8727c251ef754efb7d7ffd7c64fe67d5;hb=a23767580aebfab7f093df562ac7598e85b71b3e;hp=aad2aaade35d8929c298b194bba2e5a3f37c1874;hpb=b1e7be00cb6e1e7b3e35877cf8b8a209e456a2ea;p=fs%2Flustre-release.git

diff --git a/lustre/target/tgt_handler.c b/lustre/target/tgt_handler.c
index aad2aaa..c3213ba 100644
--- a/lustre/target/tgt_handler.c
+++ b/lustre/target/tgt_handler.c
@@ -35,10 +35,10 @@
 #define DEBUG_SUBSYSTEM S_CLASS
 
 #include <linux/user_namespace.h>
-#ifdef HAVE_UIDGID_HEADER
-# include <linux/uidgid.h>
-#endif
+#include <linux/delay.h>
+#include <linux/uidgid.h>
 
+#include <libcfs/linux/linux-mem.h>
 #include <obd.h>
 #include <obd_class.h>
 #include <obd_cksum.h>
@@ -63,7 +63,7 @@ EXPORT_SYMBOL(tgt_name);
  *  - create lu_object, corresponding to the fid in mdt_body, and save it in
  *  @tsi;
  *
- *  - if HABEO_CORPUS flag is set for this request type check whether object
+ *  - if HAS_BODY flag is set for this request type check whether object
  *  actually exists on storage (lu_object_exists()).
  *
  */
@@ -100,7 +100,7 @@ static int tgt_mdt_body_unpack(struct tgt_session_info *tsi, __u32 flags)
 			     &tsi->tsi_tgt->lut_bottom->dd_lu_dev,
 			     &body->mbo_fid1, NULL);
 	if (!IS_ERR(obj)) {
-		if ((flags & HABEO_CORPUS) && !lu_object_exists(obj)) {
+		if ((flags & HAS_BODY) && !lu_object_exists(obj)) {
 			lu_object_put(tsi->tsi_env, obj);
 			rc = -ENOENT;
 		} else {
@@ -277,9 +277,8 @@ static int tgt_ost_body_unpack(struct tgt_session_info *tsi, __u32 flags)
 	}
 
 	if (!(body->oa.o_valid & OBD_MD_FLID)) {
-		if (flags & HABEO_CORPUS) {
-			CERROR("%s: OBD_MD_FLID flag is not set in ost_body "
-			       "but OID/FID is mandatory with HABEO_CORPUS\n",
+		if (flags & HAS_BODY) {
+			CERROR("%s: OBD_MD_FLID flag is not set in ost_body but OID/FID is mandatory with HAS_BODY\n",
 			       tgt_name(tsi->tsi_tgt));
 			RETURN(-EPROTO);
 		} else {
@@ -317,15 +316,19 @@ static int tgt_request_preprocess(struct tgt_session_info *tsi,
 	LASSERT(h->th_opc == lustre_msg_get_opc(req->rq_reqmsg));
 	LASSERT(current->journal_info == NULL);
 
-	LASSERT(ergo(flags & (HABEO_CORPUS | HABEO_REFERO),
+	LASSERT(ergo(flags & (HAS_BODY | HAS_REPLY),
 		     h->th_fmt != NULL));
 	if (h->th_fmt != NULL) {
 		req_capsule_set(pill, h->th_fmt);
-		if (req_capsule_has_field(pill, &RMF_MDT_BODY, RCL_CLIENT)) {
+		if (req_capsule_has_field(pill, &RMF_MDT_BODY, RCL_CLIENT) &&
+		    req_capsule_field_present(pill, &RMF_MDT_BODY,
+					      RCL_CLIENT)) {
 			rc = tgt_mdt_body_unpack(tsi, flags);
 			if (rc < 0)
 				RETURN(rc);
 		} else if (req_capsule_has_field(pill, &RMF_OST_BODY,
+						 RCL_CLIENT) &&
+			   req_capsule_field_present(pill, &RMF_OST_BODY,
 						 RCL_CLIENT)) {
 			rc = tgt_ost_body_unpack(tsi, flags);
 			if (rc < 0)
@@ -333,10 +336,10 @@ static int tgt_request_preprocess(struct tgt_session_info *tsi,
 		}
 	}
 
-	if (flags & MUTABOR && tgt_conn_flags(tsi) & OBD_CONNECT_RDONLY)
+	if (flags & IS_MUTABLE && tgt_conn_flags(tsi) & OBD_CONNECT_RDONLY)
 		RETURN(-EROFS);
 
-	if (flags & HABEO_CLAVIS) {
+	if (flags & HAS_KEY) {
 		struct ldlm_request *dlm_req;
 
 		LASSERT(h->th_fmt != NULL);
@@ -416,9 +419,14 @@ static int tgt_handle_request0(struct tgt_session_info *tsi,
 		     OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_MULTI_NET)))
 		RETURN(0);
 
+	/* drop OUT_UPDATE rpc */
+	if (unlikely(lustre_msg_get_opc(req->rq_reqmsg) == OUT_UPDATE &&
+		     OBD_FAIL_CHECK(OBD_FAIL_OUT_UPDATE_DROP)))
+		RETURN(0);
+
 	rc = tgt_request_preprocess(tsi, h, req);
 	/* pack reply if reply format is fixed */
-	if (rc == 0 && h->th_flags & HABEO_REFERO) {
+	if (rc == 0 && h->th_flags & HAS_REPLY) {
 		/* Pack reply */
 		if (req_capsule_has_field(tsi->tsi_pill, &RMF_MDT_MD,
 					  RCL_SERVER))
@@ -443,7 +451,8 @@ static int tgt_handle_request0(struct tgt_session_info *tsi,
 
 			req_capsule_set_size(tsi->tsi_pill, &RMF_SHORT_IO,
 					 RCL_SERVER,
-					 (body->oa.o_flags & OBD_FL_SHORT_IO) ?
+					 (body->oa.o_valid & OBD_MD_FLFLAGS &&
+					  body->oa.o_flags & OBD_FL_SHORT_IO) ?
 					 remote_nb[0].rnb_len : 0);
 		}
 
@@ -461,8 +470,8 @@ static int tgt_handle_request0(struct tgt_session_info *tsi,
 		rc = h->th_act(tsi);
 		if (!is_serious(rc) &&
 		    !req->rq_no_reply && req->rq_reply_state == NULL) {
-			DEBUG_REQ(D_ERROR, req, "%s \"handler\" %s did not "
-				  "pack reply and returned 0 error\n",
+			DEBUG_REQ(D_ERROR, req,
+				  "%s: %s handler did not pack reply but returned no error",
 				  tgt_name(tsi->tsi_tgt), h->th_name);
 			LBUG();
 		}
@@ -518,6 +527,7 @@ static int tgt_filter_recovery_request(struct ptlrpc_request *req,
 	case MDS_HSM_PROGRESS:
 	case MDS_HSM_STATE_SET:
 	case MDS_HSM_REQUEST:
+	case OST_FALLOCATE:
 		*process = target_queue_recovery_request(req, obd);
 		RETURN(0);
 
@@ -553,12 +563,12 @@ static int tgt_handle_recovery(struct ptlrpc_request *req, int reply_fail_id)
 
 	/* sanity check: if the xid matches, the request must be marked as a
 	 * resent or replayed */
-	if (req_can_reconstruct(req, NULL)) {
+	if (req_can_reconstruct(req, NULL) == 1) {
 		if (!(lustre_msg_get_flags(req->rq_reqmsg) &
 		      (MSG_RESENT | MSG_REPLAY))) {
-			DEBUG_REQ(D_WARNING, req, "rq_xid %llu matches "
-				  "saved xid, expected REPLAY or RESENT flag "
-				  "(%x)", req->rq_xid,
+			DEBUG_REQ(D_WARNING, req,
+				  "rq_xid=%llu matches saved XID, expected REPLAY or RESENT flag (%x)",
+				  req->rq_xid,
 				  lustre_msg_get_flags(req->rq_reqmsg));
 			req->rq_status = -ENOTCONN;
 			RETURN(-ENOTCONN);
@@ -601,7 +611,7 @@ static struct tgt_handler *tgt_handler_find_check(struct ptlrpc_request *req)
 
 	tgt = class_exp2tgt(req->rq_export);
 	if (unlikely(tgt == NULL)) {
-		DEBUG_REQ(D_ERROR, req, "%s: No target for connected export\n",
+		DEBUG_REQ(D_ERROR, req, "%s: no target for connected export",
 			  class_exp2obd(req->rq_export)->obd_name);
 		RETURN(ERR_PTR(-EINVAL));
 	}
@@ -612,8 +622,14 @@ static struct tgt_handler *tgt_handler_find_check(struct ptlrpc_request *req)
 
 	/* opcode was not found in slice */
 	if (unlikely(s->tos_hs == NULL)) {
-		CERROR("%s: no handlers for opcode 0x%x\n", tgt_name(tgt),
-		       opc);
+		static bool printed;
+
+		/* don't spew error messages for unhandled RPCs */
+		if (!printed) {
+			CERROR("%s: no handler for opcode 0x%x from %s\n",
+			       tgt_name(tgt), opc, libcfs_id2str(req->rq_peer));
+			printed = true;
+		}
 		RETURN(ERR_PTR(-ENOTSUPP));
 	}
 
@@ -630,18 +646,20 @@ static struct tgt_handler *tgt_handler_find_check(struct ptlrpc_request *req)
 static int process_req_last_xid(struct ptlrpc_request *req)
 {
 	__u64	last_xid;
+	int rc = 0;
+	struct obd_export *exp = req->rq_export;
+	struct tg_export_data *ted = &exp->exp_target_data;
+	bool need_lock = tgt_is_multimodrpcs_client(exp);
 	ENTRY;
 
+	if (need_lock)
+		mutex_lock(&ted->ted_lcd_lock);
 	/* check request's xid is consistent with export's last_xid */
 	last_xid = lustre_msg_get_last_xid(req->rq_reqmsg);
-	if (last_xid > req->rq_export->exp_last_xid)
-		req->rq_export->exp_last_xid = last_xid;
-
-	if (req->rq_xid == 0 ||
-	    (req->rq_xid <= req->rq_export->exp_last_xid)) {
-		DEBUG_REQ(D_ERROR, req, "Unexpected xid %llx vs. "
-			  "last_xid %llx\n", req->rq_xid,
-			  req->rq_export->exp_last_xid);
+	if (last_xid > exp->exp_last_xid)
+		exp->exp_last_xid = last_xid;
+
+	if (req->rq_xid == 0 || req->rq_xid <= exp->exp_last_xid) {
 		/* Some request is allowed to be sent during replay,
 		 * such as OUT update requests, FLD requests, so it
 		 * is possible that replay requests has smaller XID
@@ -658,7 +676,13 @@ static int process_req_last_xid(struct ptlrpc_request *req)
 		 * - The former RPC got chance to be processed;
 		 */
 		if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY))
-			RETURN(-EPROTO);
+			rc = -EPROTO;
+
+		DEBUG_REQ(D_WARNING, req,
+			  "unexpected xid=%llx != exp_last_xid=%llx, rc = %d",
+			  req->rq_xid, exp->exp_last_xid, rc);
+		if (rc)
+			GOTO(out, rc);
 	}
 
 	/* The "last_xid" is the minimum xid among unreplied requests,
@@ -671,19 +695,27 @@ static int process_req_last_xid(struct ptlrpc_request *req)
 	 * replay request will have the larger xid than "exp_last_xid"
 	 */
 	if (req->rq_export->exp_conn_cnt >
-	    lustre_msg_get_conn_cnt(req->rq_reqmsg))
-		RETURN(-ESTALE);
+	    lustre_msg_get_conn_cnt(req->rq_reqmsg)) {
+		CDEBUG(D_RPCTRACE,
+		       "Dropping request %llu from an old epoch %u/%u\n",
+		       req->rq_xid,
+		       lustre_msg_get_conn_cnt(req->rq_reqmsg),
+		       req->rq_export->exp_conn_cnt);
+		req->rq_no_reply = 1;
+		GOTO(out, rc = -ESTALE);
+	}
 
 	/* try to release in-memory reply data */
-	if (tgt_is_multimodrpcs_client(req->rq_export)) {
-		tgt_handle_received_xid(req->rq_export,
-				lustre_msg_get_last_xid(req->rq_reqmsg));
-		if (!(lustre_msg_get_flags(req->rq_reqmsg) &
-		      (MSG_RESENT | MSG_REPLAY)))
-			tgt_handle_tag(req->rq_export,
-				       lustre_msg_get_tag(req->rq_reqmsg));
+	if (tgt_is_multimodrpcs_client(exp)) {
+		tgt_handle_received_xid(exp, last_xid);
+		rc = tgt_handle_tag(req);
 	}
-	RETURN(0);
+
+out:
+	if (need_lock)
+		mutex_unlock(&ted->ted_lcd_lock);
+
+	RETURN(rc);
 }
 
 int tgt_request_handle(struct ptlrpc_request *req)
@@ -704,18 +736,12 @@ int tgt_request_handle(struct ptlrpc_request *req)
 		if (cfs_fail_val == 0 &&
 		    lustre_msg_get_opc(msg) != OBD_PING &&
 		    lustre_msg_get_flags(msg) & MSG_REQ_REPLAY_DONE) {
-			struct l_wait_info lwi =  { 0 };
-
 			cfs_fail_val = 1;
 			cfs_race_state = 0;
-			l_wait_event(cfs_race_waitq, (cfs_race_state == 1),
-				     &lwi);
+			wait_event_idle(cfs_race_waitq, (cfs_race_state == 1));
 		}
 	}
 
-	/* Refill the context, to make sure all thread keys are allocated */
-	lu_env_refill(req->rq_svc_thread->t_env);
-
 	req_capsule_init(&req->rq_pill, req, RCL_SERVER);
 	tsi->tsi_pill = &req->rq_pill;
 	tsi->tsi_env = req->rq_svc_thread->t_env;
@@ -761,7 +787,7 @@ int tgt_request_handle(struct ptlrpc_request *req)
 		tsi->tsi_jobid = NULL;
 
 	if (tgt == NULL) {
-		DEBUG_REQ(D_ERROR, req, "%s: No target for connected export\n",
+		DEBUG_REQ(D_ERROR, req, "%s: No target for connected export",
 			  class_exp2obd(req->rq_export)->obd_name);
 		req->rq_status = -EINVAL;
 		rc = ptlrpc_error(req);
@@ -777,7 +803,7 @@ int tgt_request_handle(struct ptlrpc_request *req)
 		/* reset the exp_last_xid on each connection. */
 		req->rq_export->exp_last_xid = 0;
 	} else if (obd->obd_recovery_data.trd_processing_task !=
-		   current_pid()) {
+		   current->pid) {
 		rc = process_req_last_xid(req);
 		if (rc) {
 			req->rq_status = rc;
@@ -799,14 +825,16 @@ int tgt_request_handle(struct ptlrpc_request *req)
 	LASSERTF(h->th_opc == opc, "opcode mismatch %d != %d\n",
 		 h->th_opc, opc);
 
-	if (CFS_FAIL_CHECK_ORSET(request_fail_id, CFS_FAIL_ONCE))
+	if ((cfs_fail_val == 0 || cfs_fail_val == opc) &&
+	     CFS_FAIL_CHECK_ORSET(request_fail_id, CFS_FAIL_ONCE))
 		GOTO(out, rc = 0);
 
 	rc = lustre_msg_check_version(msg, h->th_version);
 	if (unlikely(rc)) {
-		DEBUG_REQ(D_ERROR, req, "%s: drop mal-formed request, version"
-			  " %08x, expecting %08x\n", tgt_name(tgt),
-			  lustre_msg_get_version(msg), h->th_version);
+		DEBUG_REQ(D_ERROR, req,
+			  "%s: drop malformed request version=%08x expect=%08x",
+			  tgt_name(tgt), lustre_msg_get_version(msg),
+			  h->th_version);
 		req->rq_status = -EINVAL;
 		rc = ptlrpc_error(req);
 		GOTO(out, rc);
@@ -878,9 +906,9 @@ EXPORT_SYMBOL(tgt_counter_incr);
 
 int tgt_connect_check_sptlrpc(struct ptlrpc_request *req, struct obd_export *exp)
 {
-	struct lu_target	*tgt = class_exp2tgt(exp);
-	struct sptlrpc_flavor	 flvr;
-	int			 rc = 0;
+	struct lu_target *tgt = class_exp2tgt(exp);
+	struct sptlrpc_flavor flvr;
+	int rc = 0;
 
 	LASSERT(tgt);
 	LASSERT(tgt->lut_obd);
@@ -905,13 +933,13 @@ int tgt_connect_check_sptlrpc(struct ptlrpc_request *req, struct obd_export *exp
 		exp->exp_sp_peer = req->rq_sp_from;
 		exp->exp_flvr = flvr;
 
-		/* when on mgs, if no restriction is set, or if client
-		 * is loopback, allow any flavor */
+		/* when on mgs, if no restriction is set, or if the client
+		 * NID is on the local node, allow any flavor
+		 */
 		if ((strcmp(exp->exp_obd->obd_type->typ_name,
 			   LUSTRE_MGS_NAME) == 0) &&
 		     (exp->exp_flvr.sf_rpc == SPTLRPC_FLVR_NULL ||
-		      LNET_NETTYP(LNET_NIDNET(exp->exp_connection->c_peer.nid))
-		      == LOLND))
+		      LNetIsPeerLocal(exp->exp_connection->c_peer.nid)))
 			exp->exp_flvr.sf_rpc = SPTLRPC_FLVR_ANY;
 
 		if (exp->exp_flvr.sf_rpc != SPTLRPC_FLVR_ANY &&
@@ -945,7 +973,7 @@ int tgt_adapt_sptlrpc_conf(struct lu_target *tgt)
 	int			 rc;
 
 	if (unlikely(tgt == NULL)) {
-		CERROR("No target passed");
+		CERROR("No target passed\n");
 		return -EINVAL;
 	}
 
@@ -991,9 +1019,40 @@ int tgt_connect(struct tgt_session_info *tsi)
 	reply = req_capsule_server_get(tsi->tsi_pill, &RMF_CONNECT_DATA);
 	spin_lock(&tsi->tsi_exp->exp_lock);
 	*exp_connect_flags_ptr(tsi->tsi_exp) = reply->ocd_connect_flags;
+	if (reply->ocd_connect_flags & OBD_CONNECT_FLAGS2)
+		*exp_connect_flags2_ptr(tsi->tsi_exp) =
+			reply->ocd_connect_flags2;
 	tsi->tsi_exp->exp_connect_data.ocd_brw_size = reply->ocd_brw_size;
 	spin_unlock(&tsi->tsi_exp->exp_lock);
 
+	if (strcmp(tsi->tsi_exp->exp_obd->obd_type->typ_name,
+		   LUSTRE_MDT_NAME) == 0) {
+		rc = req_check_sepol(tsi->tsi_pill);
+		if (rc)
+			GOTO(out, rc);
+
+		if (reply->ocd_connect_flags & OBD_CONNECT_FLAGS2 &&
+		    reply->ocd_connect_flags2 & OBD_CONNECT2_ENCRYPT &&
+		    tsi->tsi_pill->rc_req->rq_export) {
+			bool forbid_encrypt = true;
+			struct lu_nodemap *nm =
+			 nodemap_get_from_exp(tsi->tsi_pill->rc_req->rq_export);
+
+			if (!nm) {
+				/* nodemap_get_from_exp returns NULL in case
+				 * nodemap is not active, so we do not forbid
+				 */
+				forbid_encrypt = false;
+			} else if (!IS_ERR(nm)) {
+				forbid_encrypt = nm->nmf_forbid_encryption;
+				nodemap_putref(nm);
+			}
+
+			if (forbid_encrypt)
+				GOTO(out, rc = -EACCES);
+		}
+	}
+
 	RETURN(0);
 out:
 	obd_disconnect(class_export_get(tsi->tsi_exp));
@@ -1007,6 +1066,8 @@ int tgt_disconnect(struct tgt_session_info *tsi)
 
 	ENTRY;
 
+	OBD_FAIL_TIMEOUT(OBD_FAIL_OST_DISCONNECT_DELAY, cfs_fail_val);
+
 	rc = target_handle_disconnect(tgt_ses_req(tsi));
 	if (rc)
 		RETURN(err_serious(rc));
@@ -1024,7 +1085,16 @@ int tgt_obd_ping(struct tgt_session_info *tsi)
 
 	ENTRY;
 
-	rc = target_handle_ping(tgt_ses_req(tsi));
+	/* The target-specific part of OBD_PING request handling.
+	 * It controls Filter Modification Data (FMD) expiration each time
+	 * PING is received.
+	 *
+	 * Valid only for replayable targets, e.g. MDT and OFD
+	 */
+	if (tsi->tsi_exp->exp_obd->obd_replayable)
+		tgt_fmd_expire(tsi->tsi_exp);
+
+	rc = req_capsule_server_pack(tsi->tsi_pill);
 	if (rc)
 		RETURN(err_serious(rc));
 
@@ -1039,19 +1109,27 @@ int tgt_obd_log_cancel(struct tgt_session_info *tsi)
 
 int tgt_send_buffer(struct tgt_session_info *tsi, struct lu_rdbuf *rdbuf)
 {
-	struct tgt_thread_info	*tti = tgt_th_info(tsi->tsi_env);
 	struct ptlrpc_request	*req = tgt_ses_req(tsi);
 	struct obd_export	*exp = req->rq_export;
 	struct ptlrpc_bulk_desc	*desc;
-	struct l_wait_info	*lwi = &tti->tti_u.update.tti_wait_info;
 	int			 i;
 	int			 rc;
+	int			 pages = 0;
 
 	ENTRY;
 
-	desc = ptlrpc_prep_bulk_exp(req, rdbuf->rb_nbufs, 1,
-				  PTLRPC_BULK_PUT_SOURCE | PTLRPC_BULK_BUF_KVEC,
-				    MDS_BULK_PORTAL, &ptlrpc_bulk_kvec_ops);
+	for (i = 0; i < rdbuf->rb_nbufs; i++) {
+		unsigned int offset;
+
+		offset = (unsigned long)rdbuf->rb_bufs[i].lb_buf & ~PAGE_MASK;
+		pages += DIV_ROUND_UP(rdbuf->rb_bufs[i].lb_len + offset,
+				      PAGE_SIZE);
+	}
+
+	desc = ptlrpc_prep_bulk_exp(req, pages, 1,
+				  PTLRPC_BULK_PUT_SOURCE,
+				    MDS_BULK_PORTAL,
+				    &ptlrpc_bulk_kiov_nopin_ops);
 	if (desc == NULL)
 		RETURN(-ENOMEM);
 
@@ -1060,7 +1138,7 @@ int tgt_send_buffer(struct tgt_session_info *tsi, struct lu_rdbuf *rdbuf)
 					rdbuf->rb_bufs[i].lb_buf,
 					rdbuf->rb_bufs[i].lb_len);
 
-	rc = target_bulk_io(exp, desc, lwi);
+	rc = target_bulk_io(exp, desc);
 	ptlrpc_free_bulk(desc);
 	RETURN(rc);
 }
@@ -1068,11 +1146,9 @@ EXPORT_SYMBOL(tgt_send_buffer);
 
 int tgt_sendpage(struct tgt_session_info *tsi, struct lu_rdpg *rdpg, int nob)
 {
-	struct tgt_thread_info	*tti = tgt_th_info(tsi->tsi_env);
 	struct ptlrpc_request	*req = tgt_ses_req(tsi);
 	struct obd_export	*exp = req->rq_export;
 	struct ptlrpc_bulk_desc	*desc;
-	struct l_wait_info	*lwi = &tti->tti_u.rdpg.tti_wait_info;
 	int			 tmpcount;
 	int			 tmpsize;
 	int			 i;
@@ -1081,8 +1157,7 @@ int tgt_sendpage(struct tgt_session_info *tsi, struct lu_rdpg *rdpg, int nob)
 	ENTRY;
 
 	desc = ptlrpc_prep_bulk_exp(req, rdpg->rp_npages, 1,
-				    PTLRPC_BULK_PUT_SOURCE |
-					PTLRPC_BULK_BUF_KIOV,
+				    PTLRPC_BULK_PUT_SOURCE,
 				    MDS_BULK_PORTAL,
 				    &ptlrpc_bulk_kiov_pin_ops);
 	if (desc == NULL)
@@ -1101,7 +1176,7 @@ int tgt_sendpage(struct tgt_session_info *tsi, struct lu_rdpg *rdpg, int nob)
 	}
 
 	LASSERT(desc->bd_nob == nob);
-	rc = target_bulk_io(exp, desc, lwi);
+	rc = target_bulk_io(exp, desc);
 	ptlrpc_free_bulk(desc);
 	RETURN(rc);
 }
@@ -1159,7 +1234,7 @@ static int tgt_obd_idx_read(struct tgt_session_info *tsi)
 	rdpg->rp_npages = (rdpg->rp_count + PAGE_SIZE - 1) >> PAGE_SHIFT;
 
 	/* allocate pages to store the containers */
-	OBD_ALLOC(rdpg->rp_pages, rdpg->rp_npages * sizeof(rdpg->rp_pages[0]));
+	OBD_ALLOC_PTR_ARRAY(rdpg->rp_pages, rdpg->rp_npages);
 	if (rdpg->rp_pages == NULL)
 		GOTO(out, rc = -ENOMEM);
 	for (i = 0; i < rdpg->rp_npages; i++) {
@@ -1186,8 +1261,7 @@ out:
 		for (i = 0; i < rdpg->rp_npages; i++)
 			if (rdpg->rp_pages[i])
 				__free_page(rdpg->rp_pages[i]);
-		OBD_FREE(rdpg->rp_pages,
-			 rdpg->rp_npages * sizeof(rdpg->rp_pages[0]));
+		OBD_FREE_PTR_ARRAY(rdpg->rp_pages, rdpg->rp_npages);
 	}
 	return rc;
 }
@@ -1257,8 +1331,8 @@ static int tgt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
 
 	if (flag == LDLM_CB_CANCELING &&
 	    (lock->l_granted_mode & (LCK_EX | LCK_PW | LCK_GROUP)) &&
-	    (tgt->lut_sync_lock_cancel == ALWAYS_SYNC_ON_CANCEL ||
-	     (tgt->lut_sync_lock_cancel == BLOCKING_SYNC_ON_CANCEL &&
+	    (tgt->lut_sync_lock_cancel == SYNC_LOCK_CANCEL_ALWAYS ||
+	     (tgt->lut_sync_lock_cancel == SYNC_LOCK_CANCEL_BLOCKING &&
 	      ldlm_is_cbpending(lock))) &&
 	    ((exp_connect_flags(lock->l_export) & OBD_CONNECT_MDS_MDS) ||
 	     lock->l_resource->lr_type == LDLM_EXTENT)) {
@@ -1267,7 +1341,7 @@ static int tgt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
 
 		rc = lu_env_init(&env, LCT_DT_THREAD);
 		if (unlikely(rc != 0))
-			RETURN(rc);
+			GOTO(err, rc);
 
 		ost_fid_from_resid(&fid, &lock->l_resource->lr_name,
 				   tgt->lut_lsd.lsd_osd_index);
@@ -1298,7 +1372,7 @@ err_put:
 err_env:
 		lu_env_fini(&env);
 	}
-
+err:
 	rc = ldlm_server_blocking_ast(lock, desc, data, flag);
 	RETURN(rc);
 }
@@ -1369,10 +1443,10 @@ int tgt_cp_callback(struct tgt_session_info *tsi)
 
 /* generic LDLM target handler */
 struct tgt_handler tgt_dlm_handlers[] = {
-TGT_DLM_HDL    (HABEO_CLAVIS,	LDLM_ENQUEUE,		tgt_enqueue),
-TGT_DLM_HDL    (HABEO_CLAVIS,	LDLM_CONVERT,		tgt_convert),
-TGT_DLM_HDL_VAR(0,		LDLM_BL_CALLBACK,	tgt_bl_callback),
-TGT_DLM_HDL_VAR(0,		LDLM_CP_CALLBACK,	tgt_cp_callback)
+TGT_DLM_HDL(HAS_KEY, LDLM_ENQUEUE, tgt_enqueue),
+TGT_DLM_HDL(HAS_KEY, LDLM_CONVERT, tgt_convert),
+TGT_DLM_HDL_VAR(0, LDLM_BL_CALLBACK, tgt_bl_callback),
+TGT_DLM_HDL_VAR(0, LDLM_CP_CALLBACK, tgt_cp_callback)
 };
 EXPORT_SYMBOL(tgt_dlm_handlers);
 
@@ -1534,8 +1608,8 @@ static int tgt_handle_lfsck_query(struct tgt_session_info *tsi)
 }
 
 struct tgt_handler tgt_lfsck_handlers[] = {
-TGT_LFSCK_HDL(HABEO_REFERO,	LFSCK_NOTIFY,	tgt_handle_lfsck_notify),
-TGT_LFSCK_HDL(HABEO_REFERO,	LFSCK_QUERY,	tgt_handle_lfsck_query),
+TGT_LFSCK_HDL(HAS_REPLY,	LFSCK_NOTIFY,	tgt_handle_lfsck_notify),
+TGT_LFSCK_HDL(HAS_REPLY,	LFSCK_QUERY,	tgt_handle_lfsck_query),
 };
 EXPORT_SYMBOL(tgt_lfsck_handlers);
 
@@ -1601,7 +1675,7 @@ int tgt_mdt_data_lock(struct ldlm_namespace *ns, struct ldlm_res_id *res_id,
 	LASSERT(ns != NULL);
 	LASSERT(!lustre_handle_is_used(lh));
 
-	rc = ldlm_cli_enqueue_local(ns, res_id, LDLM_IBITS, &policy, mode,
+	rc = ldlm_cli_enqueue_local(NULL, ns, res_id, LDLM_IBITS, &policy, mode,
 				    flags, ldlm_blocking_ast,
 				    ldlm_completion_ast, ldlm_glimpse_ast,
 				    NULL, 0, LVB_T_NONE, NULL, lh);
@@ -1610,20 +1684,13 @@ int tgt_mdt_data_lock(struct ldlm_namespace *ns, struct ldlm_res_id *res_id,
 }
 EXPORT_SYMBOL(tgt_mdt_data_lock);
 
-void tgt_mdt_data_unlock(struct lustre_handle *lh, enum ldlm_mode mode)
-{
-	LASSERT(lustre_handle_is_used(lh));
-	ldlm_lock_decref(lh, mode);
-}
-EXPORT_SYMBOL(tgt_mdt_data_unlock);
-
 /**
  * Helper function for getting server side [start, start+count] DLM lock
  * if asked by client.
  */
-int tgt_extent_lock(struct ldlm_namespace *ns, struct ldlm_res_id *res_id,
-		    __u64 start, __u64 end, struct lustre_handle *lh,
-		    int mode, __u64 *flags)
+int tgt_extent_lock(const struct lu_env *env, struct ldlm_namespace *ns,
+		    struct ldlm_res_id *res_id, __u64 start, __u64 end,
+		    struct lustre_handle *lh, int mode, __u64 *flags)
 {
 	union ldlm_policy_data policy;
 	int rc;
@@ -1646,37 +1713,49 @@ int tgt_extent_lock(struct ldlm_namespace *ns, struct ldlm_res_id *res_id,
 	else
 		policy.l_extent.end = end | ~PAGE_MASK;
 
-	rc = ldlm_cli_enqueue_local(ns, res_id, LDLM_EXTENT, &policy, mode,
-				    flags, ldlm_blocking_ast,
+	rc = ldlm_cli_enqueue_local(env, ns, res_id, LDLM_EXTENT, &policy,
+				    mode, flags, ldlm_blocking_ast,
 				    ldlm_completion_ast, ldlm_glimpse_ast,
 				    NULL, 0, LVB_T_NONE, NULL, lh);
 	RETURN(rc == ELDLM_OK ? 0 : -EIO);
 }
 EXPORT_SYMBOL(tgt_extent_lock);
 
-void tgt_extent_unlock(struct lustre_handle *lh, enum ldlm_mode mode)
+static int tgt_data_lock(const struct lu_env *env, struct obd_export *exp,
+			 struct ldlm_res_id *res_id, __u64 start, __u64 end,
+			 struct lustre_handle *lh, enum ldlm_mode mode)
+{
+	struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
+	__u64 flags = 0;
+
+	/* MDT IO for data-on-mdt */
+	if (exp->exp_connect_data.ocd_connect_flags & OBD_CONNECT_IBITS)
+		return tgt_mdt_data_lock(ns, res_id, lh, mode, &flags);
+
+	return tgt_extent_lock(env, ns, res_id, start, end, lh, mode, &flags);
+}
+
+void tgt_data_unlock(struct lustre_handle *lh, enum ldlm_mode mode)
 {
 	LASSERT(lustre_handle_is_used(lh));
 	ldlm_lock_decref(lh, mode);
 }
-EXPORT_SYMBOL(tgt_extent_unlock);
+EXPORT_SYMBOL(tgt_data_unlock);
 
-static int tgt_brw_lock(struct obd_export *exp, struct ldlm_res_id *res_id,
-			struct obd_ioobj *obj, struct niobuf_remote *nb,
-			struct lustre_handle *lh, enum ldlm_mode mode)
+static int tgt_brw_lock(const struct lu_env *env, struct obd_export *exp,
+			struct ldlm_res_id *res_id, struct obd_ioobj *obj,
+			struct niobuf_remote *nb, struct lustre_handle *lh,
+			enum ldlm_mode mode)
 {
-	struct ldlm_namespace	*ns = exp->exp_obd->obd_namespace;
-	__u64			 flags = 0;
-	int			 nrbufs = obj->ioo_bufcnt;
-	int			 i;
-	int			 rc;
+	int nrbufs = obj->ioo_bufcnt;
+	int i;
 
 	ENTRY;
 
 	LASSERT(mode == LCK_PR || mode == LCK_PW);
 	LASSERT(!lustre_handle_is_used(lh));
 
-	if (ns->ns_obd->obd_recovering)
+	if (exp->exp_obd->obd_recovering)
 		RETURN(0);
 
 	if (nrbufs == 0 || !(nb[0].rnb_flags & OBD_BRW_SRVLOCK))
@@ -1686,46 +1765,42 @@ static int tgt_brw_lock(struct obd_export *exp, struct ldlm_res_id *res_id,
 		if (!(nb[i].rnb_flags & OBD_BRW_SRVLOCK))
 			RETURN(-EFAULT);
 
-	/* MDT IO for data-on-mdt */
-	if (exp->exp_connect_data.ocd_connect_flags & OBD_CONNECT_IBITS)
-		rc = tgt_mdt_data_lock(ns, res_id, lh, mode, &flags);
-	else
-		rc = tgt_extent_lock(ns, res_id, nb[0].rnb_offset,
-				     nb[nrbufs - 1].rnb_offset +
-				     nb[nrbufs - 1].rnb_len - 1,
-				     lh, mode, &flags);
-	RETURN(rc);
+	return tgt_data_lock(env, exp, res_id, nb[0].rnb_offset,
+			     nb[nrbufs - 1].rnb_offset +
+			     nb[nrbufs - 1].rnb_len - 1, lh, mode);
 }
 
-static void tgt_brw_unlock(struct obd_ioobj *obj, struct niobuf_remote *niob,
+static void tgt_brw_unlock(struct obd_export *exp, struct obd_ioobj *obj,
+			   struct niobuf_remote *niob,
 			   struct lustre_handle *lh, enum ldlm_mode mode)
 {
 	ENTRY;
 
 	LASSERT(mode == LCK_PR || mode == LCK_PW);
-	LASSERT((obj->ioo_bufcnt > 0 &&
-		 (niob[0].rnb_flags & OBD_BRW_SRVLOCK)) ==
+	LASSERT((!exp->exp_obd->obd_recovering && obj->ioo_bufcnt &&
+		 niob[0].rnb_flags & OBD_BRW_SRVLOCK) ==
 		lustre_handle_is_used(lh));
 
 	if (lustre_handle_is_used(lh))
-		tgt_extent_unlock(lh, mode);
+		tgt_data_unlock(lh, mode);
 	EXIT;
 }
+
 static int tgt_checksum_niobuf(struct lu_target *tgt,
 				 struct niobuf_local *local_nb, int npages,
 				 int opc, enum cksum_types cksum_type,
 				 __u32 *cksum)
 {
-	struct cfs_crypto_hash_desc	*hdesc;
+	struct ahash_request	       *req;
 	unsigned int			bufsize;
 	int				i, err;
 	unsigned char			cfs_alg = cksum_obd2cfs(cksum_type);
 
-	hdesc = cfs_crypto_hash_init(cfs_alg, NULL, 0);
-	if (IS_ERR(hdesc)) {
+	req = cfs_crypto_hash_init(cfs_alg, NULL, 0);
+	if (IS_ERR(req)) {
 		CERROR("%s: unable to initialize checksum hash %s\n",
 		       tgt_name(tgt), cfs_crypto_hash_name(cfs_alg));
-		return PTR_ERR(hdesc);
+		return PTR_ERR(req);
 	}
 
 	CDEBUG(D_INFO, "Checksum for algo %s\n", cfs_crypto_hash_name(cfs_alg));
@@ -1739,19 +1814,18 @@ static int tgt_checksum_niobuf(struct lu_target *tgt,
 			struct page *np = tgt_page_to_corrupt;
 
 			if (np) {
-				char *ptr = ll_kmap_atomic(local_nb[i].lnb_page,
-							KM_USER0);
+				char *ptr = kmap_atomic(local_nb[i].lnb_page);
 				char *ptr2 = page_address(np);
 
 				memcpy(ptr2 + off, ptr + off, len);
 				memcpy(ptr2 + off, "bad3", min(4, len));
-				ll_kunmap_atomic(ptr, KM_USER0);
+				kunmap_atomic(ptr);
 
 				/* LU-8376 to preserve original index for
 				 * display in dump_all_bulk_pages() */
 				np->index = i;
 
-				cfs_crypto_hash_update_page(hdesc, np, off,
+				cfs_crypto_hash_update_page(req, np, off,
 							    len);
 				continue;
 			} else {
@@ -1759,7 +1833,7 @@ static int tgt_checksum_niobuf(struct lu_target *tgt,
 				       tgt_name(tgt));
 			}
 		}
-		cfs_crypto_hash_update_page(hdesc, local_nb[i].lnb_page,
+		cfs_crypto_hash_update_page(req, local_nb[i].lnb_page,
 				  local_nb[i].lnb_page_offset & ~PAGE_MASK,
 				  local_nb[i].lnb_len);
 
@@ -1772,19 +1846,18 @@ static int tgt_checksum_niobuf(struct lu_target *tgt,
 			struct page *np = tgt_page_to_corrupt;
 
 			if (np) {
-				char *ptr = ll_kmap_atomic(local_nb[i].lnb_page,
-							KM_USER0);
+				char *ptr = kmap_atomic(local_nb[i].lnb_page);
 				char *ptr2 = page_address(np);
 
 				memcpy(ptr2 + off, ptr + off, len);
 				memcpy(ptr2 + off, "bad4", min(4, len));
-				ll_kunmap_atomic(ptr, KM_USER0);
+				kunmap_atomic(ptr);
 
 				/* LU-8376 to preserve original index for
 				 * display in dump_all_bulk_pages() */
 				np->index = i;
 
-				cfs_crypto_hash_update_page(hdesc, np, off,
+				cfs_crypto_hash_update_page(req, np, off,
 							    len);
 				continue;
 			} else {
@@ -1795,7 +1868,7 @@ static int tgt_checksum_niobuf(struct lu_target *tgt,
 	}
 
 	bufsize = sizeof(*cksum);
-	err = cfs_crypto_hash_final(hdesc, (unsigned char *)cksum, &bufsize);
+	err = cfs_crypto_hash_final(req, (unsigned char *)cksum, &bufsize);
 
 	return 0;
 }
@@ -1815,9 +1888,8 @@ static void dump_all_bulk_pages(struct obdo *oa, int count,
 	 * file/fid, not during the resends/retries. */
 	snprintf(dbgcksum_file_name, sizeof(dbgcksum_file_name),
 		 "%s-checksum_dump-ost-"DFID":[%llu-%llu]-%x-%x",
-		 (strncmp(libcfs_debug_file_path_arr, "NONE", 4) != 0 ?
-		  libcfs_debug_file_path_arr :
-		  LIBCFS_DEBUG_FILE_PATH_DEFAULT),
+		 (strncmp(libcfs_debug_file_path, "NONE", 4) != 0 ?
+		  libcfs_debug_file_path : LIBCFS_DEBUG_FILE_PATH_DEFAULT),
 		 oa->o_valid & OBD_MD_FLFID ? oa->o_parent_seq : (__u64)0,
 		 oa->o_valid & OBD_MD_FLFID ? oa->o_parent_oid : 0,
 		 oa->o_valid & OBD_MD_FLFID ? oa->o_parent_ver : 0,
@@ -1856,11 +1928,10 @@ static void dump_all_bulk_pages(struct obdo *oa, int count,
 		kunmap(local_nb[i].lnb_page);
 	}
 
-	rc = ll_vfs_fsync_range(filp, 0, LLONG_MAX, 1);
+	rc = vfs_fsync_range(filp, 0, LLONG_MAX, 1);
 	if (rc)
 		CERROR("%s: sync returns %d\n", dbgcksum_file_name, rc);
 	filp_close(filp, NULL);
-	return;
 }
 
 static int check_read_checksum(struct niobuf_local *local_nb, int npages,
@@ -1928,9 +1999,9 @@ static int tgt_pages2shortio(struct niobuf_local *local, int npages,
 		if (len > size)
 			return -EINVAL;
 
-		ptr = ll_kmap_atomic(local[i].lnb_page, KM_USER0);
-		memcpy(buf + off, ptr, len);
-		ll_kunmap_atomic(ptr, KM_USER0);
+		ptr = kmap_atomic(local[i].lnb_page);
+		memcpy(buf, ptr + off, len);
+		kunmap_atomic(ptr);
 		buf += len;
 		size -= len;
 	}
@@ -1944,9 +2015,10 @@ static int tgt_checksum_niobuf_t10pi(struct lu_target *tgt,
 				     int sector_size,
 				     u32 *check_sum)
 {
+	enum cksum_types t10_cksum_type = tgt->lut_dt_conf.ddp_t10_cksum_type;
 	unsigned char cfs_alg = cksum_obd2cfs(OBD_CKSUM_T10_TOP);
 	const char *obd_name = tgt->lut_obd->obd_name;
-	struct cfs_crypto_hash_desc *hdesc;
+	struct ahash_request *req;
 	unsigned int bufsize;
 	unsigned char *buffer;
 	struct page *__page;
@@ -1962,11 +2034,11 @@ static int tgt_checksum_niobuf_t10pi(struct lu_target *tgt,
 	if (__page == NULL)
 		return -ENOMEM;
 
-	hdesc = cfs_crypto_hash_init(cfs_alg, NULL, 0);
-	if (IS_ERR(hdesc)) {
+	req = cfs_crypto_hash_init(cfs_alg, NULL, 0);
+	if (IS_ERR(req)) {
 		CERROR("%s: unable to initialize checksum hash %s\n",
 		       tgt_name(tgt), cfs_crypto_hash_name(cfs_alg));
-		return PTR_ERR(hdesc);
+		return PTR_ERR(req);
 	}
 
 	buffer = kmap(__page);
@@ -1982,19 +2054,18 @@ static int tgt_checksum_niobuf_t10pi(struct lu_target *tgt,
 			struct page *np = tgt_page_to_corrupt;
 
 			if (np) {
-				char *ptr = ll_kmap_atomic(local_nb[i].lnb_page,
-							KM_USER0);
+				char *ptr = kmap_atomic(local_nb[i].lnb_page);
 				char *ptr2 = page_address(np);
 
 				memcpy(ptr2 + off, ptr + off, len);
 				memcpy(ptr2 + off, "bad3", min(4, len));
-				ll_kunmap_atomic(ptr, KM_USER0);
+				kunmap_atomic(ptr);
 
 				/* LU-8376 to preserve original index for
 				 * display in dump_all_bulk_pages() */
 				np->index = i;
 
-				cfs_crypto_hash_update_page(hdesc, np, off,
+				cfs_crypto_hash_update_page(req, np, off,
 							    len);
 				continue;
 			} else {
@@ -2007,18 +2078,50 @@ static int tgt_checksum_niobuf_t10pi(struct lu_target *tgt,
 		 * The left guard number should be able to hold checksums of a
 		 * whole page
 		 */
-		rc = obd_page_dif_generate_buffer(obd_name,
-			local_nb[i].lnb_page,
-			local_nb[i].lnb_page_offset & ~PAGE_MASK,
-			local_nb[i].lnb_len, guard_start + used_number,
-			guard_number - used_number, &used, sector_size,
-			fn);
-		if (rc)
-			break;
+		if (t10_cksum_type && opc == OST_READ &&
+		    local_nb[i].lnb_len == PAGE_SIZE &&
+		    local_nb[i].lnb_guard_disk) {
+			used = DIV_ROUND_UP(local_nb[i].lnb_len, sector_size);
+			if (used > (guard_number - used_number)) {
+				rc = -E2BIG;
+				break;
+			}
+			memcpy(guard_start + used_number,
+			       local_nb[i].lnb_guards,
+			       used * sizeof(*local_nb[i].lnb_guards));
+		} else {
+			rc = obd_page_dif_generate_buffer(obd_name,
+				local_nb[i].lnb_page,
+				local_nb[i].lnb_page_offset & ~PAGE_MASK,
+				local_nb[i].lnb_len, guard_start + used_number,
+				guard_number - used_number, &used, sector_size,
+				fn);
+			if (rc)
+				break;
+		}
+
+		LASSERT(used <= MAX_GUARD_NUMBER);
+		/*
+		 * If disk support T10PI checksum, copy guards to local_nb.
+		 * If the write is partial page, do not use the guards for bio
+		 * submission since the data might not be full-sector. The bio
+		 * guards will be generated later based on the full sectors. If
+		 * the sector size is 512B rather than 4 KB, or the page size
+		 * is larger than 4KB, this might drop some useful guards for
+		 * partial page write, but it will only add minimal extra time
+		 * of checksum calculation.
+		 */
+		if (t10_cksum_type && opc == OST_WRITE &&
+		    local_nb[i].lnb_len == PAGE_SIZE) {
+			local_nb[i].lnb_guard_rpc = 1;
+			memcpy(local_nb[i].lnb_guards,
+			       guard_start + used_number,
+			       used * sizeof(*local_nb[i].lnb_guards));
+		}
 
 		used_number += used;
 		if (used_number == guard_number) {
-			cfs_crypto_hash_update_page(hdesc, __page, 0,
+			cfs_crypto_hash_update_page(req, __page, 0,
 				used_number * sizeof(*guard_start));
 			used_number = 0;
 		}
@@ -2032,19 +2135,18 @@ static int tgt_checksum_niobuf_t10pi(struct lu_target *tgt,
 			struct page *np = tgt_page_to_corrupt;
 
 			if (np) {
-				char *ptr = ll_kmap_atomic(local_nb[i].lnb_page,
-							KM_USER0);
+				char *ptr = kmap_atomic(local_nb[i].lnb_page);
 				char *ptr2 = page_address(np);
 
 				memcpy(ptr2 + off, ptr + off, len);
 				memcpy(ptr2 + off, "bad4", min(4, len));
-				ll_kunmap_atomic(ptr, KM_USER0);
+				kunmap_atomic(ptr);
 
 				/* LU-8376 to preserve original index for
 				 * display in dump_all_bulk_pages() */
 				np->index = i;
 
-				cfs_crypto_hash_update_page(hdesc, np, off,
+				cfs_crypto_hash_update_page(req, np, off,
 							    len);
 				continue;
 			} else {
@@ -2058,11 +2160,11 @@ static int tgt_checksum_niobuf_t10pi(struct lu_target *tgt,
 		GOTO(out, rc);
 
 	if (used_number != 0)
-		cfs_crypto_hash_update_page(hdesc, __page, 0,
+		cfs_crypto_hash_update_page(req, __page, 0,
 			used_number * sizeof(*guard_start));
 
 	bufsize = sizeof(cksum);
-	rc = cfs_crypto_hash_final(hdesc, (unsigned char *)&cksum, &bufsize);
+	rc = cfs_crypto_hash_final(req, (unsigned char *)&cksum, &bufsize);
 
 	if (rc == 0)
 		*check_sum = cksum;
@@ -2102,7 +2204,6 @@ int tgt_brw_read(struct tgt_session_info *tsi)
 	struct niobuf_local	*local_nb;
 	struct obd_ioobj	*ioo;
 	struct ost_body		*body, *repbody;
-	struct l_wait_info	 lwi;
 	struct lustre_handle	 lockh = { 0 };
 	int			 npages, nob = 0, rc, i, no_reply = 0,
 				 npages_read;
@@ -2132,10 +2233,9 @@ int tgt_brw_read(struct tgt_session_info *tsi)
 	 * finish */
 	if (unlikely(atomic_read(&exp->exp_obd->obd_evict_inprogress))) {
 		/* We do not care how long it takes */
-		lwi = LWI_INTR(NULL, NULL);
-		rc = l_wait_event(exp->exp_obd->obd_evict_inprogress_waitq,
-			 !atomic_read(&exp->exp_obd->obd_evict_inprogress),
-			 &lwi);
+		wait_event_idle(
+			exp->exp_obd->obd_evict_inprogress_waitq,
+			!atomic_read(&exp->exp_obd->obd_evict_inprogress));
 	}
 
 	/* There must be big cache in current thread to process this request
@@ -2155,8 +2255,8 @@ int tgt_brw_read(struct tgt_session_info *tsi)
 
 	local_nb = tbc->local;
 
-	rc = tgt_brw_lock(exp, &tsi->tsi_resid, ioo, remote_nb, &lockh,
-			  LCK_PR);
+	rc = tgt_brw_lock(tsi->tsi_env, exp, &tsi->tsi_resid, ioo, remote_nb,
+			  &lockh, LCK_PR);
 	if (rc != 0)
 		RETURN(rc);
 
@@ -2183,12 +2283,12 @@ int tgt_brw_read(struct tgt_session_info *tsi)
 	if (rc != 0)
 		GOTO(out_lock, rc);
 
-	if (body->oa.o_flags & OBD_FL_SHORT_IO) {
+	if (body->oa.o_valid & OBD_MD_FLFLAGS &&
+	    body->oa.o_flags & OBD_FL_SHORT_IO) {
 		desc = NULL;
 	} else {
 		desc = ptlrpc_prep_bulk_exp(req, npages, ioobj_max_brw_get(ioo),
-					    PTLRPC_BULK_PUT_SOURCE |
-						PTLRPC_BULK_BUF_KIOV,
+					    PTLRPC_BULK_PUT_SOURCE,
 					    OST_BULK_PORTAL,
 					    &ptlrpc_bulk_kiov_nopin_ops);
 		if (desc == NULL)
@@ -2224,9 +2324,6 @@ int tgt_brw_read(struct tgt_session_info *tsi)
 			break;
 		}
 	}
-	if (OBD_FAIL_CHECK(OBD_FAIL_OST_READ_SIZE) &&
-	    nob != cfs_fail_val)
-		rc = -E2BIG;
 
 	if (body->oa.o_valid & OBD_MD_FLCKSUM) {
 		u32 flag = body->oa.o_valid & OBD_MD_FLFLAGS ?
@@ -2262,7 +2359,8 @@ int tgt_brw_read(struct tgt_session_info *tsi)
 	/* Check if client was evicted while we were doing i/o before touching
 	 * network */
 	if (rc == 0) {
-		if (body->oa.o_flags & OBD_FL_SHORT_IO) {
+		if (body->oa.o_valid & OBD_MD_FLFLAGS &&
+		    body->oa.o_flags & OBD_FL_SHORT_IO) {
 			unsigned char *short_io_buf;
 			int short_io_size;
 
@@ -2279,11 +2377,12 @@ int tgt_brw_read(struct tgt_session_info *tsi)
 						   RCL_SERVER);
 			rc = rc > 0 ? 0 : rc;
 		} else if (!CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2)) {
-			rc = target_bulk_io(exp, desc, &lwi);
+			rc = target_bulk_io(exp, desc);
 		}
 		no_reply = rc != 0;
 	} else {
-		if (body->oa.o_flags & OBD_FL_SHORT_IO)
+		if (body->oa.o_valid & OBD_MD_FLFLAGS &&
+		    body->oa.o_flags & OBD_FL_SHORT_IO)
 			req_capsule_shrink(&req->rq_pill, &RMF_SHORT_IO, 0,
 					   RCL_SERVER);
 	}
@@ -2293,7 +2392,7 @@ out_commitrw:
 	rc = obd_commitrw(tsi->tsi_env, OBD_BRW_READ, exp, &repbody->oa, 1, ioo,
 			  remote_nb, npages, local_nb, rc);
 out_lock:
-	tgt_brw_unlock(ioo, remote_nb, &lockh, LCK_PR);
+	tgt_brw_unlock(exp, ioo, remote_nb, &lockh, LCK_PR);
 
 	if (desc && !CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2))
 		ptlrpc_free_bulk(desc);
@@ -2317,15 +2416,10 @@ out_lock:
 	 * to reorder. */
 	if (unlikely(CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2)) &&
 	    desc) {
-		wait_queue_head_t	 waitq;
-		struct l_wait_info	 lwi1;
-
 		CDEBUG(D_INFO, "reorder BULK\n");
-		init_waitqueue_head(&waitq);
 
-		lwi1 = LWI_TIMEOUT_INTR(cfs_time_seconds(3), NULL, NULL, NULL);
-		l_wait_event(waitq, 0, &lwi1);
-		target_bulk_io(exp, desc, &lwi);
+		ssleep(3);
+		target_bulk_io(exp, desc);
 		ptlrpc_free_bulk(desc);
 	}
 
@@ -2334,7 +2428,7 @@ out_lock:
 EXPORT_SYMBOL(tgt_brw_read);
 
 static int tgt_shortio2pages(struct niobuf_local *local, int npages,
-			     unsigned char *buf, int size)
+			     unsigned char *buf, unsigned int size)
 {
 	int	i, off, len;
 	char	*ptr;
@@ -2348,11 +2442,11 @@ static int tgt_shortio2pages(struct niobuf_local *local, int npages,
 
 		CDEBUG(D_PAGE, "index %d offset = %d len = %d left = %d\n",
 		       i, off, len, size);
-		ptr = ll_kmap_atomic(local[i].lnb_page, KM_USER0);
+		ptr = kmap_atomic(local[i].lnb_page);
 		if (ptr == NULL)
 			return -EINVAL;
 		memcpy(ptr + off, buf, len < size ? len : size);
-		ll_kunmap_atomic(ptr, KM_USER0);
+		kunmap_atomic(ptr);
 		buf += len;
 		size -= len;
 	}
@@ -2415,7 +2509,6 @@ int tgt_brw_write(struct tgt_session_info *tsi)
 	struct niobuf_local	*local_nb;
 	struct obd_ioobj	*ioo;
 	struct ost_body		*body, *repbody;
-	struct l_wait_info	 lwi;
 	struct lustre_handle	 lockh = {0};
 	__u32			*rcs;
 	int			 objcount, niocount, npages;
@@ -2425,6 +2518,8 @@ int tgt_brw_write(struct tgt_session_info *tsi)
 	struct tgt_thread_big_cache *tbc = req->rq_svc_thread->t_data;
 	bool wait_sync = false;
 	const char *obd_name = exp->exp_obd->obd_name;
+	/* '1' for consistency with code that checks !mpflag to restore */
+	unsigned int mpflags = 1;
 
 	ENTRY;
 
@@ -2453,6 +2548,9 @@ int tgt_brw_write(struct tgt_session_info *tsi)
 	CFS_FAIL_TIMEOUT(OBD_FAIL_OST_BRW_PAUSE_BULK, cfs_fail_val > 0 ?
 			 cfs_fail_val : (obd_timeout + 1) / 4);
 
+	/* Delay write commit to show stale size information */
+	CFS_FAIL_TIMEOUT(OBD_FAIL_OSC_NO_SIZE_DATA, cfs_fail_val);
+
 	/* There must be big cache in current thread to process this request
 	 * if it is NULL then something went wrong and it wasn't allocated,
 	 * report -ENOMEM in that case */
@@ -2480,7 +2578,7 @@ int tgt_brw_write(struct tgt_session_info *tsi)
 
 	if ((remote_nb[0].rnb_flags & OBD_BRW_MEMALLOC) &&
 	    ptlrpc_connection_is_local(exp->exp_connection))
-		memory_pressure_set();
+		mpflags = memalloc_noreclaim_save();
 
 	req_capsule_set_size(&req->rq_pill, &RMF_RCS, RCL_SERVER,
 			     niocount * sizeof(*rcs));
@@ -2493,8 +2591,8 @@ int tgt_brw_write(struct tgt_session_info *tsi)
 
 	local_nb = tbc->local;
 
-	rc = tgt_brw_lock(exp, &tsi->tsi_resid, ioo, remote_nb, &lockh,
-			  LCK_PW);
+	rc = tgt_brw_lock(tsi->tsi_env, exp, &tsi->tsi_resid, ioo, remote_nb,
+			  &lockh, LCK_PW);
 	if (rc != 0)
 		GOTO(out, rc);
 
@@ -2531,8 +2629,9 @@ int tgt_brw_write(struct tgt_session_info *tsi)
 			objcount, ioo, remote_nb, &npages, local_nb);
 	if (rc < 0)
 		GOTO(out_lock, rc);
-	if (body->oa.o_flags & OBD_FL_SHORT_IO) {
-		int short_io_size;
+	if (body->oa.o_valid & OBD_MD_FLFLAGS &&
+	    body->oa.o_flags & OBD_FL_SHORT_IO) {
+		unsigned int short_io_size;
 		unsigned char *short_io_buf;
 
 		short_io_size = req_capsule_get_size(&req->rq_pill,
@@ -2549,8 +2648,7 @@ int tgt_brw_write(struct tgt_session_info *tsi)
 		desc = NULL;
 	} else {
 		desc = ptlrpc_prep_bulk_exp(req, npages, ioobj_max_brw_get(ioo),
-					    PTLRPC_BULK_GET_SINK |
-					    PTLRPC_BULK_BUF_KIOV,
+					    PTLRPC_BULK_GET_SINK,
 					    OST_BULK_PORTAL,
 					    &ptlrpc_bulk_kiov_nopin_ops);
 		if (desc == NULL)
@@ -2567,7 +2665,7 @@ int tgt_brw_write(struct tgt_session_info *tsi)
 		if (rc != 0)
 			GOTO(skip_transfer, rc);
 
-		rc = target_bulk_io(exp, desc, &lwi);
+		rc = target_bulk_io(exp, desc);
 	}
 
 	no_reply = rc != 0;
@@ -2608,6 +2706,8 @@ skip_transfer:
 		}
 	}
 
+	OBD_FAIL_TIMEOUT(OBD_FAIL_OST_BRW_PAUSE_BULK2, cfs_fail_val);
+
 out_commitrw:
 	/* Must commit after prep above in all cases */
 	rc = obd_commitrw(tsi->tsi_env, OBD_BRW_WRITE, exp, &repbody->oa,
@@ -2653,7 +2753,7 @@ out_commitrw:
 		ptlrpc_lprocfs_brw(req, nob);
 	}
 out_lock:
-	tgt_brw_unlock(ioo, remote_nb, &lockh, LCK_PW);
+	tgt_brw_unlock(exp, ioo, remote_nb, &lockh, LCK_PW);
 	if (desc)
 		ptlrpc_free_bulk(desc);
 out:
@@ -2668,20 +2768,89 @@ out:
 				      obd_uuid2str(&exp->exp_client_uuid),
 				      obd_export_nid2str(exp), rc);
 	}
-	memory_pressure_clr();
+
+	if (mpflags)
+		memalloc_noreclaim_restore(mpflags);
+
 	RETURN(rc);
 }
 EXPORT_SYMBOL(tgt_brw_write);
 
+/**
+ * Common request handler for OST_SEEK RPC.
+ *
+ * Unified request handling for OST_SEEK RPC.
+ * It takes object by its FID, does needed lseek and packs result
+ * into reply. Only SEEK_HOLE and SEEK_DATA are supported.
+ *
+ * \param[in] tsi	target session environment for this request
+ *
+ * \retval		0 if successful
+ * \retval		negative value on error
+ */
+int tgt_lseek(struct tgt_session_info *tsi)
+{
+	struct lustre_handle lh = { 0 };
+	struct dt_object *dob;
+	struct ost_body *repbody;
+	loff_t offset = tsi->tsi_ost_body->oa.o_size;
+	int whence = tsi->tsi_ost_body->oa.o_mode;
+	bool srvlock;
+	int rc = 0;
+
+	ENTRY;
+
+	if (whence != SEEK_HOLE && whence != SEEK_DATA)
+		RETURN(-EPROTO);
+
+	/* Negative offset is prohibited on wire and must be handled on client
+	 * prior sending RPC.
+	 */
+	if (offset < 0)
+		RETURN(-EPROTO);
+
+	repbody = req_capsule_server_get(tsi->tsi_pill, &RMF_OST_BODY);
+	if (repbody == NULL)
+		RETURN(-ENOMEM);
+	repbody->oa = tsi->tsi_ost_body->oa;
+
+	srvlock = tsi->tsi_ost_body->oa.o_valid & OBD_MD_FLFLAGS &&
+		  tsi->tsi_ost_body->oa.o_flags & OBD_FL_SRVLOCK;
+	if (srvlock) {
+		rc = tgt_data_lock(tsi->tsi_env, tsi->tsi_exp, &tsi->tsi_resid,
+				   offset, OBD_OBJECT_EOF, &lh, LCK_PR);
+		if (rc)
+			RETURN(rc);
+	}
+
+	dob = dt_locate(tsi->tsi_env, tsi->tsi_tgt->lut_bottom, &tsi->tsi_fid);
+	if (IS_ERR(dob))
+		GOTO(out, rc = PTR_ERR(dob));
+
+	if (!dt_object_exists(dob))
+		GOTO(obj_put, rc = -ENOENT);
+
+	repbody->oa.o_size = dt_lseek(tsi->tsi_env, dob, offset, whence);
+	rc = 0;
+obj_put:
+	dt_object_put(tsi->tsi_env, dob);
+out:
+	if (srvlock)
+		tgt_data_unlock(&lh, LCK_PR);
+
+	RETURN(rc);
+}
+EXPORT_SYMBOL(tgt_lseek);
+
 /* Check if request can be reconstructed from saved reply data
  * A copy of the reply data is returned in @trd if the pointer is not NULL
  */
-bool req_can_reconstruct(struct ptlrpc_request *req,
+int req_can_reconstruct(struct ptlrpc_request *req,
 			 struct tg_reply_data *trd)
 {
 	struct tg_export_data *ted = &req->rq_export->exp_target_data;
 	struct lsd_client_data *lcd = ted->ted_lcd;
-	bool found;
+	int found;
 
 	if (tgt_is_multimodrpcs_client(req->rq_export))
 		return tgt_lookup_reply(req, trd);