X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Ftarget%2Ftgt_handler.c;h=24e6936e9472b012a0b2d4febeb57a4a2d6cca71;hp=2fcb794b40aeb7f356c9f97ad601895f8dff1246;hb=c1d465de13ccf0eda8020c88661c3cc4d78538ca;hpb=e9389613eb29297f7b5a6b6cd896bafc7a5551f8

diff --git a/lustre/target/tgt_handler.c b/lustre/target/tgt_handler.c
index 2fcb794..24e6936 100644
--- a/lustre/target/tgt_handler.c
+++ b/lustre/target/tgt_handler.c
@@ -21,7 +21,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2013, 2016, Intel Corporation.
+ * Copyright (c) 2013, 2017, Intel Corporation.
  */
 /*
  * lustre/target/tgt_handler.c
@@ -44,6 +44,7 @@
 #include <obd_cksum.h>
 #include <lustre_lfsck.h>
 #include <lustre_nodemap.h>
+#include <lustre_acl.h>
 
 #include "tgt_internal.h"
 
@@ -342,10 +343,13 @@ static int tgt_request_preprocess(struct tgt_session_info *tsi,
 
 		dlm_req = req_capsule_client_get(pill, &RMF_DLM_REQ);
 		if (dlm_req != NULL) {
+			union ldlm_wire_policy_data *policy =
+					&dlm_req->lock_desc.l_policy_data;
+
 			if (unlikely(dlm_req->lock_desc.l_resource.lr_type ==
 				     LDLM_IBITS &&
-				     dlm_req->lock_desc.l_policy_data.\
-				     l_inodebits.bits == 0)) {
+				     (policy->l_inodebits.bits |
+				      policy->l_inodebits.try_bits) == 0)) {
 				/*
 				 * Lock without inodebits makes no sense and
 				 * will oops later in ldlm. If client miss to
@@ -425,6 +429,23 @@ static int tgt_handle_request0(struct tgt_session_info *tsi,
 					  RCL_SERVER))
 			req_capsule_set_size(tsi->tsi_pill, &RMF_LOGCOOKIES,
 					     RCL_SERVER, 0);
+		if (req_capsule_has_field(tsi->tsi_pill, &RMF_ACL, RCL_SERVER))
+			req_capsule_set_size(tsi->tsi_pill,
+					     &RMF_ACL, RCL_SERVER,
+					     LUSTRE_POSIX_ACL_MAX_SIZE_OLD);
+
+		if (req_capsule_has_field(tsi->tsi_pill, &RMF_SHORT_IO,
+					  RCL_SERVER)) {
+			struct niobuf_remote *remote_nb =
+				req_capsule_client_get(tsi->tsi_pill,
+						       &RMF_NIOBUF_REMOTE);
+			struct ost_body *body = tsi->tsi_ost_body;
+
+			req_capsule_set_size(tsi->tsi_pill, &RMF_SHORT_IO,
+					 RCL_SERVER,
+					 (body->oa.o_flags & OBD_FL_SHORT_IO) ?
+					 remote_nb[0].rnb_len : 0);
+		}
 
 		rc = req_capsule_server_pack(tsi->tsi_pill);
 	}
@@ -640,6 +661,19 @@ static int process_req_last_xid(struct ptlrpc_request *req)
 			RETURN(-EPROTO);
 	}
 
+	/* The "last_xid" is the minimum xid among unreplied requests,
+	 * if the request is from the previous connection, its xid can
+	 * still be larger than "exp_last_xid", then the above check of
+	 * xid is not enough to determine whether the request is delayed.
+	 *
+	 * For example, if some replay request was delayed and caused
+	 * timeout at client and the replay is restarted, the delayed
+	 * 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);
+
 	/* try to release in-memory reply data */
 	if (tgt_is_multimodrpcs_client(req->rq_export)) {
 		tgt_handle_received_xid(req->rq_export,
@@ -666,6 +700,19 @@ int tgt_request_handle(struct ptlrpc_request *req)
 	bool			 is_connect = false;
 	ENTRY;
 
+	if (unlikely(OBD_FAIL_CHECK(OBD_FAIL_TGT_RECOVERY_REQ_RACE))) {
+		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);
+		}
+	}
+
 	/* Refill the context, to make sure all thread keys are allocated */
 	lu_env_refill(req->rq_svc_thread->t_env);
 
@@ -1562,6 +1609,41 @@ void tgt_io_thread_done(struct ptlrpc_thread *thread)
 	EXIT;
 }
 EXPORT_SYMBOL(tgt_io_thread_done);
+
+/**
+ * Helper function for getting Data-on-MDT file server DLM lock
+ * if asked by client.
+ */
+int tgt_mdt_data_lock(struct ldlm_namespace *ns, struct ldlm_res_id *res_id,
+		      struct lustre_handle *lh, int mode, __u64 *flags)
+{
+	union ldlm_policy_data policy = {
+		.l_inodebits.bits = MDS_INODELOCK_DOM,
+	};
+	int rc;
+
+	ENTRY;
+
+	LASSERT(lh != NULL);
+	LASSERT(ns != NULL);
+	LASSERT(!lustre_handle_is_used(lh));
+
+	rc = ldlm_cli_enqueue_local(ns, res_id, LDLM_IBITS, &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_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.
@@ -1606,13 +1688,15 @@ void tgt_extent_unlock(struct lustre_handle *lh, enum ldlm_mode mode)
 }
 EXPORT_SYMBOL(tgt_extent_unlock);
 
-int tgt_brw_lock(struct ldlm_namespace *ns, 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(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;
 
 	ENTRY;
 
@@ -1629,14 +1713,19 @@ int tgt_brw_lock(struct ldlm_namespace *ns, struct ldlm_res_id *res_id,
 		if (!(nb[i].rnb_flags & OBD_BRW_SRVLOCK))
 			RETURN(-EFAULT);
 
-	RETURN(tgt_extent_lock(ns, res_id, nb[0].rnb_offset,
-			       nb[nrbufs - 1].rnb_offset +
-			       nb[nrbufs - 1].rnb_len - 1,
-			       lh, mode, &flags));
+	/* 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);
 }
 
-void tgt_brw_unlock(struct obd_ioobj *obj, struct niobuf_remote *niob,
-		    struct lustre_handle *lh, enum ldlm_mode mode)
+static void tgt_brw_unlock(struct obd_ioobj *obj, struct niobuf_remote *niob,
+			   struct lustre_handle *lh, enum ldlm_mode mode)
 {
 	ENTRY;
 
@@ -1649,18 +1738,15 @@ void tgt_brw_unlock(struct obd_ioobj *obj, struct niobuf_remote *niob,
 		tgt_extent_unlock(lh, mode);
 	EXIT;
 }
-
-static __u32 tgt_checksum_bulk(struct lu_target *tgt,
-			       struct ptlrpc_bulk_desc *desc, int opc,
-			       cksum_type_t cksum_type)
+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;
 	unsigned int			bufsize;
 	int				i, err;
 	unsigned char			cfs_alg = cksum_obd2cfs(cksum_type);
-	__u32				cksum;
-
-	LASSERT(ptlrpc_is_bulk_desc_kiov(desc->bd_type));
 
 	hdesc = cfs_crypto_hash_init(cfs_alg, NULL, 0);
 	if (IS_ERR(hdesc)) {
@@ -1670,53 +1756,64 @@ static __u32 tgt_checksum_bulk(struct lu_target *tgt,
 	}
 
 	CDEBUG(D_INFO, "Checksum for algo %s\n", cfs_crypto_hash_name(cfs_alg));
-	for (i = 0; i < desc->bd_iov_count; i++) {
+	for (i = 0; i < npages; i++) {
 		/* corrupt the data before we compute the checksum, to
 		 * simulate a client->OST data error */
 		if (i == 0 && opc == OST_WRITE &&
 		    OBD_FAIL_CHECK(OBD_FAIL_OST_CHECKSUM_RECEIVE)) {
-			int off = BD_GET_KIOV(desc, i).kiov_offset &
-				~PAGE_MASK;
-			int len = BD_GET_KIOV(desc, i).kiov_len;
+			int off = local_nb[i].lnb_page_offset & ~PAGE_MASK;
+			int len = local_nb[i].lnb_len;
 			struct page *np = tgt_page_to_corrupt;
-			char *ptr = kmap(BD_GET_KIOV(desc, i).kiov_page) + off;
 
 			if (np) {
-				char *ptr2 = kmap(np) + off;
+				char *ptr = ll_kmap_atomic(local_nb[i].lnb_page,
+							KM_USER0);
+				char *ptr2 = page_address(np);
 
-				memcpy(ptr2, ptr, len);
-				memcpy(ptr2, "bad3", min(4, len));
-				kunmap(np);
-				BD_GET_KIOV(desc, i).kiov_page = np;
+				memcpy(ptr2 + off, ptr + off, len);
+				memcpy(ptr2 + off, "bad3", min(4, len));
+				ll_kunmap_atomic(ptr, KM_USER0);
+
+				/* LU-8376 to preserve original index for
+				 * display in dump_all_bulk_pages() */
+				np->index = i;
+
+				cfs_crypto_hash_update_page(hdesc, np, off,
+							    len);
+				continue;
 			} else {
 				CERROR("%s: can't alloc page for corruption\n",
 				       tgt_name(tgt));
 			}
 		}
-		cfs_crypto_hash_update_page(hdesc,
-				  BD_GET_KIOV(desc, i).kiov_page,
-				  BD_GET_KIOV(desc, i).kiov_offset &
-					~PAGE_MASK,
-				  BD_GET_KIOV(desc, i).kiov_len);
+		cfs_crypto_hash_update_page(hdesc, local_nb[i].lnb_page,
+				  local_nb[i].lnb_page_offset & ~PAGE_MASK,
+				  local_nb[i].lnb_len);
 
 		 /* corrupt the data after we compute the checksum, to
 		 * simulate an OST->client data error */
 		if (i == 0 && opc == OST_READ &&
 		    OBD_FAIL_CHECK(OBD_FAIL_OST_CHECKSUM_SEND)) {
-			int off = BD_GET_KIOV(desc, i).kiov_offset
-			  & ~PAGE_MASK;
-			int len = BD_GET_KIOV(desc, i).kiov_len;
+			int off = local_nb[i].lnb_page_offset & ~PAGE_MASK;
+			int len = local_nb[i].lnb_len;
 			struct page *np = tgt_page_to_corrupt;
-			char *ptr =
-			  kmap(BD_GET_KIOV(desc, i).kiov_page) + off;
 
 			if (np) {
-				char *ptr2 = kmap(np) + off;
+				char *ptr = ll_kmap_atomic(local_nb[i].lnb_page,
+							KM_USER0);
+				char *ptr2 = page_address(np);
+
+				memcpy(ptr2 + off, ptr + off, len);
+				memcpy(ptr2 + off, "bad4", min(4, len));
+				ll_kunmap_atomic(ptr, KM_USER0);
+
+				/* LU-8376 to preserve original index for
+				 * display in dump_all_bulk_pages() */
+				np->index = i;
 
-				memcpy(ptr2, ptr, len);
-				memcpy(ptr2, "bad4", min(4, len));
-				kunmap(np);
-				BD_GET_KIOV(desc, i).kiov_page = np;
+				cfs_crypto_hash_update_page(hdesc, np, off,
+							    len);
+				continue;
 			} else {
 				CERROR("%s: can't alloc page for corruption\n",
 				       tgt_name(tgt));
@@ -1724,10 +1821,147 @@ static __u32 tgt_checksum_bulk(struct lu_target *tgt,
 		}
 	}
 
-	bufsize = sizeof(cksum);
-	err = cfs_crypto_hash_final(hdesc, (unsigned char *)&cksum, &bufsize);
+	bufsize = sizeof(*cksum);
+	err = cfs_crypto_hash_final(hdesc, (unsigned char *)cksum, &bufsize);
 
-	return cksum;
+	return 0;
+}
+
+char dbgcksum_file_name[PATH_MAX];
+
+static void dump_all_bulk_pages(struct obdo *oa, int count,
+				struct niobuf_local *local_nb,
+				__u32 server_cksum, __u32 client_cksum)
+{
+	struct file *filp;
+	int rc, i;
+	unsigned int len;
+	char *buf;
+
+	/* will only keep dump of pages on first error for the same range in
+	 * 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),
+		 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,
+		 local_nb[0].lnb_file_offset,
+		 local_nb[count-1].lnb_file_offset +
+		 local_nb[count-1].lnb_len - 1, client_cksum, server_cksum);
+	filp = filp_open(dbgcksum_file_name,
+			 O_CREAT | O_EXCL | O_WRONLY | O_LARGEFILE, 0600);
+	if (IS_ERR(filp)) {
+		rc = PTR_ERR(filp);
+		if (rc == -EEXIST)
+			CDEBUG(D_INFO, "%s: can't open to dump pages with "
+			       "checksum error: rc = %d\n", dbgcksum_file_name,
+			       rc);
+		else
+			CERROR("%s: can't open to dump pages with checksum "
+			       "error: rc = %d\n", dbgcksum_file_name, rc);
+		return;
+	}
+
+	for (i = 0; i < count; i++) {
+		len = local_nb[i].lnb_len;
+		buf = kmap(local_nb[i].lnb_page);
+		while (len != 0) {
+			rc = cfs_kernel_write(filp, buf, len, &filp->f_pos);
+			if (rc < 0) {
+				CERROR("%s: wanted to write %u but got %d "
+				       "error\n", dbgcksum_file_name, len, rc);
+				break;
+			}
+			len -= rc;
+			buf += rc;
+			CDEBUG(D_INFO, "%s: wrote %d bytes\n",
+			       dbgcksum_file_name, rc);
+		}
+		kunmap(local_nb[i].lnb_page);
+	}
+
+	rc = ll_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,
+			       struct obd_export *exp, struct obdo *oa,
+			       const struct lnet_process_id *peer,
+			       __u32 client_cksum, __u32 server_cksum,
+			       enum cksum_types server_cksum_type)
+{
+	char *msg;
+	enum cksum_types cksum_type;
+	loff_t start, end;
+
+	/* unlikely to happen and only if resend does not occur due to cksum
+	 * control failure on Client */
+	if (unlikely(server_cksum == client_cksum)) {
+		CDEBUG(D_PAGE, "checksum %x confirmed upon retry\n",
+		       client_cksum);
+		return 0;
+	}
+
+	if (exp->exp_obd->obd_checksum_dump)
+		dump_all_bulk_pages(oa, npages, local_nb, server_cksum,
+				    client_cksum);
+
+	cksum_type = cksum_type_unpack(oa->o_valid & OBD_MD_FLFLAGS ?
+				       oa->o_flags : 0);
+
+	if (cksum_type != server_cksum_type)
+		msg = "the server may have not used the checksum type specified"
+		      " in the original request - likely a protocol problem";
+	else
+		msg = "should have changed on the client or in transit";
+
+	start = local_nb[0].lnb_file_offset;
+	end = local_nb[npages-1].lnb_file_offset +
+					local_nb[npages-1].lnb_len - 1;
+
+	LCONSOLE_ERROR_MSG(0x132, "%s: BAD READ CHECKSUM: %s: from %s inode "
+		DFID " object "DOSTID" extent [%llu-%llu], client returned csum"
+		" %x (type %x), server csum %x (type %x)\n",
+		exp->exp_obd->obd_name,
+		msg, libcfs_nid2str(peer->nid),
+		oa->o_valid & OBD_MD_FLFID ? oa->o_parent_seq : 0ULL,
+		oa->o_valid & OBD_MD_FLFID ? oa->o_parent_oid : 0,
+		oa->o_valid & OBD_MD_FLFID ? oa->o_parent_ver : 0,
+		POSTID(&oa->o_oi),
+		start, end, client_cksum, cksum_type, server_cksum,
+		server_cksum_type);
+
+	return 1;
+}
+
+static int tgt_pages2shortio(struct niobuf_local *local, int npages,
+			     unsigned char *buf, int size)
+{
+	int	i, off, len, copied = size;
+	char	*ptr;
+
+	for (i = 0; i < npages; i++) {
+		off = local[i].lnb_page_offset & ~PAGE_MASK;
+		len = local[i].lnb_len;
+
+		CDEBUG(D_PAGE, "index %d offset = %d len = %d left = %d\n",
+		       i, off, len, size);
+		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);
+		buf += len;
+		size -= len;
+	}
+	return copied - size;
 }
 
 int tgt_brw_read(struct tgt_session_info *tsi)
@@ -1741,12 +1975,14 @@ int tgt_brw_read(struct tgt_session_info *tsi)
 	struct ost_body		*body, *repbody;
 	struct l_wait_info	 lwi;
 	struct lustre_handle	 lockh = { 0 };
-	int			 npages, nob = 0, rc, i, no_reply = 0;
+	int			 npages, nob = 0, rc, i, no_reply = 0,
+				 npages_read;
 	struct tgt_thread_big_cache *tbc = req->rq_svc_thread->t_data;
 
 	ENTRY;
 
-	if (ptlrpc_req2svc(req)->srv_req_portal != OST_IO_PORTAL) {
+	if (ptlrpc_req2svc(req)->srv_req_portal != OST_IO_PORTAL &&
+	    ptlrpc_req2svc(req)->srv_req_portal != MDS_IO_PORTAL) {
 		CERROR("%s: deny read request from %s to portal %u\n",
 		       tgt_name(tsi->tsi_tgt),
 		       obd_export_nid2str(req->rq_export),
@@ -1789,8 +2025,8 @@ int tgt_brw_read(struct tgt_session_info *tsi)
 
 	local_nb = tbc->local;
 
-	rc = tgt_brw_lock(exp->exp_obd->obd_namespace, &tsi->tsi_resid, ioo,
-			  remote_nb, &lockh, LCK_PR);
+	rc = tgt_brw_lock(exp, &tsi->tsi_resid, ioo, remote_nb, &lockh,
+			  LCK_PR);
 	if (rc != 0)
 		RETURN(rc);
 
@@ -1798,13 +2034,12 @@ int tgt_brw_read(struct tgt_session_info *tsi)
 	 * If getting the lock took more time than
 	 * client was willing to wait, drop it. b=11330
 	 */
-	if (cfs_time_current_sec() > req->rq_deadline ||
+	if (ktime_get_real_seconds() > req->rq_deadline ||
 	    OBD_FAIL_CHECK(OBD_FAIL_OST_DROP_REQ)) {
 		no_reply = 1;
-		CERROR("Dropping timed-out read from %s because locking"
-		       "object "DOSTID" took %ld seconds (limit was %ld).\n",
+		CERROR("Dropping timed-out read from %s because locking object " DOSTID " took %lld seconds (limit was %lld).\n",
 		       libcfs_id2str(req->rq_peer), POSTID(&ioo->ioo_oid),
-		       cfs_time_current_sec() - req->rq_arrival_time.tv_sec,
+		       ktime_get_real_seconds() - req->rq_arrival_time.tv_sec,
 		       req->rq_deadline - req->rq_arrival_time.tv_sec);
 		GOTO(out_lock, rc = -ETIMEDOUT);
 	}
@@ -1818,33 +2053,41 @@ int tgt_brw_read(struct tgt_session_info *tsi)
 	if (rc != 0)
 		GOTO(out_lock, rc);
 
-	desc = ptlrpc_prep_bulk_exp(req, npages, ioobj_max_brw_get(ioo),
-				    PTLRPC_BULK_PUT_SOURCE |
-					PTLRPC_BULK_BUF_KIOV,
-				    OST_BULK_PORTAL,
-				    &ptlrpc_bulk_kiov_nopin_ops);
-	if (desc == NULL)
-		GOTO(out_commitrw, rc = -ENOMEM);
+	if (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,
+					    OST_BULK_PORTAL,
+					    &ptlrpc_bulk_kiov_nopin_ops);
+		if (desc == NULL)
+			GOTO(out_commitrw, rc = -ENOMEM);
+	}
 
 	nob = 0;
+	npages_read = npages;
 	for (i = 0; i < npages; i++) {
 		int page_rc = local_nb[i].lnb_rc;
 
 		if (page_rc < 0) {
 			rc = page_rc;
+			npages_read = i;
 			break;
 		}
 
 		nob += page_rc;
-		if (page_rc != 0) { /* some data! */
+		if (page_rc != 0 && desc != NULL) { /* some data! */
 			LASSERT(local_nb[i].lnb_page != NULL);
 			desc->bd_frag_ops->add_kiov_frag
 			  (desc, local_nb[i].lnb_page,
-			   local_nb[i].lnb_page_offset,
+			   local_nb[i].lnb_page_offset & ~PAGE_MASK,
 			   page_rc);
 		}
 
 		if (page_rc != local_nb[i].lnb_len) { /* short read */
+			local_nb[i].lnb_len = page_rc;
+			npages_read = i + (page_rc != 0 ? 1 : 0);
 			/* All subsequent pages should be 0 */
 			while (++i < npages)
 				LASSERT(local_nb[i].lnb_rc == 0);
@@ -1856,15 +2099,30 @@ int tgt_brw_read(struct tgt_session_info *tsi)
 		rc = -E2BIG;
 
 	if (body->oa.o_valid & OBD_MD_FLCKSUM) {
-		cksum_type_t cksum_type =
+		enum cksum_types cksum_type =
 			cksum_type_unpack(body->oa.o_valid & OBD_MD_FLFLAGS ?
 					  body->oa.o_flags : 0);
+
 		repbody->oa.o_flags = cksum_type_pack(cksum_type);
 		repbody->oa.o_valid = OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
-		repbody->oa.o_cksum = tgt_checksum_bulk(tsi->tsi_tgt, desc,
-							OST_READ, cksum_type);
+		rc = tgt_checksum_niobuf(tsi->tsi_tgt, local_nb,
+					 npages_read, OST_READ, cksum_type,
+					 &repbody->oa.o_cksum);
+		if (rc < 0)
+			GOTO(out_commitrw, rc);
+
 		CDEBUG(D_PAGE, "checksum at read origin: %x\n",
 		       repbody->oa.o_cksum);
+
+		/* if a resend it could be for a cksum error, so check Server
+		 * cksum with returned Client cksum (this should even cover
+		 * zero-cksum case) */
+		if ((body->oa.o_valid & OBD_MD_FLFLAGS) &&
+		    (body->oa.o_flags & OBD_FL_RECOV_RESEND))
+			check_read_checksum(local_nb, npages_read, exp,
+					    &body->oa, &req->rq_peer,
+					    body->oa.o_cksum,
+					    repbody->oa.o_cksum, cksum_type);
 	} else {
 		repbody->oa.o_valid = 0;
 	}
@@ -1872,11 +2130,31 @@ int tgt_brw_read(struct tgt_session_info *tsi)
 
 	/* Check if client was evicted while we were doing i/o before touching
 	 * network */
-	if (likely(rc == 0 &&
-		   !CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2) &&
-		   !CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_DROP_BULK))) {
-		rc = target_bulk_io(exp, desc, &lwi);
+	if (rc == 0) {
+		if (body->oa.o_flags & OBD_FL_SHORT_IO) {
+			unsigned char *short_io_buf;
+			int short_io_size;
+
+			short_io_buf = req_capsule_server_get(&req->rq_pill,
+							      &RMF_SHORT_IO);
+			short_io_size = req_capsule_get_size(&req->rq_pill,
+							     &RMF_SHORT_IO,
+							     RCL_SERVER);
+			rc = tgt_pages2shortio(local_nb, npages_read,
+					       short_io_buf, short_io_size);
+			if (rc >= 0)
+				req_capsule_shrink(&req->rq_pill,
+						   &RMF_SHORT_IO, rc,
+						   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);
+		}
 		no_reply = rc != 0;
+	} else {
+		if (body->oa.o_flags & OBD_FL_SHORT_IO)
+			req_capsule_shrink(&req->rq_pill, &RMF_SHORT_IO, 0,
+					   RCL_SERVER);
 	}
 
 out_commitrw:
@@ -1904,8 +2182,10 @@ out_lock:
 			      obd_export_nid2str(exp), rc);
 	}
 	/* send a bulk after reply to simulate a network delay or reordering
-	 * by a router */
-	if (unlikely(CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2))) {
+	 * by a router - Note that !desc implies short io, so there is no bulk
+	 * to reorder. */
+	if (unlikely(CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2)) &&
+	    desc) {
 		wait_queue_head_t	 waitq;
 		struct l_wait_info	 lwi1;
 
@@ -1922,6 +2202,32 @@ out_lock:
 }
 EXPORT_SYMBOL(tgt_brw_read);
 
+static int tgt_shortio2pages(struct niobuf_local *local, int npages,
+			     unsigned char *buf, int size)
+{
+	int	i, off, len;
+	char	*ptr;
+
+	for (i = 0; i < npages; i++) {
+		off = local[i].lnb_page_offset & ~PAGE_MASK;
+		len = local[i].lnb_len;
+
+		if (len == 0)
+			continue;
+
+		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);
+		if (ptr == NULL)
+			return -EINVAL;
+		memcpy(ptr + off, buf, len < size ? len : size);
+		ll_kunmap_atomic(ptr, KM_USER0);
+		buf += len;
+		size -= len;
+	}
+	return 0;
+}
+
 static void tgt_warn_on_cksum(struct ptlrpc_request *req,
 			      struct ptlrpc_bulk_desc *desc,
 			      struct niobuf_local *local_nb, int npages,
@@ -1936,18 +2242,22 @@ static void tgt_warn_on_cksum(struct ptlrpc_request *req,
 	body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 	LASSERT(body != NULL);
 
-	if (req->rq_peer.nid != desc->bd_sender) {
+	if (desc && req->rq_peer.nid != desc->bd_sender) {
 		via = " via ";
 		router = libcfs_nid2str(desc->bd_sender);
 	}
 
+	if (exp->exp_obd->obd_checksum_dump)
+		dump_all_bulk_pages(&body->oa, npages, local_nb, server_cksum,
+				    client_cksum);
+
 	if (mmap) {
 		CDEBUG_LIMIT(D_INFO, "client csum %x, server csum %x\n",
 			     client_cksum, server_cksum);
 		return;
 	}
 
-	LCONSOLE_ERROR_MSG(0x168, "BAD WRITE CHECKSUM: %s from %s%s%s inode "
+	LCONSOLE_ERROR_MSG(0x168, "%s: BAD WRITE CHECKSUM: from %s%s%s inode "
 			   DFID" object "DOSTID" extent [%llu-%llu"
 			   "]: client csum %x, server csum %x\n",
 			   exp->exp_obd->obd_name, libcfs_id2str(req->rq_peer),
@@ -1979,14 +2289,15 @@ int tgt_brw_write(struct tgt_session_info *tsi)
 	__u32			*rcs;
 	int			 objcount, niocount, npages;
 	int			 rc, i, j;
-	cksum_type_t		 cksum_type = OBD_CKSUM_CRC32;
+	enum cksum_types cksum_type = OBD_CKSUM_CRC32;
 	bool			 no_reply = false, mmap;
 	struct tgt_thread_big_cache *tbc = req->rq_svc_thread->t_data;
 	bool wait_sync = false;
 
 	ENTRY;
 
-	if (ptlrpc_req2svc(req)->srv_req_portal != OST_IO_PORTAL) {
+	if (ptlrpc_req2svc(req)->srv_req_portal != OST_IO_PORTAL &&
+	    ptlrpc_req2svc(req)->srv_req_portal != MDS_IO_PORTAL) {
 		CERROR("%s: deny write request from %s to portal %u\n",
 		       tgt_name(tsi->tsi_tgt),
 		       obd_export_nid2str(req->rq_export),
@@ -2036,7 +2347,7 @@ int tgt_brw_write(struct tgt_session_info *tsi)
 		RETURN(err_serious(-EPROTO));
 
 	if ((remote_nb[0].rnb_flags & OBD_BRW_MEMALLOC) &&
-	    (exp->exp_connection->c_peer.nid == exp->exp_connection->c_self))
+	    ptlrpc_connection_is_local(exp->exp_connection))
 		memory_pressure_set();
 
 	req_capsule_set_size(&req->rq_pill, &RMF_RCS, RCL_SERVER,
@@ -2050,8 +2361,8 @@ int tgt_brw_write(struct tgt_session_info *tsi)
 
 	local_nb = tbc->local;
 
-	rc = tgt_brw_lock(exp->exp_obd->obd_namespace, &tsi->tsi_resid, ioo,
-			  remote_nb, &lockh, LCK_PW);
+	rc = tgt_brw_lock(exp, &tsi->tsi_resid, ioo, remote_nb, &lockh,
+			  LCK_PW);
 	if (rc != 0)
 		GOTO(out, rc);
 
@@ -2059,14 +2370,13 @@ int tgt_brw_write(struct tgt_session_info *tsi)
 	 * If getting the lock took more time than
 	 * client was willing to wait, drop it. b=11330
 	 */
-	if (cfs_time_current_sec() > req->rq_deadline ||
+	if (ktime_get_real_seconds() > req->rq_deadline ||
 	    OBD_FAIL_CHECK(OBD_FAIL_OST_DROP_REQ)) {
 		no_reply = true;
-		CERROR("%s: Dropping timed-out write from %s because locking "
-		       "object "DOSTID" took %ld seconds (limit was %ld).\n",
+		CERROR("%s: Dropping timed-out write from %s because locking object " DOSTID " took %lld seconds (limit was %lld).\n",
 		       tgt_name(tsi->tsi_tgt), libcfs_id2str(req->rq_peer),
 		       POSTID(&ioo->ioo_oid),
-		       cfs_time_current_sec() - req->rq_arrival_time.tv_sec,
+		       ktime_get_real_seconds() - req->rq_arrival_time.tv_sec,
 		       req->rq_deadline - req->rq_arrival_time.tv_sec);
 		GOTO(out_lock, rc = -ETIMEDOUT);
 	}
@@ -2089,26 +2399,45 @@ 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;
+		unsigned char *short_io_buf;
+
+		short_io_size = req_capsule_get_size(&req->rq_pill,
+						     &RMF_SHORT_IO,
+						     RCL_CLIENT);
+		short_io_buf = req_capsule_client_get(&req->rq_pill,
+						      &RMF_SHORT_IO);
+		CDEBUG(D_INFO, "Client use short io for data transfer,"
+			       " size = %d\n", short_io_size);
+
+		/* Copy short io buf to pages */
+		rc = tgt_shortio2pages(local_nb, npages, short_io_buf,
+				       short_io_size);
+		desc = NULL;
+	} else {
+		desc = ptlrpc_prep_bulk_exp(req, npages, ioobj_max_brw_get(ioo),
+					    PTLRPC_BULK_GET_SINK |
+					    PTLRPC_BULK_BUF_KIOV,
+					    OST_BULK_PORTAL,
+					    &ptlrpc_bulk_kiov_nopin_ops);
+		if (desc == NULL)
+			GOTO(skip_transfer, rc = -ENOMEM);
+
+		/* NB Having prepped, we must commit... */
+		for (i = 0; i < npages; i++)
+			desc->bd_frag_ops->add_kiov_frag(desc,
+					local_nb[i].lnb_page,
+					local_nb[i].lnb_page_offset & ~PAGE_MASK,
+					local_nb[i].lnb_len);
+
+		rc = sptlrpc_svc_prep_bulk(req, desc);
+		if (rc != 0)
+			GOTO(skip_transfer, rc);
 
-	desc = ptlrpc_prep_bulk_exp(req, npages, ioobj_max_brw_get(ioo),
-				    PTLRPC_BULK_GET_SINK | PTLRPC_BULK_BUF_KIOV,
-				    OST_BULK_PORTAL,
-				    &ptlrpc_bulk_kiov_nopin_ops);
-	if (desc == NULL)
-		GOTO(skip_transfer, rc = -ENOMEM);
-
-	/* NB Having prepped, we must commit... */
-	for (i = 0; i < npages; i++)
-		desc->bd_frag_ops->add_kiov_frag(desc,
-						 local_nb[i].lnb_page,
-						 local_nb[i].lnb_page_offset,
-						 local_nb[i].lnb_len);
-
-	rc = sptlrpc_svc_prep_bulk(req, desc);
-	if (rc != 0)
-		GOTO(skip_transfer, rc);
+		rc = target_bulk_io(exp, desc, &lwi);
+	}
 
-	rc = target_bulk_io(exp, desc, &lwi);
 	no_reply = rc != 0;
 
 skip_transfer:
@@ -2121,8 +2450,12 @@ skip_transfer:
 		repbody->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
 		repbody->oa.o_flags &= ~OBD_FL_CKSUM_ALL;
 		repbody->oa.o_flags |= cksum_type_pack(cksum_type);
-		repbody->oa.o_cksum = tgt_checksum_bulk(tsi->tsi_tgt, desc,
-							OST_WRITE, cksum_type);
+		rc = tgt_checksum_niobuf(tsi->tsi_tgt, local_nb,
+					 npages, OST_WRITE, cksum_type,
+					 &repbody->oa.o_cksum);
+		if (rc < 0)
+			GOTO(out_commitrw, rc);
+
 		cksum_counter++;
 
 		if (unlikely(body->oa.o_cksum != repbody->oa.o_cksum)) {
@@ -2141,6 +2474,7 @@ skip_transfer:
 		}
 	}
 
+out_commitrw:
 	/* Must commit after prep above in all cases */
 	rc = obd_commitrw(tsi->tsi_env, OBD_BRW_WRITE, exp, &repbody->oa,
 			  objcount, ioo, remote_nb, npages, local_nb, rc);