lustre/target/tgt_handler.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 021110-1307, USA
  20  *
  21  * GPL HEADER END
  22  */
  23 /*
  24  * Copyright (c) 2013, 2017, Intel Corporation.
  25  */
  26 /*
  27  * lustre/target/tgt_handler.c
  28  *
  29  * Lustre Unified Target request handler code
  30  *
  31  * Author: Brian Behlendorf <behlendorf1@llnl.gov>
  32  * Author: Mikhail Pershin <mike.pershin@intel.com>
  33  */
  34
  35 #define DEBUG_SUBSYSTEM S_CLASS
  36
  37 #include <linux/user_namespace.h>
  38 #include <linux/delay.h>
  39 #include <linux/uidgid.h>
  40
  41 #include <libcfs/linux/linux-mem.h>
  42 #include <obd.h>
  43 #include <obd_class.h>
  44 #include <obd_cksum.h>
  45 #include <lustre_lfsck.h>
  46 #include <lustre_nodemap.h>
  47 #include <lustre_acl.h>
  48
  49 #include "tgt_internal.h"
  50
  51 char *tgt_name(struct lu_target *tgt)
  52 {
  53         LASSERT(tgt->lut_obd != NULL);
  54         return tgt->lut_obd->obd_name;
  55 }
  56 EXPORT_SYMBOL(tgt_name);
  57
  58 /*
  59  * Generic code handling requests that have struct mdt_body passed in:
  60  *
  61  *  - extract mdt_body from request and save it in @tsi, if present;
  62  *
  63  *  - create lu_object, corresponding to the fid in mdt_body, and save it in
  64  *  @tsi;
  65  *
  66  *  - if HAS_BODY flag is set for this request type check whether object
  67  *  actually exists on storage (lu_object_exists()).
  68  *
  69  */
  70 static int tgt_mdt_body_unpack(struct tgt_session_info *tsi, __u32 flags)
  71 {
  72         const struct mdt_body   *body;
  73         struct lu_object        *obj;
  74         struct req_capsule      *pill = tsi->tsi_pill;
  75         int                      rc;
  76
  77         ENTRY;
  78
  79         body = req_capsule_client_get(pill, &RMF_MDT_BODY);
  80         if (body == NULL)
  81                 RETURN(-EFAULT);
  82
  83         tsi->tsi_mdt_body = body;
  84
  85         if (!(body->mbo_valid & OBD_MD_FLID))
  86                 RETURN(0);
  87
  88         /* mdc_pack_body() doesn't check if fid is zero and set OBD_ML_FID
  89          * in any case in pre-2.5 clients. Fix that here if needed */
  90         if (unlikely(fid_is_zero(&body->mbo_fid1)))
  91                 RETURN(0);
  92
  93         if (!fid_is_sane(&body->mbo_fid1)) {
  94                 CERROR("%s: invalid FID: "DFID"\n", tgt_name(tsi->tsi_tgt),
  95                        PFID(&body->mbo_fid1));
  96                 RETURN(-EINVAL);
  97         }
  98
  99         obj = lu_object_find(tsi->tsi_env,
 100                              &tsi->tsi_tgt->lut_bottom->dd_lu_dev,
 101                              &body->mbo_fid1, NULL);
 102         if (!IS_ERR(obj)) {
 103                 if ((flags & HAS_BODY) && !lu_object_exists(obj)) {
 104                         lu_object_put(tsi->tsi_env, obj);
 105                         rc = -ENOENT;
 106                 } else {
 107                         tsi->tsi_corpus = obj;
 108                         rc = 0;
 109                 }
 110         } else {
 111                 rc = PTR_ERR(obj);
 112         }
 113
 114         tsi->tsi_fid = body->mbo_fid1;
 115
 116         RETURN(rc);
 117 }
 118
 119 /**
 120  * Validate oa from client.
 121  * If the request comes from 2.0 clients, currently only RSVD seq and IDIF
 122  * req are valid.
 123  *    a. objects in Single MDT FS  seq = FID_SEQ_OST_MDT0, oi_id != 0
 124  *    b. Echo objects(seq = 2), old echo client still use oi_id/oi_seq to
 125  *       pack ost_id. Because non-zero oi_seq will make it diffcult to tell
 126  *       whether this is oi_fid or real ostid. So it will check
 127  *       OBD_CONNECT_FID, then convert the ostid to FID for old client.
 128  *    c. Old FID-disable osc will send IDIF.
 129  *    d. new FID-enable osc/osp will send normal FID.
 130  *
 131  * And also oi_id/f_oid should always start from 1. oi_id/f_oid = 0 will
 132  * be used for LAST_ID file, and only being accessed inside OST now.
 133  */
 134 int tgt_validate_obdo(struct tgt_session_info *tsi, struct obdo *oa)
 135 {
 136         struct ost_id   *oi     = &oa->o_oi;
 137         u64              seq    = ostid_seq(oi);
 138         u64              id     = ostid_id(oi);
 139         int              rc;
 140         ENTRY;
 141
 142         if (unlikely(!(exp_connect_flags(tsi->tsi_exp) & OBD_CONNECT_FID) &&
 143                      fid_seq_is_echo(seq))) {
 144                 /* Sigh 2.[123] client still sends echo req with oi_id = 0
 145                  * during create, and we will reset this to 1, since this
 146                  * oi_id is basically useless in the following create process,
 147                  * but oi_id == 0 will make it difficult to tell whether it is
 148                  * real FID or ost_id. */
 149                 oi->oi_fid.f_seq = FID_SEQ_ECHO;
 150                 oi->oi_fid.f_oid = id ?: 1;
 151                 oi->oi_fid.f_ver = 0;
 152         } else {
 153                 struct tgt_thread_info *tti = tgt_th_info(tsi->tsi_env);
 154
 155                 if (unlikely((oa->o_valid & OBD_MD_FLID) && id == 0))
 156                         GOTO(out, rc = -EPROTO);
 157
 158                 /* Note: this check might be forced in 2.5 or 2.6, i.e.
 159                  * all of the requests are required to setup FLGROUP */
 160                 if (unlikely(!(oa->o_valid & OBD_MD_FLGROUP)))
 161                         oa->o_valid |= OBD_MD_FLGROUP;
 162
 163                 if (unlikely(!(fid_seq_is_idif(seq) || fid_seq_is_mdt0(seq) ||
 164                                fid_seq_is_norm(seq) || fid_seq_is_echo(seq))))
 165                         GOTO(out, rc = -EPROTO);
 166
 167                 rc = ostid_to_fid(&tti->tti_fid1, oi,
 168                                   tsi->tsi_tgt->lut_lsd.lsd_osd_index);
 169                 if (unlikely(rc != 0))
 170                         GOTO(out, rc);
 171
 172                 oi->oi_fid = tti->tti_fid1;
 173         }
 174
 175         RETURN(0);
 176
 177 out:
 178         CERROR("%s: client %s sent bad object "DOSTID": rc = %d\n",
 179                tgt_name(tsi->tsi_tgt), obd_export_nid2str(tsi->tsi_exp),
 180                seq, id, rc);
 181         return rc;
 182 }
 183 EXPORT_SYMBOL(tgt_validate_obdo);
 184
 185 static int tgt_io_data_unpack(struct tgt_session_info *tsi, struct ost_id *oi)
 186 {
 187         unsigned                 max_brw;
 188         struct niobuf_remote    *rnb;
 189         struct obd_ioobj        *ioo;
 190         int                      obj_count;
 191
 192         ENTRY;
 193
 194         ioo = req_capsule_client_get(tsi->tsi_pill, &RMF_OBD_IOOBJ);
 195         if (ioo == NULL)
 196                 RETURN(-EPROTO);
 197
 198         rnb = req_capsule_client_get(tsi->tsi_pill, &RMF_NIOBUF_REMOTE);
 199         if (rnb == NULL)
 200                 RETURN(-EPROTO);
 201
 202         max_brw = ioobj_max_brw_get(ioo);
 203         if (unlikely((max_brw & (max_brw - 1)) != 0)) {
 204                 CERROR("%s: client %s sent bad ioobj max %u for "DOSTID
 205                        ": rc = %d\n", tgt_name(tsi->tsi_tgt),
 206                        obd_export_nid2str(tsi->tsi_exp), max_brw,
 207                        POSTID(oi), -EPROTO);
 208                 RETURN(-EPROTO);
 209         }
 210         ioo->ioo_oid = *oi;
 211
 212         obj_count = req_capsule_get_size(tsi->tsi_pill, &RMF_OBD_IOOBJ,
 213                                         RCL_CLIENT) / sizeof(*ioo);
 214         if (obj_count == 0) {
 215                 CERROR("%s: short ioobj\n", tgt_name(tsi->tsi_tgt));
 216                 RETURN(-EPROTO);
 217         } else if (obj_count > 1) {
 218                 CERROR("%s: too many ioobjs (%d)\n", tgt_name(tsi->tsi_tgt),
 219                        obj_count);
 220                 RETURN(-EPROTO);
 221         }
 222
 223         if (ioo->ioo_bufcnt == 0) {
 224                 CERROR("%s: ioo has zero bufcnt\n", tgt_name(tsi->tsi_tgt));
 225                 RETURN(-EPROTO);
 226         }
 227
 228         if (ioo->ioo_bufcnt > PTLRPC_MAX_BRW_PAGES) {
 229                 DEBUG_REQ(D_RPCTRACE, tgt_ses_req(tsi),
 230                           "bulk has too many pages (%d)",
 231                           ioo->ioo_bufcnt);
 232                 RETURN(-EPROTO);
 233         }
 234
 235         RETURN(0);
 236 }
 237
 238 static int tgt_ost_body_unpack(struct tgt_session_info *tsi, __u32 flags)
 239 {
 240         struct ost_body         *body;
 241         struct req_capsule      *pill = tsi->tsi_pill;
 242         struct lu_nodemap       *nodemap;
 243         int                      rc;
 244
 245         ENTRY;
 246
 247         body = req_capsule_client_get(pill, &RMF_OST_BODY);
 248         if (body == NULL)
 249                 RETURN(-EFAULT);
 250
 251         rc = tgt_validate_obdo(tsi, &body->oa);
 252         if (rc)
 253                 RETURN(rc);
 254
 255         nodemap = nodemap_get_from_exp(tsi->tsi_exp);
 256         if (IS_ERR(nodemap))
 257                 RETURN(PTR_ERR(nodemap));
 258
 259         body->oa.o_uid = nodemap_map_id(nodemap, NODEMAP_UID,
 260                                         NODEMAP_CLIENT_TO_FS,
 261                                         body->oa.o_uid);
 262         body->oa.o_gid = nodemap_map_id(nodemap, NODEMAP_GID,
 263                                         NODEMAP_CLIENT_TO_FS,
 264                                         body->oa.o_gid);
 265         body->oa.o_projid = nodemap_map_id(nodemap, NODEMAP_PROJID,
 266                                            NODEMAP_CLIENT_TO_FS,
 267                                            body->oa.o_projid);
 268         nodemap_putref(nodemap);
 269
 270         tsi->tsi_ost_body = body;
 271         tsi->tsi_fid = body->oa.o_oi.oi_fid;
 272
 273         if (req_capsule_has_field(pill, &RMF_OBD_IOOBJ, RCL_CLIENT)) {
 274                 rc = tgt_io_data_unpack(tsi, &body->oa.o_oi);
 275                 if (rc < 0)
 276                         RETURN(rc);
 277         }
 278
 279         if (!(body->oa.o_valid & OBD_MD_FLID)) {
 280                 if (flags & HAS_BODY) {
 281                         CERROR("%s: OBD_MD_FLID flag is not set in ost_body but OID/FID is mandatory with HAS_BODY\n",
 282                                tgt_name(tsi->tsi_tgt));
 283                         RETURN(-EPROTO);
 284                 } else {
 285                         RETURN(0);
 286                 }
 287         }
 288
 289         ost_fid_build_resid(&tsi->tsi_fid, &tsi->tsi_resid);
 290
 291         /*
 292          * OST doesn't get object in advance for further use to prevent
 293          * situations with nested object_find which is potential deadlock.
 294          */
 295         tsi->tsi_corpus = NULL;
 296         RETURN(rc);
 297 }
 298
 299 /*
 300  * Do necessary preprocessing according to handler ->th_flags.
 301  */
 302 static int tgt_request_preprocess(struct tgt_session_info *tsi,
 303                                   struct tgt_handler *h,
 304                                   struct ptlrpc_request *req)
 305 {
 306         struct req_capsule      *pill = tsi->tsi_pill;
 307         __u32                    flags = h->th_flags;
 308         int                      rc = 0;
 309
 310         ENTRY;
 311
 312         if (tsi->tsi_preprocessed)
 313                 RETURN(0);
 314
 315         LASSERT(h->th_act != NULL);
 316         LASSERT(h->th_opc == lustre_msg_get_opc(req->rq_reqmsg));
 317         LASSERT(current->journal_info == NULL);
 318
 319         LASSERT(ergo(flags & (HAS_BODY | HAS_REPLY),
 320                      h->th_fmt != NULL));
 321         if (h->th_fmt != NULL) {
 322                 req_capsule_set(pill, h->th_fmt);
 323                 if (req_capsule_has_field(pill, &RMF_MDT_BODY, RCL_CLIENT) &&
 324                     req_capsule_field_present(pill, &RMF_MDT_BODY,
 325                                               RCL_CLIENT)) {
 326                         rc = tgt_mdt_body_unpack(tsi, flags);
 327                         if (rc < 0)
 328                                 RETURN(rc);
 329                 } else if (req_capsule_has_field(pill, &RMF_OST_BODY,
 330                                                  RCL_CLIENT) &&
 331                            req_capsule_field_present(pill, &RMF_OST_BODY,
 332                                                  RCL_CLIENT)) {
 333                         rc = tgt_ost_body_unpack(tsi, flags);
 334                         if (rc < 0)
 335                                 RETURN(rc);
 336                 }
 337         }
 338
 339         if (flags & IS_MUTABLE && tgt_conn_flags(tsi) & OBD_CONNECT_RDONLY)
 340                 RETURN(-EROFS);
 341
 342         if (flags & HAS_KEY) {
 343                 struct ldlm_request *dlm_req;
 344
 345                 LASSERT(h->th_fmt != NULL);
 346
 347                 dlm_req = req_capsule_client_get(pill, &RMF_DLM_REQ);
 348                 if (dlm_req != NULL) {
 349                         union ldlm_wire_policy_data *policy =
 350                                         &dlm_req->lock_desc.l_policy_data;
 351
 352                         if (unlikely(dlm_req->lock_desc.l_resource.lr_type ==
 353                                      LDLM_IBITS &&
 354                                      (policy->l_inodebits.bits |
 355                                       policy->l_inodebits.try_bits) == 0)) {
 356                                 /*
 357                                  * Lock without inodebits makes no sense and
 358                                  * will oops later in ldlm. If client miss to
 359                                  * set such bits, do not trigger ASSERTION.
 360                                  *
 361                                  * For liblustre flock case, it maybe zero.
 362                                  */
 363                                 rc = -EPROTO;
 364                         } else {
 365                                 tsi->tsi_dlm_req = dlm_req;
 366                         }
 367                 } else {
 368                         rc = -EFAULT;
 369                 }
 370         }
 371         tsi->tsi_preprocessed = 1;
 372         RETURN(rc);
 373 }
 374
 375 /*
 376  * Invoke handler for this request opc. Also do necessary preprocessing
 377  * (according to handler ->th_flags), and post-processing (setting of
 378  * ->last_{xid,committed}).
 379  */
 380 static int tgt_handle_request0(struct tgt_session_info *tsi,
 381                                struct tgt_handler *h,
 382                                struct ptlrpc_request *req)
 383 {
 384         int      serious = 0;
 385         int      rc;
 386         __u32    opc = lustre_msg_get_opc(req->rq_reqmsg);
 387
 388         ENTRY;
 389
 390
 391         /* When dealing with sec context requests, no export is associated yet,
 392          * because these requests are sent before *_CONNECT requests.
 393          * A NULL req->rq_export means the normal *_common_slice handlers will
 394          * not be called, because there is no reference to the target.
 395          * So deal with them by hand and jump directly to target_send_reply().
 396          */
 397         switch (opc) {
 398         case SEC_CTX_INIT:
 399         case SEC_CTX_INIT_CONT:
 400         case SEC_CTX_FINI:
 401                 CFS_FAIL_TIMEOUT(OBD_FAIL_SEC_CTX_HDL_PAUSE, cfs_fail_val);
 402                 GOTO(out, rc = 0);
 403         }
 404
 405         /*
 406          * Checking for various OBD_FAIL_$PREF_$OPC_NET codes. _Do_ not try
 407          * to put same checks into handlers like mdt_close(), mdt_reint(),
 408          * etc., without talking to mdt authors first. Checking same thing
 409          * there again is useless and returning 0 error without packing reply
 410          * is buggy! Handlers either pack reply or return error.
 411          *
 412          * We return 0 here and do not send any reply in order to emulate
 413          * network failure. Do not send any reply in case any of NET related
 414          * fail_id has occured.
 415          */
 416         if (OBD_FAIL_CHECK_ORSET(h->th_fail_id, OBD_FAIL_ONCE))
 417                 RETURN(0);
 418         if (unlikely(lustre_msg_get_opc(req->rq_reqmsg) == MDS_REINT &&
 419                      OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_MULTI_NET)))
 420                 RETURN(0);
 421
 422         /* drop OUT_UPDATE rpc */
 423         if (unlikely(lustre_msg_get_opc(req->rq_reqmsg) == OUT_UPDATE &&
 424                      OBD_FAIL_CHECK(OBD_FAIL_OUT_UPDATE_DROP)))
 425                 RETURN(0);
 426
 427         rc = tgt_request_preprocess(tsi, h, req);
 428         /* pack reply if reply format is fixed */
 429         if (rc == 0 && h->th_flags & HAS_REPLY) {
 430                 /* Pack reply */
 431                 if (req_capsule_has_field(tsi->tsi_pill, &RMF_MDT_MD,
 432                                           RCL_SERVER))
 433                         req_capsule_set_size(tsi->tsi_pill, &RMF_MDT_MD,
 434                                              RCL_SERVER,
 435                                              tsi->tsi_mdt_body->mbo_eadatasize);
 436                 if (req_capsule_has_field(tsi->tsi_pill, &RMF_LOGCOOKIES,
 437                                           RCL_SERVER))
 438                         req_capsule_set_size(tsi->tsi_pill, &RMF_LOGCOOKIES,
 439                                              RCL_SERVER, 0);
 440                 if (req_capsule_has_field(tsi->tsi_pill, &RMF_ACL, RCL_SERVER))
 441                         req_capsule_set_size(tsi->tsi_pill,
 442                                              &RMF_ACL, RCL_SERVER,
 443                                              LUSTRE_POSIX_ACL_MAX_SIZE_OLD);
 444
 445                 if (req_capsule_has_field(tsi->tsi_pill, &RMF_SHORT_IO,
 446                                           RCL_SERVER)) {
 447                         struct niobuf_remote *remote_nb =
 448                                 req_capsule_client_get(tsi->tsi_pill,
 449                                                        &RMF_NIOBUF_REMOTE);
 450                         struct ost_body *body = tsi->tsi_ost_body;
 451
 452                         req_capsule_set_size(tsi->tsi_pill, &RMF_SHORT_IO,
 453                                          RCL_SERVER,
 454                                          (body->oa.o_valid & OBD_MD_FLFLAGS &&
 455                                           body->oa.o_flags & OBD_FL_SHORT_IO) ?
 456                                          remote_nb[0].rnb_len : 0);
 457                 }
 458                 if (req_capsule_has_field(tsi->tsi_pill, &RMF_FILE_ENCCTX,
 459                                           RCL_SERVER))
 460                         req_capsule_set_size(tsi->tsi_pill, &RMF_FILE_ENCCTX,
 461                                              RCL_SERVER, 0);
 462
 463                 rc = req_capsule_server_pack(tsi->tsi_pill);
 464         }
 465
 466         if (likely(rc == 0)) {
 467                 /*
 468                  * Process request, there can be two types of rc:
 469                  * 1) errors with msg unpack/pack, other failures outside the
 470                  * operation itself. This is counted as serious errors;
 471                  * 2) errors during fs operation, should be placed in rq_status
 472                  * only
 473                  */
 474                 rc = h->th_act(tsi);
 475                 if (!is_serious(rc) &&
 476                     !req->rq_no_reply && req->rq_reply_state == NULL) {
 477                         DEBUG_REQ(D_ERROR, req,
 478                                   "%s: %s handler did not pack reply but returned no error",
 479                                   tgt_name(tsi->tsi_tgt), h->th_name);
 480                         LBUG();
 481                 }
 482                 serious = is_serious(rc);
 483                 rc = clear_serious(rc);
 484         } else {
 485                 serious = 1;
 486         }
 487
 488         req->rq_status = rc;
 489
 490         /*
 491          * ELDLM_* codes which > 0 should be in rq_status only as well as
 492          * all non-serious errors.
 493          */
 494         if (rc > 0 || !serious)
 495                 rc = 0;
 496
 497         LASSERT(current->journal_info == NULL);
 498
 499         if (likely(rc == 0 && req->rq_export))
 500                 target_committed_to_req(req);
 501
 502 out:
 503         target_send_reply(req, rc, tsi->tsi_reply_fail_id);
 504         RETURN(0);
 505 }
 506
 507 static int tgt_filter_recovery_request(struct ptlrpc_request *req,
 508                                        struct obd_device *obd, int *process)
 509 {
 510         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
 511         case MDS_DISCONNECT:
 512         case OST_DISCONNECT:
 513         case OBD_IDX_READ:
 514                 *process = 1;
 515                 RETURN(0);
 516         case MDS_CLOSE:
 517         case MDS_SYNC: /* used in unmounting */
 518         case OBD_PING:
 519         case MDS_REINT:
 520         case OUT_UPDATE:
 521         case SEQ_QUERY:
 522         case FLD_QUERY:
 523         case FLD_READ:
 524         case LDLM_ENQUEUE:
 525         case OST_CREATE:
 526         case OST_DESTROY:
 527         case OST_PUNCH:
 528         case OST_SETATTR:
 529         case OST_SYNC:
 530         case OST_WRITE:
 531         case MDS_HSM_PROGRESS:
 532         case MDS_HSM_STATE_SET:
 533         case MDS_HSM_REQUEST:
 534         case OST_FALLOCATE:
 535                 *process = target_queue_recovery_request(req, obd);
 536                 RETURN(0);
 537
 538         default:
 539                 DEBUG_REQ(D_ERROR, req, "not permitted during recovery");
 540                 *process = -EAGAIN;
 541                 RETURN(0);
 542         }
 543 }
 544
 545 /*
 546  * Handle recovery. Return:
 547  *        +1: continue request processing;
 548  *       -ve: abort immediately with the given error code;
 549  *         0: send reply with error code in req->rq_status;
 550  */
 551 static int tgt_handle_recovery(struct ptlrpc_request *req, int reply_fail_id)
 552 {
 553         ENTRY;
 554
 555         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
 556         case MDS_CONNECT:
 557         case OST_CONNECT:
 558         case MGS_CONNECT:
 559         case SEC_CTX_INIT:
 560         case SEC_CTX_INIT_CONT:
 561         case SEC_CTX_FINI:
 562                 RETURN(+1);
 563         }
 564
 565         if (!req->rq_export->exp_obd->obd_replayable)
 566                 RETURN(+1);
 567
 568         /* sanity check: if the xid matches, the request must be marked as a
 569          * resent or replayed */
 570         if (req_can_reconstruct(req, NULL) == 1) {
 571                 if (!(lustre_msg_get_flags(req->rq_reqmsg) &
 572                       (MSG_RESENT | MSG_REPLAY))) {
 573                         DEBUG_REQ(D_WARNING, req,
 574                                   "rq_xid=%llu matches saved XID, expected REPLAY or RESENT flag (%x)",
 575                                   req->rq_xid,
 576                                   lustre_msg_get_flags(req->rq_reqmsg));
 577                         req->rq_status = -ENOTCONN;
 578                         RETURN(-ENOTCONN);
 579                 }
 580         }
 581         /* else: note the opposite is not always true; a RESENT req after a
 582          * failover will usually not match the last_xid, since it was likely
 583          * never committed. A REPLAYed request will almost never match the
 584          * last xid, however it could for a committed, but still retained,
 585          * open. */
 586
 587         /* Check for aborted recovery... */
 588         if (unlikely(req->rq_export->exp_obd->obd_recovering)) {
 589                 int rc;
 590                 int should_process;
 591
 592                 DEBUG_REQ(D_INFO, req, "Got new replay");
 593                 rc = tgt_filter_recovery_request(req, req->rq_export->exp_obd,
 594                                                  &should_process);
 595                 if (rc != 0 || !should_process)
 596                         RETURN(rc);
 597                 else if (should_process < 0) {
 598                         req->rq_status = should_process;
 599                         rc = ptlrpc_error(req);
 600                         RETURN(rc);
 601                 }
 602         }
 603         RETURN(+1);
 604 }
 605
 606 /* Initial check for request, it is validation mostly */
 607 static struct tgt_handler *tgt_handler_find_check(struct ptlrpc_request *req)
 608 {
 609         struct tgt_handler      *h;
 610         struct tgt_opc_slice    *s;
 611         struct lu_target        *tgt;
 612         __u32                    opc = lustre_msg_get_opc(req->rq_reqmsg);
 613         /* don't spew error messages for unhandled RPCs */
 614         static bool              printed;
 615
 616         ENTRY;
 617
 618         tgt = class_exp2tgt(req->rq_export);
 619         if (unlikely(tgt == NULL)) {
 620                 DEBUG_REQ(D_ERROR, req, "%s: no target for connected export",
 621                           class_exp2obd(req->rq_export)->obd_name);
 622                 RETURN(ERR_PTR(-EINVAL));
 623         }
 624
 625         for (s = tgt->lut_slice; s->tos_hs != NULL; s++)
 626                 if (s->tos_opc_start <= opc && opc < s->tos_opc_end)
 627                         break;
 628
 629         /* opcode was not found in slice */
 630         if (unlikely(s->tos_hs == NULL)) {
 631                 if (!printed) {
 632                         CERROR("%s: no handler for opcode 0x%x from %s\n",
 633                                tgt_name(tgt), opc, libcfs_idstr(&req->rq_peer));
 634                         printed = true;
 635                 }
 636                 goto err_unsupported;
 637         }
 638
 639         LASSERT(opc >= s->tos_opc_start && opc < s->tos_opc_end);
 640         h = s->tos_hs + (opc - s->tos_opc_start);
 641         if (unlikely(h->th_opc == 0)) {
 642                 if (!printed) {
 643                         CERROR("%s: unsupported opcode 0x%x\n",
 644                                tgt_name(tgt), opc);
 645                         printed = true;
 646                 }
 647                 goto err_unsupported;
 648         }
 649
 650         if (OBD_FAIL_CHECK(OBD_FAIL_OST_OPCODE) && opc == cfs_fail_val)
 651                 goto err_unsupported;
 652
 653         RETURN(h);
 654 err_unsupported:
 655         /*
 656          * Unknown opcode does not necessarily means insane client. A new
 657          * client might send RPCs with new opcodes to an old server. The
 658          * client might desperately stuck there waiting for a reply. So,
 659          * send an error back here.
 660          *
 661          * An old client might also send RPCs with deprecated opcodes (e.g.
 662          * OST_QUOTACHECK).
 663          *
 664          * Error in ptlrpc_send_error() is ignored.
 665          */
 666         req->rq_status = -EOPNOTSUPP;
 667         ptlrpc_send_error(req, PTLRPC_REPLY_MAYBE_DIFFICULT);
 668         RETURN(ERR_PTR(-EOPNOTSUPP));
 669 }
 670
 671 static int process_req_last_xid(struct ptlrpc_request *req)
 672 {
 673         __u64   last_xid;
 674         int rc = 0;
 675         struct obd_export *exp = req->rq_export;
 676         struct tg_export_data *ted = &exp->exp_target_data;
 677         bool need_lock = tgt_is_multimodrpcs_client(exp);
 678         ENTRY;
 679
 680         if (need_lock)
 681                 mutex_lock(&ted->ted_lcd_lock);
 682         /* check request's xid is consistent with export's last_xid */
 683         last_xid = lustre_msg_get_last_xid(req->rq_reqmsg);
 684         if (last_xid > exp->exp_last_xid)
 685                 exp->exp_last_xid = last_xid;
 686
 687         if (req->rq_xid == 0 || req->rq_xid <= exp->exp_last_xid) {
 688                 /* Some request is allowed to be sent during replay,
 689                  * such as OUT update requests, FLD requests, so it
 690                  * is possible that replay requests has smaller XID
 691                  * than the exp_last_xid.
 692                  *
 693                  * Some non-replay requests may have smaller XID as
 694                  * well:
 695                  *
 696                  * - Client send a no_resend RPC, like statfs;
 697                  * - The RPC timedout (or some other error) on client,
 698                  *   then it's removed from the unreplied list;
 699                  * - Client send some other request to bump the
 700                  *   exp_last_xid on server;
 701                  * - The former RPC got chance to be processed;
 702                  */
 703                 if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY))
 704                         rc = -EPROTO;
 705
 706                 DEBUG_REQ(D_WARNING, req,
 707                           "unexpected xid=%llx != exp_last_xid=%llx, rc = %d",
 708                           req->rq_xid, exp->exp_last_xid, rc);
 709                 if (rc)
 710                         GOTO(out, rc);
 711         }
 712
 713         /* The "last_xid" is the minimum xid among unreplied requests,
 714          * if the request is from the previous connection, its xid can
 715          * still be larger than "exp_last_xid", then the above check of
 716          * xid is not enough to determine whether the request is delayed.
 717          *
 718          * For example, if some replay request was delayed and caused
 719          * timeout at client and the replay is restarted, the delayed
 720          * replay request will have the larger xid than "exp_last_xid"
 721          */
 722         if (req->rq_export->exp_conn_cnt >
 723             lustre_msg_get_conn_cnt(req->rq_reqmsg)) {
 724                 CDEBUG(D_RPCTRACE,
 725                        "Dropping request %llu from an old epoch %u/%u\n",
 726                        req->rq_xid,
 727                        lustre_msg_get_conn_cnt(req->rq_reqmsg),
 728                        req->rq_export->exp_conn_cnt);
 729                 req->rq_no_reply = 1;
 730                 GOTO(out, rc = -ESTALE);
 731         }
 732
 733         /* try to release in-memory reply data */
 734         if (tgt_is_multimodrpcs_client(exp)) {
 735                 tgt_handle_received_xid(exp, last_xid);
 736                 rc = tgt_handle_tag(req);
 737         }
 738
 739 out:
 740         if (need_lock)
 741                 mutex_unlock(&ted->ted_lcd_lock);
 742
 743         RETURN(rc);
 744 }
 745
 746 int tgt_request_handle(struct ptlrpc_request *req)
 747 {
 748         struct tgt_session_info *tsi = tgt_ses_info(req->rq_svc_thread->t_env);
 749
 750         struct lustre_msg       *msg = req->rq_reqmsg;
 751         struct tgt_handler      *h;
 752         struct lu_target        *tgt;
 753         int                      request_fail_id = 0;
 754         __u32                    opc = lustre_msg_get_opc(msg);
 755         struct obd_device       *obd;
 756         int                      rc;
 757         bool                     is_connect = false;
 758         ENTRY;
 759
 760         if (unlikely(OBD_FAIL_CHECK(OBD_FAIL_TGT_RECOVERY_REQ_RACE))) {
 761                 if (cfs_fail_val == 0 &&
 762                     lustre_msg_get_opc(msg) != OBD_PING &&
 763                     lustre_msg_get_flags(msg) & MSG_REQ_REPLAY_DONE) {
 764                         cfs_fail_val = 1;
 765                         cfs_race_state = 0;
 766                         wait_event_idle(cfs_race_waitq, (cfs_race_state == 1));
 767                 }
 768         }
 769
 770         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
 771         tsi->tsi_pill = &req->rq_pill;
 772         tsi->tsi_env = req->rq_svc_thread->t_env;
 773
 774         /* if request has export then get handlers slice from corresponding
 775          * target, otherwise that should be connect operation */
 776         if (opc == MDS_CONNECT || opc == OST_CONNECT ||
 777             opc == MGS_CONNECT) {
 778                 is_connect = true;
 779                 req_capsule_set(&req->rq_pill, &RQF_CONNECT);
 780                 rc = target_handle_connect(req);
 781                 if (rc != 0) {
 782                         rc = ptlrpc_error(req);
 783                         GOTO(out, rc);
 784                 }
 785                 /* recovery-small test 18c asks to drop connect reply */
 786                 if (unlikely(opc == OST_CONNECT &&
 787                              OBD_FAIL_CHECK(OBD_FAIL_OST_CONNECT_NET2)))
 788                         GOTO(out, rc = 0);
 789         }
 790
 791         if (unlikely(!class_connected_export(req->rq_export))) {
 792                 if (opc == SEC_CTX_INIT || opc == SEC_CTX_INIT_CONT ||
 793                     opc == SEC_CTX_FINI) {
 794                         /* sec context initialization has to be handled
 795                          * by hand in tgt_handle_request0() */
 796                         tsi->tsi_reply_fail_id = OBD_FAIL_SEC_CTX_INIT_NET;
 797                         h = NULL;
 798                         GOTO(handle_recov, rc = 0);
 799                 }
 800                 CDEBUG(D_HA, "operation %d on unconnected OST from %s\n",
 801                        opc, libcfs_idstr(&req->rq_peer));
 802                 req->rq_status = -ENOTCONN;
 803                 rc = ptlrpc_error(req);
 804                 GOTO(out, rc);
 805         }
 806
 807         tsi->tsi_tgt = tgt = class_exp2tgt(req->rq_export);
 808         tsi->tsi_exp = req->rq_export;
 809         if (exp_connect_flags(req->rq_export) & OBD_CONNECT_JOBSTATS)
 810                 tsi->tsi_jobid = lustre_msg_get_jobid(req->rq_reqmsg);
 811         else
 812                 tsi->tsi_jobid = NULL;
 813
 814         if (tgt == NULL) {
 815                 DEBUG_REQ(D_ERROR, req, "%s: No target for connected export",
 816                           class_exp2obd(req->rq_export)->obd_name);
 817                 req->rq_status = -EINVAL;
 818                 rc = ptlrpc_error(req);
 819                 GOTO(out, rc);
 820         }
 821
 822         /* Skip last_xid processing for the recovery thread, otherwise, the
 823          * last_xid on same request could be processed twice: first time when
 824          * processing the incoming request, second time when the request is
 825          * being processed by recovery thread. */
 826         obd = class_exp2obd(req->rq_export);
 827         if (is_connect) {
 828                 /* reset the exp_last_xid on each connection. */
 829                 req->rq_export->exp_last_xid = 0;
 830         } else if (obd->obd_recovery_data.trd_processing_task !=
 831                    current->pid) {
 832                 rc = process_req_last_xid(req);
 833                 if (rc) {
 834                         req->rq_status = rc;
 835                         rc = ptlrpc_error(req);
 836                         GOTO(out, rc);
 837                 }
 838         }
 839
 840         request_fail_id = tgt->lut_request_fail_id;
 841         tsi->tsi_reply_fail_id = tgt->lut_reply_fail_id;
 842
 843         h = tgt_handler_find_check(req);
 844         if (IS_ERR(h)) {
 845                 req->rq_status = PTR_ERR(h);
 846                 rc = ptlrpc_error(req);
 847                 GOTO(out, rc);
 848         }
 849
 850         LASSERTF(h->th_opc == opc, "opcode mismatch %d != %d\n",
 851                  h->th_opc, opc);
 852
 853         if ((cfs_fail_val == 0 || cfs_fail_val == opc) &&
 854              CFS_FAIL_CHECK_ORSET(request_fail_id, CFS_FAIL_ONCE))
 855                 GOTO(out, rc = 0);
 856
 857         rc = lustre_msg_check_version(msg, h->th_version);
 858         if (unlikely(rc)) {
 859                 DEBUG_REQ(D_ERROR, req,
 860                           "%s: drop malformed request version=%08x expect=%08x",
 861                           tgt_name(tgt), lustre_msg_get_version(msg),
 862                           h->th_version);
 863                 req->rq_status = -EINVAL;
 864                 rc = ptlrpc_error(req);
 865                 GOTO(out, rc);
 866         }
 867
 868 handle_recov:
 869         rc = tgt_handle_recovery(req, tsi->tsi_reply_fail_id);
 870         if (likely(rc == 1)) {
 871                 rc = tgt_handle_request0(tsi, h, req);
 872                 if (rc)
 873                         GOTO(out, rc);
 874         }
 875         EXIT;
 876 out:
 877         req_capsule_fini(tsi->tsi_pill);
 878         if (tsi->tsi_corpus != NULL) {
 879                 lu_object_put(tsi->tsi_env, tsi->tsi_corpus);
 880                 tsi->tsi_corpus = NULL;
 881         }
 882         return rc;
 883 }
 884 EXPORT_SYMBOL(tgt_request_handle);
 885
 886 /** Assign high priority operations to the request if needed. */
 887 int tgt_hpreq_handler(struct ptlrpc_request *req)
 888 {
 889         struct tgt_session_info *tsi = tgt_ses_info(req->rq_svc_thread->t_env);
 890         struct tgt_handler      *h;
 891         int                      rc;
 892
 893         ENTRY;
 894
 895         if (req->rq_export == NULL)
 896                 RETURN(0);
 897
 898         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
 899         tsi->tsi_pill = &req->rq_pill;
 900         tsi->tsi_env = req->rq_svc_thread->t_env;
 901         tsi->tsi_tgt = class_exp2tgt(req->rq_export);
 902         tsi->tsi_exp = req->rq_export;
 903
 904         h = tgt_handler_find_check(req);
 905         if (IS_ERR(h)) {
 906                 rc = PTR_ERR(h);
 907                 RETURN(rc);
 908         }
 909
 910         rc = tgt_request_preprocess(tsi, h, req);
 911         if (unlikely(rc != 0))
 912                 RETURN(rc);
 913
 914         if (h->th_hp != NULL)
 915                 h->th_hp(tsi);
 916         RETURN(0);
 917 }
 918 EXPORT_SYMBOL(tgt_hpreq_handler);
 919
 920 void tgt_counter_incr(struct obd_export *exp, int opcode)
 921 {
 922         lprocfs_counter_incr(exp->exp_obd->obd_stats, opcode);
 923         if (exp->exp_nid_stats && exp->exp_nid_stats->nid_stats != NULL)
 924                 lprocfs_counter_incr(exp->exp_nid_stats->nid_stats, opcode);
 925 }
 926 EXPORT_SYMBOL(tgt_counter_incr);
 927
 928 /*
 929  * Unified target generic handlers.
 930  */
 931
 932 int tgt_connect_check_sptlrpc(struct ptlrpc_request *req, struct obd_export *exp)
 933 {
 934         struct lu_target *tgt = class_exp2tgt(exp);
 935         struct sptlrpc_flavor flvr;
 936         int rc = 0;
 937
 938         LASSERT(tgt);
 939         LASSERT(tgt->lut_obd);
 940         LASSERT(tgt->lut_slice);
 941
 942         /* always allow ECHO client */
 943         if (unlikely(strcmp(exp->exp_obd->obd_type->typ_name,
 944                             LUSTRE_ECHO_NAME) == 0)) {
 945                 exp->exp_flvr.sf_rpc = SPTLRPC_FLVR_ANY;
 946                 return 0;
 947         }
 948
 949         if (exp->exp_flvr.sf_rpc == SPTLRPC_FLVR_INVALID) {
 950                 read_lock(&tgt->lut_sptlrpc_lock);
 951                 sptlrpc_target_choose_flavor(&tgt->lut_sptlrpc_rset,
 952                                              req->rq_sp_from,
 953                                              lnet_nid_to_nid4(&req->rq_peer.nid),
 954                                              &flvr);
 955                 read_unlock(&tgt->lut_sptlrpc_lock);
 956
 957                 spin_lock(&exp->exp_lock);
 958                 exp->exp_sp_peer = req->rq_sp_from;
 959                 exp->exp_flvr = flvr;
 960
 961                 /* when on mgs, if no restriction is set, or if the client
 962                  * NID is on the local node, allow any flavor
 963                  */
 964                 if ((strcmp(exp->exp_obd->obd_type->typ_name,
 965                             LUSTRE_MGS_NAME) == 0) &&
 966                     (exp->exp_flvr.sf_rpc == SPTLRPC_FLVR_NULL ||
 967                      LNetIsPeerLocal(&exp->exp_connection->c_peer.nid)))
 968                         exp->exp_flvr.sf_rpc = SPTLRPC_FLVR_ANY;
 969
 970                 if (exp->exp_flvr.sf_rpc != SPTLRPC_FLVR_ANY &&
 971                     exp->exp_flvr.sf_rpc != req->rq_flvr.sf_rpc) {
 972                         CERROR("%s: unauthorized rpc flavor %x from %s, "
 973                                "expect %x\n", tgt_name(tgt),
 974                                req->rq_flvr.sf_rpc,
 975                                libcfs_nidstr(&req->rq_peer.nid),
 976                                exp->exp_flvr.sf_rpc);
 977                         rc = -EACCES;
 978                 }
 979                 spin_unlock(&exp->exp_lock);
 980         } else {
 981                 if (exp->exp_sp_peer != req->rq_sp_from) {
 982                         CERROR("%s: RPC source %s doesn't match %s\n",
 983                                tgt_name(tgt),
 984                                sptlrpc_part2name(req->rq_sp_from),
 985                                sptlrpc_part2name(exp->exp_sp_peer));
 986                         rc = -EACCES;
 987                 } else {
 988                         rc = sptlrpc_target_export_check(exp, req);
 989                 }
 990         }
 991
 992         return rc;
 993 }
 994
 995 int tgt_adapt_sptlrpc_conf(struct lu_target *tgt)
 996 {
 997         struct sptlrpc_rule_set  tmp_rset;
 998         int                      rc;
 999
1000         if (unlikely(tgt == NULL)) {
1001                 CERROR("No target passed\n");
1002                 return -EINVAL;
1003         }
1004
1005         sptlrpc_rule_set_init(&tmp_rset);
1006         rc = sptlrpc_conf_target_get_rules(tgt->lut_obd, &tmp_rset);
1007         if (rc) {
1008                 CERROR("%s: failed get sptlrpc rules: rc = %d\n",
1009                        tgt_name(tgt), rc);
1010                 return rc;
1011         }
1012
1013         sptlrpc_target_update_exp_flavor(tgt->lut_obd, &tmp_rset);
1014
1015         write_lock(&tgt->lut_sptlrpc_lock);
1016         sptlrpc_rule_set_free(&tgt->lut_sptlrpc_rset);
1017         tgt->lut_sptlrpc_rset = tmp_rset;
1018         write_unlock(&tgt->lut_sptlrpc_lock);
1019
1020         return 0;
1021 }
1022 EXPORT_SYMBOL(tgt_adapt_sptlrpc_conf);
1023
1024 int tgt_connect(struct tgt_session_info *tsi)
1025 {
1026         struct ptlrpc_request   *req = tgt_ses_req(tsi);
1027         struct obd_connect_data *reply;
1028         int                      rc;
1029
1030         ENTRY;
1031
1032         /* XXX: better to call this check right after getting new export but
1033          * before last_rcvd slot allocation to avoid server load upon insecure
1034          * connects. This is to be fixed after unifiyng all targets.
1035          */
1036         rc = tgt_connect_check_sptlrpc(req, tsi->tsi_exp);
1037         if (rc)
1038                 GOTO(out, rc);
1039
1040         /* To avoid exposing partially initialized connection flags, changes up
1041          * to this point have been staged in reply->ocd_connect_flags. Now that
1042          * connection handling has completed successfully, atomically update
1043          * the connect flags in the shared export data structure. LU-1623 */
1044         reply = req_capsule_server_get(tsi->tsi_pill, &RMF_CONNECT_DATA);
1045         spin_lock(&tsi->tsi_exp->exp_lock);
1046         *exp_connect_flags_ptr(tsi->tsi_exp) = reply->ocd_connect_flags;
1047         if (reply->ocd_connect_flags & OBD_CONNECT_FLAGS2)
1048                 *exp_connect_flags2_ptr(tsi->tsi_exp) =
1049                         reply->ocd_connect_flags2;
1050         tsi->tsi_exp->exp_connect_data.ocd_brw_size = reply->ocd_brw_size;
1051         spin_unlock(&tsi->tsi_exp->exp_lock);
1052
1053         if (strcmp(tsi->tsi_exp->exp_obd->obd_type->typ_name,
1054                    LUSTRE_MDT_NAME) == 0) {
1055                 struct lu_nodemap *nm = NULL;
1056
1057                 rc = req_check_sepol(tsi->tsi_pill);
1058                 if (rc)
1059                         GOTO(out, rc);
1060
1061                 if (tsi->tsi_pill->rc_req->rq_export)
1062                         nm =
1063                          nodemap_get_from_exp(tsi->tsi_pill->rc_req->rq_export);
1064
1065                 if (reply->ocd_connect_flags & OBD_CONNECT_FLAGS2 &&
1066                     reply->ocd_connect_flags2 & OBD_CONNECT2_ENCRYPT) {
1067                         bool forbid_encrypt = true;
1068
1069                         if (!nm)
1070                                 /* nodemap_get_from_exp returns NULL in case
1071                                  * nodemap is not active, so we do not forbid
1072                                  */
1073                                 forbid_encrypt = false;
1074                         else if (!IS_ERR(nm))
1075                                 forbid_encrypt = nm->nmf_forbid_encryption;
1076                         if (forbid_encrypt)
1077                                 GOTO(put_nm, rc = -EACCES);
1078                 }
1079
1080                 if (!(reply->ocd_connect_flags & OBD_CONNECT_RDONLY)) {
1081                         bool readonly = false;
1082
1083                         if (!IS_ERR_OR_NULL(nm))
1084                                 readonly = nm->nmf_readonly_mount;
1085                         if (unlikely(readonly))
1086                                 GOTO(put_nm, rc = -EROFS);
1087                 }
1088
1089 put_nm:
1090                 if (!IS_ERR_OR_NULL(nm))
1091                         nodemap_putref(nm);
1092                 if (rc)
1093                         GOTO(out, rc);
1094         }
1095
1096         RETURN(0);
1097 out:
1098         obd_disconnect(class_export_get(tsi->tsi_exp));
1099         return rc;
1100 }
1101 EXPORT_SYMBOL(tgt_connect);
1102
1103 int tgt_disconnect(struct tgt_session_info *tsi)
1104 {
1105         int rc;
1106
1107         ENTRY;
1108
1109         OBD_FAIL_TIMEOUT(OBD_FAIL_OST_DISCONNECT_DELAY, cfs_fail_val);
1110
1111         rc = target_handle_disconnect(tgt_ses_req(tsi));
1112         if (rc)
1113                 RETURN(err_serious(rc));
1114
1115         RETURN(rc);
1116 }
1117 EXPORT_SYMBOL(tgt_disconnect);
1118
1119 /*
1120  * Unified target OBD handlers
1121  */
1122 int tgt_obd_ping(struct tgt_session_info *tsi)
1123 {
1124         int rc;
1125
1126         ENTRY;
1127
1128         /* The target-specific part of OBD_PING request handling.
1129          * It controls Filter Modification Data (FMD) expiration each time
1130          * PING is received.
1131          *
1132          * Valid only for replayable targets, e.g. MDT and OFD
1133          */
1134         if (tsi->tsi_exp->exp_obd->obd_replayable)
1135                 tgt_fmd_expire(tsi->tsi_exp);
1136
1137         rc = req_capsule_server_pack(tsi->tsi_pill);
1138         if (rc)
1139                 RETURN(err_serious(rc));
1140
1141         RETURN(rc);
1142 }
1143 EXPORT_SYMBOL(tgt_obd_ping);
1144
1145 int tgt_obd_log_cancel(struct tgt_session_info *tsi)
1146 {
1147         return err_serious(-EOPNOTSUPP);
1148 }
1149
1150 int tgt_send_buffer(struct tgt_session_info *tsi, struct lu_rdbuf *rdbuf)
1151 {
1152         struct ptlrpc_request   *req = tgt_ses_req(tsi);
1153         struct obd_export       *exp = req->rq_export;
1154         struct ptlrpc_bulk_desc *desc;
1155         int                      i;
1156         int                      rc;
1157         int                      pages = 0;
1158
1159         ENTRY;
1160
1161         for (i = 0; i < rdbuf->rb_nbufs; i++) {
1162                 unsigned int offset;
1163
1164                 offset = (unsigned long)rdbuf->rb_bufs[i].lb_buf & ~PAGE_MASK;
1165                 pages += DIV_ROUND_UP(rdbuf->rb_bufs[i].lb_len + offset,
1166                                       PAGE_SIZE);
1167         }
1168
1169         desc = ptlrpc_prep_bulk_exp(req, pages, 1,
1170                                   PTLRPC_BULK_PUT_SOURCE,
1171                                     MDS_BULK_PORTAL,
1172                                     &ptlrpc_bulk_kiov_nopin_ops);
1173         if (desc == NULL)
1174                 RETURN(-ENOMEM);
1175
1176         for (i = 0; i < rdbuf->rb_nbufs; i++)
1177                 desc->bd_frag_ops->add_iov_frag(desc,
1178                                         rdbuf->rb_bufs[i].lb_buf,
1179                                         rdbuf->rb_bufs[i].lb_len);
1180
1181         rc = target_bulk_io(exp, desc);
1182         ptlrpc_free_bulk(desc);
1183         RETURN(rc);
1184 }
1185 EXPORT_SYMBOL(tgt_send_buffer);
1186
1187 int tgt_sendpage(struct tgt_session_info *tsi, struct lu_rdpg *rdpg, int nob)
1188 {
1189         struct ptlrpc_request   *req = tgt_ses_req(tsi);
1190         struct obd_export       *exp = req->rq_export;
1191         struct ptlrpc_bulk_desc *desc;
1192         int                      tmpcount;
1193         int                      tmpsize;
1194         int                      i;
1195         int                      rc;
1196
1197         ENTRY;
1198
1199         desc = ptlrpc_prep_bulk_exp(req, rdpg->rp_npages, 1,
1200                                     PTLRPC_BULK_PUT_SOURCE,
1201                                     MDS_BULK_PORTAL,
1202                                     &ptlrpc_bulk_kiov_pin_ops);
1203         if (desc == NULL)
1204                 RETURN(-ENOMEM);
1205
1206         if (!(exp_connect_flags(exp) & OBD_CONNECT_BRW_SIZE))
1207                 /* old client requires reply size in it's PAGE_SIZE,
1208                  * which is rdpg->rp_count */
1209                 nob = rdpg->rp_count;
1210
1211         for (i = 0, tmpcount = nob; i < rdpg->rp_npages && tmpcount > 0;
1212              i++, tmpcount -= tmpsize) {
1213                 tmpsize = min_t(int, tmpcount, PAGE_SIZE);
1214                 desc->bd_frag_ops->add_kiov_frag(desc, rdpg->rp_pages[i], 0,
1215                                                  tmpsize);
1216         }
1217
1218         LASSERT(desc->bd_nob == nob);
1219         rc = target_bulk_io(exp, desc);
1220         ptlrpc_free_bulk(desc);
1221         RETURN(rc);
1222 }
1223 EXPORT_SYMBOL(tgt_sendpage);
1224
1225 /*
1226  * OBD_IDX_READ handler
1227  */
1228 static int tgt_obd_idx_read(struct tgt_session_info *tsi)
1229 {
1230         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
1231         struct lu_rdpg          *rdpg = &tti->tti_u.rdpg.tti_rdpg;
1232         struct idx_info         *req_ii, *rep_ii;
1233         int                      rc, i;
1234
1235         ENTRY;
1236
1237         memset(rdpg, 0, sizeof(*rdpg));
1238         req_capsule_set(tsi->tsi_pill, &RQF_OBD_IDX_READ);
1239
1240         /* extract idx_info buffer from request & reply */
1241         req_ii = req_capsule_client_get(tsi->tsi_pill, &RMF_IDX_INFO);
1242         if (req_ii == NULL || req_ii->ii_magic != IDX_INFO_MAGIC)
1243                 RETURN(err_serious(-EPROTO));
1244
1245         rc = req_capsule_server_pack(tsi->tsi_pill);
1246         if (rc)
1247                 RETURN(err_serious(rc));
1248
1249         rep_ii = req_capsule_server_get(tsi->tsi_pill, &RMF_IDX_INFO);
1250         if (rep_ii == NULL)
1251                 RETURN(err_serious(-EFAULT));
1252         rep_ii->ii_magic = IDX_INFO_MAGIC;
1253
1254         /* extract hash to start with */
1255         rdpg->rp_hash = req_ii->ii_hash_start;
1256
1257         /* extract requested attributes */
1258         rdpg->rp_attrs = req_ii->ii_attrs;
1259
1260         /* check that fid packed in request is valid and supported */
1261         if (!fid_is_sane(&req_ii->ii_fid))
1262                 RETURN(-EINVAL);
1263         rep_ii->ii_fid = req_ii->ii_fid;
1264
1265         /* copy flags */
1266         rep_ii->ii_flags = req_ii->ii_flags;
1267
1268         /* compute number of pages to allocate, ii_count is the number of 4KB
1269          * containers */
1270         if (req_ii->ii_count <= 0)
1271                 GOTO(out, rc = -EFAULT);
1272         rdpg->rp_count = min_t(unsigned int, req_ii->ii_count << LU_PAGE_SHIFT,
1273                                exp_max_brw_size(tsi->tsi_exp));
1274         rdpg->rp_npages = (rdpg->rp_count + PAGE_SIZE - 1) >> PAGE_SHIFT;
1275
1276         /* allocate pages to store the containers */
1277         OBD_ALLOC_PTR_ARRAY(rdpg->rp_pages, rdpg->rp_npages);
1278         if (rdpg->rp_pages == NULL)
1279                 GOTO(out, rc = -ENOMEM);
1280         for (i = 0; i < rdpg->rp_npages; i++) {
1281                 rdpg->rp_pages[i] = alloc_page(GFP_NOFS);
1282                 if (rdpg->rp_pages[i] == NULL)
1283                         GOTO(out, rc = -ENOMEM);
1284         }
1285
1286         /* populate pages with key/record pairs */
1287         rc = dt_index_read(tsi->tsi_env, tsi->tsi_tgt->lut_bottom, rep_ii, rdpg);
1288         if (rc < 0)
1289                 GOTO(out, rc);
1290
1291         LASSERTF(rc <= rdpg->rp_count, "dt_index_read() returned more than "
1292                  "asked %d > %d\n", rc, rdpg->rp_count);
1293
1294         /* send pages to client */
1295         rc = tgt_sendpage(tsi, rdpg, rc);
1296         if (rc)
1297                 GOTO(out, rc);
1298         EXIT;
1299 out:
1300         if (rdpg->rp_pages) {
1301                 for (i = 0; i < rdpg->rp_npages; i++)
1302                         if (rdpg->rp_pages[i])
1303                                 __free_page(rdpg->rp_pages[i]);
1304                 OBD_FREE_PTR_ARRAY(rdpg->rp_pages, rdpg->rp_npages);
1305         }
1306         return rc;
1307 }
1308
1309 struct tgt_handler tgt_obd_handlers[] = {
1310 TGT_OBD_HDL    (0,      OBD_PING,               tgt_obd_ping),
1311 TGT_OBD_HDL    (0,      OBD_IDX_READ,           tgt_obd_idx_read)
1312 };
1313 EXPORT_SYMBOL(tgt_obd_handlers);
1314
1315 int tgt_sync(const struct lu_env *env, struct lu_target *tgt,
1316              struct dt_object *obj, __u64 start, __u64 end)
1317 {
1318         int rc = 0;
1319
1320         ENTRY;
1321
1322         /* if no objid is specified, it means "sync whole filesystem" */
1323         if (obj == NULL) {
1324                 rc = dt_sync(env, tgt->lut_bottom);
1325         } else if (dt_version_get(env, obj) >
1326                    tgt->lut_obd->obd_last_committed) {
1327                 rc = dt_object_sync(env, obj, start, end);
1328         }
1329         atomic_inc(&tgt->lut_sync_count);
1330
1331         RETURN(rc);
1332 }
1333 EXPORT_SYMBOL(tgt_sync);
1334 /*
1335  * Unified target DLM handlers.
1336  */
1337
1338 /**
1339  * Unified target BAST
1340  *
1341  * Ensure data and metadata are synced to disk when lock is canceled if Sync on
1342  * Cancel (SOC) is enabled. If it's extent lock, normally sync obj is enough,
1343  * but if it's cross-MDT lock, because remote object version is not set, a
1344  * filesystem sync is needed.
1345  *
1346  * \param lock server side lock
1347  * \param desc lock desc
1348  * \param data ldlm_cb_set_arg
1349  * \param flag  indicates whether this cancelling or blocking callback
1350  * \retval      0 on success
1351  * \retval      negative number on error
1352  */
1353 static int tgt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
1354                             void *data, int flag)
1355 {
1356         struct lu_env            env;
1357         struct lu_target        *tgt;
1358         struct dt_object        *obj = NULL;
1359         struct lu_fid            fid;
1360         int                      rc = 0;
1361
1362         ENTRY;
1363
1364         tgt = class_exp2tgt(lock->l_export);
1365
1366         if (unlikely(tgt == NULL)) {
1367                 CDEBUG(D_ERROR, "%s: No target for connected export\n",
1368                        class_exp2obd(lock->l_export)->obd_name);
1369                 RETURN(-EINVAL);
1370         }
1371
1372         if (flag == LDLM_CB_CANCELING &&
1373             (lock->l_granted_mode & (LCK_EX | LCK_PW | LCK_GROUP)) &&
1374             (tgt->lut_sync_lock_cancel == SYNC_LOCK_CANCEL_ALWAYS ||
1375              (tgt->lut_sync_lock_cancel == SYNC_LOCK_CANCEL_BLOCKING &&
1376               ldlm_is_cbpending(lock))) &&
1377             ((exp_connect_flags(lock->l_export) & OBD_CONNECT_MDS_MDS) ||
1378              lock->l_resource->lr_type == LDLM_EXTENT)) {
1379                 __u64 start = 0;
1380                 __u64 end = OBD_OBJECT_EOF;
1381
1382                 rc = lu_env_init(&env, LCT_DT_THREAD);
1383                 if (unlikely(rc != 0))
1384                         GOTO(err, rc);
1385
1386                 ost_fid_from_resid(&fid, &lock->l_resource->lr_name,
1387                                    tgt->lut_lsd.lsd_osd_index);
1388
1389                 if (lock->l_resource->lr_type == LDLM_EXTENT) {
1390                         obj = dt_locate(&env, tgt->lut_bottom, &fid);
1391                         if (IS_ERR(obj))
1392                                 GOTO(err_env, rc = PTR_ERR(obj));
1393
1394                         if (!dt_object_exists(obj))
1395                                 GOTO(err_put, rc = -ENOENT);
1396
1397                         start = lock->l_policy_data.l_extent.start;
1398                         end = lock->l_policy_data.l_extent.end;
1399                 }
1400
1401                 rc = tgt_sync(&env, tgt, obj, start, end);
1402                 if (rc < 0) {
1403                         CERROR("%s: syncing "DFID" (%llu-%llu) on lock "
1404                                "cancel: rc = %d\n",
1405                                tgt_name(tgt), PFID(&fid),
1406                                lock->l_policy_data.l_extent.start,
1407                                lock->l_policy_data.l_extent.end, rc);
1408                 }
1409 err_put:
1410                 if (obj != NULL)
1411                         dt_object_put(&env, obj);
1412 err_env:
1413                 lu_env_fini(&env);
1414         }
1415 err:
1416         rc = ldlm_server_blocking_ast(lock, desc, data, flag);
1417         RETURN(rc);
1418 }
1419
1420 static struct ldlm_callback_suite tgt_dlm_cbs = {
1421         .lcs_completion = ldlm_server_completion_ast,
1422         .lcs_blocking   = tgt_blocking_ast,
1423         .lcs_glimpse    = ldlm_server_glimpse_ast
1424 };
1425
1426 int tgt_enqueue(struct tgt_session_info *tsi)
1427 {
1428         struct ptlrpc_request *req = tgt_ses_req(tsi);
1429         int rc;
1430
1431         ENTRY;
1432         /*
1433          * tsi->tsi_dlm_req was already swapped and (if necessary) converted,
1434          * tsi->tsi_dlm_cbs was set by the *_req_handle() function.
1435          */
1436         LASSERT(tsi->tsi_dlm_req != NULL);
1437         rc = ldlm_handle_enqueue0(tsi->tsi_exp->exp_obd->obd_namespace, req,
1438                                   tsi->tsi_dlm_req, &tgt_dlm_cbs);
1439         if (rc)
1440                 RETURN(err_serious(rc));
1441
1442         switch (LUT_FAIL_CLASS(tsi->tsi_reply_fail_id)) {
1443         case LUT_FAIL_MDT:
1444                 tsi->tsi_reply_fail_id = OBD_FAIL_MDS_LDLM_REPLY_NET;
1445                 break;
1446         case LUT_FAIL_OST:
1447                 tsi->tsi_reply_fail_id = OBD_FAIL_OST_LDLM_REPLY_NET;
1448                 break;
1449         case LUT_FAIL_MGT:
1450                 tsi->tsi_reply_fail_id = OBD_FAIL_MGS_LDLM_REPLY_NET;
1451                 break;
1452         default:
1453                 tsi->tsi_reply_fail_id = OBD_FAIL_LDLM_REPLY;
1454                 break;
1455         }
1456         RETURN(req->rq_status);
1457 }
1458 EXPORT_SYMBOL(tgt_enqueue);
1459
1460 int tgt_convert(struct tgt_session_info *tsi)
1461 {
1462         struct ptlrpc_request *req = tgt_ses_req(tsi);
1463         int rc;
1464
1465         ENTRY;
1466         LASSERT(tsi->tsi_dlm_req);
1467         rc = ldlm_handle_convert0(req, tsi->tsi_dlm_req);
1468         if (rc)
1469                 RETURN(err_serious(rc));
1470
1471         RETURN(req->rq_status);
1472 }
1473
1474 int tgt_bl_callback(struct tgt_session_info *tsi)
1475 {
1476         return err_serious(-EOPNOTSUPP);
1477 }
1478
1479 int tgt_cp_callback(struct tgt_session_info *tsi)
1480 {
1481         return err_serious(-EOPNOTSUPP);
1482 }
1483
1484 /* generic LDLM target handler */
1485 struct tgt_handler tgt_dlm_handlers[] = {
1486 TGT_DLM_HDL(HAS_KEY, LDLM_ENQUEUE, tgt_enqueue),
1487 TGT_DLM_HDL(HAS_KEY, LDLM_CONVERT, tgt_convert),
1488 TGT_DLM_HDL_VAR(0, LDLM_BL_CALLBACK, tgt_bl_callback),
1489 TGT_DLM_HDL_VAR(0, LDLM_CP_CALLBACK, tgt_cp_callback)
1490 };
1491 EXPORT_SYMBOL(tgt_dlm_handlers);
1492
1493 /*
1494  * Unified target LLOG handlers.
1495  */
1496 int tgt_llog_open(struct tgt_session_info *tsi)
1497 {
1498         int rc;
1499
1500         ENTRY;
1501
1502         rc = llog_origin_handle_open(tgt_ses_req(tsi));
1503
1504         RETURN(rc);
1505 }
1506 EXPORT_SYMBOL(tgt_llog_open);
1507
1508 int tgt_llog_read_header(struct tgt_session_info *tsi)
1509 {
1510         int rc;
1511
1512         ENTRY;
1513
1514         rc = llog_origin_handle_read_header(tgt_ses_req(tsi));
1515
1516         RETURN(rc);
1517 }
1518 EXPORT_SYMBOL(tgt_llog_read_header);
1519
1520 int tgt_llog_next_block(struct tgt_session_info *tsi)
1521 {
1522         int rc;
1523
1524         ENTRY;
1525
1526         rc = llog_origin_handle_next_block(tgt_ses_req(tsi));
1527
1528         RETURN(rc);
1529 }
1530 EXPORT_SYMBOL(tgt_llog_next_block);
1531
1532 int tgt_llog_prev_block(struct tgt_session_info *tsi)
1533 {
1534         int rc;
1535
1536         ENTRY;
1537
1538         rc = llog_origin_handle_prev_block(tgt_ses_req(tsi));
1539
1540         RETURN(rc);
1541 }
1542 EXPORT_SYMBOL(tgt_llog_prev_block);
1543
1544 /* generic llog target handler */
1545 struct tgt_handler tgt_llog_handlers[] = {
1546 TGT_LLOG_HDL    (0,     LLOG_ORIGIN_HANDLE_CREATE,      tgt_llog_open),
1547 TGT_LLOG_HDL    (0,     LLOG_ORIGIN_HANDLE_NEXT_BLOCK,  tgt_llog_next_block),
1548 TGT_LLOG_HDL    (0,     LLOG_ORIGIN_HANDLE_READ_HEADER, tgt_llog_read_header),
1549 TGT_LLOG_HDL    (0,     LLOG_ORIGIN_HANDLE_PREV_BLOCK,  tgt_llog_prev_block),
1550 };
1551 EXPORT_SYMBOL(tgt_llog_handlers);
1552
1553 /*
1554  * sec context handlers
1555  */
1556 /* XXX: Implement based on mdt_sec_ctx_handle()? */
1557 static int tgt_sec_ctx_handle(struct tgt_session_info *tsi)
1558 {
1559         return 0;
1560 }
1561
1562 struct tgt_handler tgt_sec_ctx_handlers[] = {
1563 TGT_SEC_HDL_VAR(0,      SEC_CTX_INIT,           tgt_sec_ctx_handle),
1564 TGT_SEC_HDL_VAR(0,      SEC_CTX_INIT_CONT,      tgt_sec_ctx_handle),
1565 TGT_SEC_HDL_VAR(0,      SEC_CTX_FINI,           tgt_sec_ctx_handle),
1566 };
1567 EXPORT_SYMBOL(tgt_sec_ctx_handlers);
1568
1569 int (*tgt_lfsck_in_notify_local)(const struct lu_env *env,
1570                                  struct dt_device *key,
1571                                  struct lfsck_req_local *lrl,
1572                                  struct thandle *th) = NULL;
1573
1574 void tgt_register_lfsck_in_notify_local(int (*notify)(const struct lu_env *,
1575                                                       struct dt_device *,
1576                                                       struct lfsck_req_local *,
1577                                                       struct thandle *))
1578 {
1579         tgt_lfsck_in_notify_local = notify;
1580 }
1581 EXPORT_SYMBOL(tgt_register_lfsck_in_notify_local);
1582
1583 int (*tgt_lfsck_in_notify)(const struct lu_env *env,
1584                            struct dt_device *key,
1585                            struct lfsck_request *lr) = NULL;
1586
1587 void tgt_register_lfsck_in_notify(int (*notify)(const struct lu_env *,
1588                                                 struct dt_device *,
1589                                                 struct lfsck_request *))
1590 {
1591         tgt_lfsck_in_notify = notify;
1592 }
1593 EXPORT_SYMBOL(tgt_register_lfsck_in_notify);
1594
1595 static int (*tgt_lfsck_query)(const struct lu_env *env,
1596                               struct dt_device *key,
1597                               struct lfsck_request *req,
1598                               struct lfsck_reply *rep,
1599                               struct lfsck_query *que) = NULL;
1600
1601 void tgt_register_lfsck_query(int (*query)(const struct lu_env *,
1602                                            struct dt_device *,
1603                                            struct lfsck_request *,
1604                                            struct lfsck_reply *,
1605                                            struct lfsck_query *))
1606 {
1607         tgt_lfsck_query = query;
1608 }
1609 EXPORT_SYMBOL(tgt_register_lfsck_query);
1610
1611 /* LFSCK request handlers */
1612 static int tgt_handle_lfsck_notify(struct tgt_session_info *tsi)
1613 {
1614         const struct lu_env     *env = tsi->tsi_env;
1615         struct dt_device        *key = tsi->tsi_tgt->lut_bottom;
1616         struct lfsck_request    *lr;
1617         int                      rc;
1618         ENTRY;
1619
1620         lr = req_capsule_client_get(tsi->tsi_pill, &RMF_LFSCK_REQUEST);
1621         if (lr == NULL)
1622                 RETURN(-EPROTO);
1623
1624         rc = tgt_lfsck_in_notify(env, key, lr);
1625
1626         RETURN(rc);
1627 }
1628
1629 static int tgt_handle_lfsck_query(struct tgt_session_info *tsi)
1630 {
1631         struct lfsck_request    *request;
1632         struct lfsck_reply      *reply;
1633         int                      rc;
1634         ENTRY;
1635
1636         request = req_capsule_client_get(tsi->tsi_pill, &RMF_LFSCK_REQUEST);
1637         if (request == NULL)
1638                 RETURN(-EPROTO);
1639
1640         reply = req_capsule_server_get(tsi->tsi_pill, &RMF_LFSCK_REPLY);
1641         if (reply == NULL)
1642                 RETURN(-ENOMEM);
1643
1644         rc = tgt_lfsck_query(tsi->tsi_env, tsi->tsi_tgt->lut_bottom,
1645                              request, reply, NULL);
1646
1647         RETURN(rc < 0 ? rc : 0);
1648 }
1649
1650 struct tgt_handler tgt_lfsck_handlers[] = {
1651 TGT_LFSCK_HDL(HAS_REPLY,        LFSCK_NOTIFY,   tgt_handle_lfsck_notify),
1652 TGT_LFSCK_HDL(HAS_REPLY,        LFSCK_QUERY,    tgt_handle_lfsck_query),
1653 };
1654 EXPORT_SYMBOL(tgt_lfsck_handlers);
1655
1656 /*
1657  * initialize per-thread page pool (bug 5137).
1658  */
1659 int tgt_io_thread_init(struct ptlrpc_thread *thread)
1660 {
1661         struct tgt_thread_big_cache *tbc;
1662
1663         ENTRY;
1664
1665         LASSERT(thread != NULL);
1666         LASSERT(thread->t_data == NULL);
1667
1668         OBD_ALLOC_LARGE(tbc, sizeof(*tbc));
1669         if (tbc == NULL)
1670                 RETURN(-ENOMEM);
1671         thread->t_data = tbc;
1672         RETURN(0);
1673 }
1674 EXPORT_SYMBOL(tgt_io_thread_init);
1675
1676 /*
1677  * free per-thread pool created by tgt_thread_init().
1678  */
1679 void tgt_io_thread_done(struct ptlrpc_thread *thread)
1680 {
1681         struct tgt_thread_big_cache *tbc;
1682
1683         ENTRY;
1684
1685         LASSERT(thread != NULL);
1686
1687         /*
1688          * be prepared to handle partially-initialized pools (because this is
1689          * called from ost_io_thread_init() for cleanup.
1690          */
1691         tbc = thread->t_data;
1692         if (tbc != NULL) {
1693                 OBD_FREE_LARGE(tbc, sizeof(*tbc));
1694                 thread->t_data = NULL;
1695         }
1696         EXIT;
1697 }
1698 EXPORT_SYMBOL(tgt_io_thread_done);
1699
1700 /**
1701  * Helper function for getting Data-on-MDT file server DLM lock
1702  * if asked by client.
1703  */
1704 int tgt_mdt_data_lock(struct ldlm_namespace *ns, struct ldlm_res_id *res_id,
1705                       struct lustre_handle *lh, int mode, __u64 *flags)
1706 {
1707         union ldlm_policy_data policy = {
1708                 .l_inodebits.bits = MDS_INODELOCK_DOM,
1709         };
1710         int rc;
1711
1712         ENTRY;
1713
1714         LASSERT(lh != NULL);
1715         LASSERT(ns != NULL);
1716         LASSERT(!lustre_handle_is_used(lh));
1717
1718         rc = ldlm_cli_enqueue_local(NULL, ns, res_id, LDLM_IBITS, &policy, mode,
1719                                     flags, ldlm_blocking_ast,
1720                                     ldlm_completion_ast, ldlm_glimpse_ast,
1721                                     NULL, 0, LVB_T_NONE, NULL, lh);
1722
1723         RETURN(rc == ELDLM_OK ? 0 : -EIO);
1724 }
1725 EXPORT_SYMBOL(tgt_mdt_data_lock);
1726
1727 /**
1728  * Helper function for getting server side [start, start+count] DLM lock
1729  * if asked by client.
1730  */
1731 int tgt_extent_lock(const struct lu_env *env, struct ldlm_namespace *ns,
1732                     struct ldlm_res_id *res_id, __u64 start, __u64 end,
1733                     struct lustre_handle *lh, int mode, __u64 *flags)
1734 {
1735         union ldlm_policy_data policy;
1736         int rc;
1737
1738         ENTRY;
1739
1740         LASSERT(lh != NULL);
1741         LASSERT(ns != NULL);
1742         LASSERT(!lustre_handle_is_used(lh));
1743
1744         policy.l_extent.gid = 0;
1745         policy.l_extent.start = start & PAGE_MASK;
1746
1747         /*
1748          * If ->o_blocks is EOF it means "lock till the end of the file".
1749          * Otherwise, it's size of an extent or hole being punched (in bytes).
1750          */
1751         if (end == OBD_OBJECT_EOF || end < start)
1752                 policy.l_extent.end = OBD_OBJECT_EOF;
1753         else
1754                 policy.l_extent.end = end | ~PAGE_MASK;
1755
1756         rc = ldlm_cli_enqueue_local(env, ns, res_id, LDLM_EXTENT, &policy,
1757                                     mode, flags, ldlm_blocking_ast,
1758                                     ldlm_completion_ast, ldlm_glimpse_ast,
1759                                     NULL, 0, LVB_T_NONE, NULL, lh);
1760         RETURN(rc == ELDLM_OK ? 0 : -EIO);
1761 }
1762 EXPORT_SYMBOL(tgt_extent_lock);
1763
1764 static int tgt_data_lock(const struct lu_env *env, struct obd_export *exp,
1765                          struct ldlm_res_id *res_id, __u64 start, __u64 end,
1766                          struct lustre_handle *lh, enum ldlm_mode mode)
1767 {
1768         struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
1769         __u64 flags = 0;
1770
1771         /* MDT IO for data-on-mdt */
1772         if (exp->exp_connect_data.ocd_connect_flags & OBD_CONNECT_IBITS)
1773                 return tgt_mdt_data_lock(ns, res_id, lh, mode, &flags);
1774
1775         return tgt_extent_lock(env, ns, res_id, start, end, lh, mode, &flags);
1776 }
1777
1778 void tgt_data_unlock(struct lustre_handle *lh, enum ldlm_mode mode)
1779 {
1780         LASSERT(lustre_handle_is_used(lh));
1781         ldlm_lock_decref(lh, mode);
1782 }
1783 EXPORT_SYMBOL(tgt_data_unlock);
1784
1785 static int tgt_brw_lock(const struct lu_env *env, struct obd_export *exp,
1786                         struct ldlm_res_id *res_id, struct obd_ioobj *obj,
1787                         struct niobuf_remote *nb, struct lustre_handle *lh,
1788                         enum ldlm_mode mode)
1789 {
1790         int nrbufs = obj->ioo_bufcnt;
1791         int i;
1792
1793         ENTRY;
1794
1795         LASSERT(mode == LCK_PR || mode == LCK_PW);
1796         LASSERT(!lustre_handle_is_used(lh));
1797
1798         if (exp->exp_obd->obd_recovering)
1799                 RETURN(0);
1800
1801         if (nrbufs == 0 || !(nb[0].rnb_flags & OBD_BRW_SRVLOCK))
1802                 RETURN(0);
1803
1804         for (i = 1; i < nrbufs; i++)
1805                 if (!(nb[i].rnb_flags & OBD_BRW_SRVLOCK))
1806                         RETURN(-EFAULT);
1807
1808         return tgt_data_lock(env, exp, res_id, nb[0].rnb_offset,
1809                              nb[nrbufs - 1].rnb_offset +
1810                              nb[nrbufs - 1].rnb_len - 1, lh, mode);
1811 }
1812
1813 static void tgt_brw_unlock(struct obd_export *exp, struct obd_ioobj *obj,
1814                            struct niobuf_remote *niob,
1815                            struct lustre_handle *lh, enum ldlm_mode mode)
1816 {
1817         ENTRY;
1818
1819         LASSERT(mode == LCK_PR || mode == LCK_PW);
1820         LASSERT((!exp->exp_obd->obd_recovering && obj->ioo_bufcnt &&
1821                  niob[0].rnb_flags & OBD_BRW_SRVLOCK) ==
1822                 lustre_handle_is_used(lh));
1823
1824         if (lustre_handle_is_used(lh))
1825                 tgt_data_unlock(lh, mode);
1826         EXIT;
1827 }
1828
1829 static int tgt_checksum_niobuf(struct lu_target *tgt,
1830                                  struct niobuf_local *local_nb, int npages,
1831                                  int opc, enum cksum_types cksum_type,
1832                                  __u32 *cksum)
1833 {
1834         struct ahash_request           *req;
1835         unsigned int                    bufsize;
1836         int                             i, err;
1837         unsigned char                   cfs_alg = cksum_obd2cfs(cksum_type);
1838
1839         req = cfs_crypto_hash_init(cfs_alg, NULL, 0);
1840         if (IS_ERR(req)) {
1841                 CERROR("%s: unable to initialize checksum hash %s\n",
1842                        tgt_name(tgt), cfs_crypto_hash_name(cfs_alg));
1843                 return PTR_ERR(req);
1844         }
1845
1846         CDEBUG(D_INFO, "Checksum for algo %s\n", cfs_crypto_hash_name(cfs_alg));
1847         for (i = 0; i < npages; i++) {
1848                 /* corrupt the data before we compute the checksum, to
1849                  * simulate a client->OST data error */
1850                 if (i == 0 && opc == OST_WRITE &&
1851                     OBD_FAIL_CHECK(OBD_FAIL_OST_CHECKSUM_RECEIVE)) {
1852                         int off = local_nb[i].lnb_page_offset & ~PAGE_MASK;
1853                         int len = local_nb[i].lnb_len;
1854                         struct page *np = tgt_page_to_corrupt;
1855
1856                         if (np) {
1857                                 char *ptr = kmap_atomic(local_nb[i].lnb_page);
1858                                 char *ptr2 = page_address(np);
1859
1860                                 memcpy(ptr2 + off, ptr + off, len);
1861                                 memcpy(ptr2 + off, "bad3", min(4, len));
1862                                 kunmap_atomic(ptr);
1863
1864                                 /* LU-8376 to preserve original index for
1865                                  * display in dump_all_bulk_pages() */
1866                                 np->index = i;
1867
1868                                 cfs_crypto_hash_update_page(req, np, off,
1869                                                             len);
1870                                 continue;
1871                         } else {
1872                                 CERROR("%s: can't alloc page for corruption\n",
1873                                        tgt_name(tgt));
1874                         }
1875                 }
1876                 cfs_crypto_hash_update_page(req, local_nb[i].lnb_page,
1877                                   local_nb[i].lnb_page_offset & ~PAGE_MASK,
1878                                   local_nb[i].lnb_len);
1879
1880                  /* corrupt the data after we compute the checksum, to
1881                  * simulate an OST->client data error */
1882                 if (i == 0 && opc == OST_READ &&
1883                     OBD_FAIL_CHECK(OBD_FAIL_OST_CHECKSUM_SEND)) {
1884                         int off = local_nb[i].lnb_page_offset & ~PAGE_MASK;
1885                         int len = local_nb[i].lnb_len;
1886                         struct page *np = tgt_page_to_corrupt;
1887
1888                         if (np) {
1889                                 char *ptr = kmap_atomic(local_nb[i].lnb_page);
1890                                 char *ptr2 = page_address(np);
1891
1892                                 memcpy(ptr2 + off, ptr + off, len);
1893                                 memcpy(ptr2 + off, "bad4", min(4, len));
1894                                 kunmap_atomic(ptr);
1895
1896                                 /* LU-8376 to preserve original index for
1897                                  * display in dump_all_bulk_pages() */
1898                                 np->index = i;
1899
1900                                 cfs_crypto_hash_update_page(req, np, off,
1901                                                             len);
1902                                 continue;
1903                         } else {
1904                                 CERROR("%s: can't alloc page for corruption\n",
1905                                        tgt_name(tgt));
1906                         }
1907                 }
1908         }
1909
1910         bufsize = sizeof(*cksum);
1911         err = cfs_crypto_hash_final(req, (unsigned char *)cksum, &bufsize);
1912
1913         return 0;
1914 }
1915
1916 char dbgcksum_file_name[PATH_MAX];
1917
1918 static void dump_all_bulk_pages(struct obdo *oa, int count,
1919                                 struct niobuf_local *local_nb,
1920                                 __u32 server_cksum, __u32 client_cksum)
1921 {
1922         struct file *filp;
1923         int rc, i;
1924         unsigned int len;
1925         char *buf;
1926
1927         /* will only keep dump of pages on first error for the same range in
1928          * file/fid, not during the resends/retries. */
1929         snprintf(dbgcksum_file_name, sizeof(dbgcksum_file_name),
1930                  "%s-checksum_dump-ost-"DFID":[%llu-%llu]-%x-%x",
1931                  (strncmp(libcfs_debug_file_path, "NONE", 4) != 0 ?
1932                   libcfs_debug_file_path : LIBCFS_DEBUG_FILE_PATH_DEFAULT),
1933                  oa->o_valid & OBD_MD_FLFID ? oa->o_parent_seq : (__u64)0,
1934                  oa->o_valid & OBD_MD_FLFID ? oa->o_parent_oid : 0,
1935                  oa->o_valid & OBD_MD_FLFID ? oa->o_parent_ver : 0,
1936                  local_nb[0].lnb_file_offset,
1937                  local_nb[count-1].lnb_file_offset +
1938                  local_nb[count-1].lnb_len - 1, client_cksum, server_cksum);
1939         CWARN("dumping checksum data to %s\n", dbgcksum_file_name);
1940         filp = filp_open(dbgcksum_file_name,
1941                          O_CREAT | O_EXCL | O_WRONLY | O_LARGEFILE, 0600);
1942         if (IS_ERR(filp)) {
1943                 rc = PTR_ERR(filp);
1944                 if (rc == -EEXIST)
1945                         CDEBUG(D_INFO, "%s: can't open to dump pages with "
1946                                "checksum error: rc = %d\n", dbgcksum_file_name,
1947                                rc);
1948                 else
1949                         CERROR("%s: can't open to dump pages with checksum "
1950                                "error: rc = %d\n", dbgcksum_file_name, rc);
1951                 return;
1952         }
1953
1954         for (i = 0; i < count; i++) {
1955                 len = local_nb[i].lnb_len;
1956                 buf = kmap(local_nb[i].lnb_page);
1957                 while (len != 0) {
1958                         rc = cfs_kernel_write(filp, buf, len, &filp->f_pos);
1959                         if (rc < 0) {
1960                                 CERROR("%s: wanted to write %u but got %d "
1961                                        "error\n", dbgcksum_file_name, len, rc);
1962                                 break;
1963                         }
1964                         len -= rc;
1965                         buf += rc;
1966                 }
1967                 kunmap(local_nb[i].lnb_page);
1968         }
1969
1970         rc = vfs_fsync_range(filp, 0, LLONG_MAX, 1);
1971         if (rc)
1972                 CERROR("%s: sync returns %d\n", dbgcksum_file_name, rc);
1973         filp_close(filp, NULL);
1974
1975         libcfs_debug_dumplog();
1976 }
1977
1978 static int check_read_checksum(struct niobuf_local *local_nb, int npages,
1979                                struct obd_export *exp, struct obdo *oa,
1980                                const struct lnet_processid *peer,
1981                                __u32 client_cksum, __u32 server_cksum,
1982                                enum cksum_types server_cksum_type)
1983 {
1984         char *msg;
1985         enum cksum_types cksum_type;
1986         loff_t start, end;
1987
1988         if (unlikely(npages <= 0))
1989                 return 0;
1990
1991         /* unlikely to happen and only if resend does not occur due to cksum
1992          * control failure on Client */
1993         if (unlikely(server_cksum == client_cksum)) {
1994                 CDEBUG(D_PAGE, "checksum %x confirmed upon retry\n",
1995                        client_cksum);
1996                 return 0;
1997         }
1998
1999         if (exp->exp_obd->obd_checksum_dump)
2000                 dump_all_bulk_pages(oa, npages, local_nb, server_cksum,
2001                                     client_cksum);
2002
2003         cksum_type = obd_cksum_type_unpack(oa->o_valid & OBD_MD_FLFLAGS ?
2004                                            oa->o_flags : 0);
2005
2006         if (cksum_type != server_cksum_type)
2007                 msg = "the server may have not used the checksum type specified"
2008                       " in the original request - likely a protocol problem";
2009         else
2010                 msg = "should have changed on the client or in transit";
2011
2012         start = local_nb[0].lnb_file_offset;
2013         end = local_nb[npages-1].lnb_file_offset +
2014                                         local_nb[npages-1].lnb_len - 1;
2015
2016         LCONSOLE_ERROR_MSG(0x132, "%s: BAD READ CHECKSUM: %s: from %s inode "
2017                 DFID " object "DOSTID" extent [%llu-%llu], client returned csum"
2018                 " %x (type %x), server csum %x (type %x)\n",
2019                 exp->exp_obd->obd_name,
2020                 msg, libcfs_nidstr(&peer->nid),
2021                 oa->o_valid & OBD_MD_FLFID ? oa->o_parent_seq : 0ULL,
2022                 oa->o_valid & OBD_MD_FLFID ? oa->o_parent_oid : 0,
2023                 oa->o_valid & OBD_MD_FLFID ? oa->o_parent_ver : 0,
2024                 POSTID(&oa->o_oi),
2025                 start, end, client_cksum, cksum_type, server_cksum,
2026                 server_cksum_type);
2027
2028         return 1;
2029 }
2030
2031 static int tgt_pages2shortio(struct niobuf_local *local, int npages,
2032                              unsigned char *buf, int size)
2033 {
2034         int     i, off, len, copied = size;
2035         char    *ptr;
2036
2037         for (i = 0; i < npages; i++) {
2038                 off = local[i].lnb_page_offset & ~PAGE_MASK;
2039                 len = local[i].lnb_len;
2040
2041                 CDEBUG(D_PAGE, "index %d offset = %d len = %d left = %d\n",
2042                        i, off, len, size);
2043                 if (len > size)
2044                         return -EINVAL;
2045
2046                 ptr = kmap_atomic(local[i].lnb_page);
2047                 memcpy(buf, ptr + off, len);
2048                 kunmap_atomic(ptr);
2049                 buf += len;
2050                 size -= len;
2051         }
2052         return copied - size;
2053 }
2054
2055 static int tgt_checksum_niobuf_t10pi(struct lu_target *tgt,
2056                                      struct niobuf_local *local_nb, int npages,
2057                                      int opc, obd_dif_csum_fn *fn,
2058                                      int sector_size, u32 *check_sum,
2059                                      bool resend)
2060 {
2061         enum cksum_types t10_cksum_type = tgt->lut_dt_conf.ddp_t10_cksum_type;
2062         unsigned char cfs_alg = cksum_obd2cfs(OBD_CKSUM_T10_TOP);
2063         const char *obd_name = tgt->lut_obd->obd_name;
2064         struct ahash_request *req;
2065         unsigned int bufsize;
2066         unsigned char *buffer;
2067         struct page *__page;
2068         __u16 *guard_start;
2069         int guard_number;
2070         int used_number = 0;
2071         __u32 cksum;
2072         int rc = 0;
2073         int used;
2074         int i;
2075
2076         __page = alloc_page(GFP_KERNEL);
2077         if (__page == NULL)
2078                 return -ENOMEM;
2079
2080         req = cfs_crypto_hash_init(cfs_alg, NULL, 0);
2081         if (IS_ERR(req)) {
2082                 rc = PTR_ERR(req);
2083                 CERROR("%s: unable to initialize checksum hash %s: rc = %d\n",
2084                        tgt_name(tgt), cfs_crypto_hash_name(cfs_alg), rc);
2085                 goto out;
2086         }
2087
2088         buffer = kmap(__page);
2089         guard_start = (__u16 *)buffer;
2090         guard_number = PAGE_SIZE / sizeof(*guard_start);
2091         if (unlikely(resend))
2092                 CDEBUG(D_PAGE | D_HA, "GRD tags per page = %u\n", guard_number);
2093         for (i = 0; i < npages; i++) {
2094                 bool use_t10_grd;
2095
2096                 /* corrupt the data before we compute the checksum, to
2097                  * simulate a client->OST data error */
2098                 if (i == 0 && opc == OST_WRITE &&
2099                     OBD_FAIL_CHECK(OBD_FAIL_OST_CHECKSUM_RECEIVE)) {
2100                         int off = local_nb[i].lnb_page_offset & ~PAGE_MASK;
2101                         int len = local_nb[i].lnb_len;
2102                         struct page *np = tgt_page_to_corrupt;
2103
2104                         if (np) {
2105                                 char *ptr = kmap_atomic(local_nb[i].lnb_page);
2106                                 char *ptr2 = page_address(np);
2107
2108                                 memcpy(ptr2 + off, ptr + off, len);
2109                                 memcpy(ptr2 + off, "bad3", min(4, len));
2110                                 kunmap_atomic(ptr);
2111
2112                                 /* LU-8376 to preserve original index for
2113                                  * display in dump_all_bulk_pages() */
2114                                 np->index = i;
2115
2116                                 cfs_crypto_hash_update_page(req, np, off,
2117                                                             len);
2118                                 continue;
2119                         } else {
2120                                 CERROR("%s: can't alloc page for corruption\n",
2121                                        tgt_name(tgt));
2122                         }
2123                 }
2124
2125                 /*
2126                  * The left guard number should be able to hold checksums of a
2127                  * whole page
2128                  */
2129                 use_t10_grd = t10_cksum_type && opc == OST_READ &&
2130                               local_nb[i].lnb_len == PAGE_SIZE &&
2131                               local_nb[i].lnb_guard_disk;
2132                 if (use_t10_grd) {
2133                         used = DIV_ROUND_UP(local_nb[i].lnb_len, sector_size);
2134                         if (used > (guard_number - used_number)) {
2135                                 rc = -E2BIG;
2136                                 break;
2137                         }
2138                         memcpy(guard_start + used_number,
2139                                local_nb[i].lnb_guards,
2140                                used * sizeof(*guard_start));
2141                         if (unlikely(resend))
2142                                 CDEBUG(D_PAGE | D_HA,
2143                                        "lnb[%u]: used %u off %u+%u lnb checksum: %*phN\n",
2144                                        i, used,
2145                                        local_nb[i].lnb_page_offset,
2146                                        local_nb[i].lnb_len,
2147                                        (int)(used * sizeof(*guard_start)),
2148                                        guard_start + used_number);
2149                 }
2150                 if (!use_t10_grd || unlikely(resend)) {
2151                         __u16 guard_tmp[MAX_GUARD_NUMBER];
2152                         __u16 *guards = guard_start + used_number;
2153                         int used_tmp = -1, *usedp = &used;
2154
2155                         if (unlikely(use_t10_grd)) {
2156                                 guards = guard_tmp;
2157                                 usedp = &used_tmp;
2158                         }
2159                         rc = obd_page_dif_generate_buffer(obd_name,
2160                                 local_nb[i].lnb_page,
2161                                 local_nb[i].lnb_page_offset & ~PAGE_MASK,
2162                                 local_nb[i].lnb_len, guards,
2163                                 guard_number - used_number, usedp, sector_size,
2164                                 fn);
2165                         if (unlikely(resend)) {
2166                                 bool bad = use_t10_grd &&
2167                                         memcmp(guard_tmp,
2168                                                local_nb[i].lnb_guards,
2169                                                used_tmp * sizeof(*guard_tmp));
2170
2171                                 if (bad)
2172                                         CERROR("lnb[%u]: used %u/%u off %u+%u tmp checksum: %*phN\n",
2173                                                i, used, used_tmp,
2174                                                local_nb[i].lnb_page_offset,
2175                                                local_nb[i].lnb_len,
2176                                                (int)(used_tmp * sizeof(*guard_start)),
2177                                                guard_tmp);
2178                                 CDEBUG_LIMIT(D_PAGE | D_HA | (bad ? D_ERROR : 0),
2179                                        "lnb[%u]: used %u/%u off %u+%u gen checksum: %*phN\n",
2180                                        i, used, used_tmp,
2181                                        local_nb[i].lnb_page_offset,
2182                                        local_nb[i].lnb_len,
2183                                        (int)(used * sizeof(*guard_start)),
2184                                        guard_start + used_number);
2185                         }
2186                         if (rc)
2187                                 break;
2188                 }
2189
2190                 LASSERT(used <= MAX_GUARD_NUMBER);
2191                 /*
2192                  * If disk support T10PI checksum, copy guards to local_nb.
2193                  * If the write is partial page, do not use the guards for bio
2194                  * submission since the data might not be full-sector. The bio
2195                  * guards will be generated later based on the full sectors. If
2196                  * the sector size is 512B rather than 4 KB, or the page size
2197                  * is larger than 4KB, this might drop some useful guards for
2198                  * partial page write, but it will only add minimal extra time
2199                  * of checksum calculation.
2200                  */
2201                 if (t10_cksum_type && opc == OST_WRITE &&
2202                     local_nb[i].lnb_len == PAGE_SIZE) {
2203                         local_nb[i].lnb_guard_rpc = 1;
2204                         memcpy(local_nb[i].lnb_guards,
2205                                guard_start + used_number,
2206                                used * sizeof(*local_nb[i].lnb_guards));
2207                 }
2208
2209                 used_number += used;
2210                 if (used_number == guard_number) {
2211                         cfs_crypto_hash_update_page(req, __page, 0,
2212                                 used_number * sizeof(*guard_start));
2213                         used_number = 0;
2214                 }
2215
2216                  /* corrupt the data after we compute the checksum, to
2217                  * simulate an OST->client data error */
2218                 if (unlikely(i == 0 && opc == OST_READ &&
2219                              OBD_FAIL_CHECK(OBD_FAIL_OST_CHECKSUM_SEND))) {
2220                         int off = local_nb[i].lnb_page_offset & ~PAGE_MASK;
2221                         int len = local_nb[i].lnb_len;
2222                         struct page *np = tgt_page_to_corrupt;
2223
2224                         if (np) {
2225                                 char *ptr = kmap_atomic(local_nb[i].lnb_page);
2226                                 char *ptr2 = page_address(np);
2227
2228                                 memcpy(ptr2 + off, ptr + off, len);
2229                                 memcpy(ptr2 + off, "bad4", min(4, len));
2230                                 kunmap_atomic(ptr);
2231
2232                                 /* LU-8376 to preserve original index for
2233                                  * display in dump_all_bulk_pages() */
2234                                 np->index = i;
2235
2236                                 cfs_crypto_hash_update_page(req, np, off,
2237                                                             len);
2238                                 continue;
2239                         } else {
2240                                 CERROR("%s: can't alloc page for corruption\n",
2241                                        tgt_name(tgt));
2242                         }
2243                 }
2244         }
2245         kunmap(__page);
2246         if (rc)
2247                 GOTO(out, rc);
2248
2249         if (used_number != 0)
2250                 cfs_crypto_hash_update_page(req, __page, 0,
2251                         used_number * sizeof(*guard_start));
2252
2253         bufsize = sizeof(cksum);
2254         rc = cfs_crypto_hash_final(req, (unsigned char *)&cksum, &bufsize);
2255
2256         if (rc == 0)
2257                 *check_sum = cksum;
2258 out:
2259         __free_page(__page);
2260         return rc;
2261 }
2262
2263 static int tgt_checksum_niobuf_rw(struct lu_target *tgt,
2264                                   enum cksum_types cksum_type,
2265                                   struct niobuf_local *local_nb,
2266                                   int npages, int opc, u32 *check_sum,
2267                                   bool resend)
2268 {
2269         obd_dif_csum_fn *fn = NULL;
2270         int sector_size = 0;
2271         int rc;
2272
2273         ENTRY;
2274         obd_t10_cksum2dif(cksum_type, &fn, &sector_size);
2275
2276         if (fn)
2277                 rc = tgt_checksum_niobuf_t10pi(tgt, local_nb, npages,
2278                                                opc, fn, sector_size,
2279                                                check_sum, resend);
2280         else
2281                 rc = tgt_checksum_niobuf(tgt, local_nb, npages, opc,
2282                                          cksum_type, check_sum);
2283
2284         RETURN(rc);
2285 }
2286
2287 int tgt_brw_read(struct tgt_session_info *tsi)
2288 {
2289         struct ptlrpc_request   *req = tgt_ses_req(tsi);
2290         struct ptlrpc_bulk_desc *desc = NULL;
2291         struct obd_export       *exp = tsi->tsi_exp;
2292         struct niobuf_remote    *remote_nb;
2293         struct niobuf_local     *local_nb;
2294         struct obd_ioobj        *ioo;
2295         struct ost_body         *body, *repbody;
2296         struct lustre_handle     lockh = { 0 };
2297         int                      npages, nob = 0, rc, i, no_reply = 0,
2298                                  npages_read;
2299         struct tgt_thread_big_cache *tbc = req->rq_svc_thread->t_data;
2300         const char *obd_name = exp->exp_obd->obd_name;
2301         ktime_t kstart;
2302
2303         ENTRY;
2304
2305         if (ptlrpc_req2svc(req)->srv_req_portal != OST_IO_PORTAL &&
2306             ptlrpc_req2svc(req)->srv_req_portal != MDS_IO_PORTAL) {
2307                 CERROR("%s: deny read request from %s to portal %u\n",
2308                        tgt_name(tsi->tsi_tgt),
2309                        obd_export_nid2str(req->rq_export),
2310                        ptlrpc_req2svc(req)->srv_req_portal);
2311                 RETURN(-EPROTO);
2312         }
2313
2314         req->rq_bulk_read = 1;
2315
2316         if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_READ_BULK)) {
2317                 /* optionally use cfs_fail_val - 1 to select a specific OST on
2318                  * this server to fail requests.
2319                  */
2320                 char fail_ost_name[MAX_OBD_NAME];
2321
2322                 if (cfs_fail_val > 0) {
2323                         snprintf(fail_ost_name, MAX_OBD_NAME, "OST%04X",
2324                                  cfs_fail_val - 1);
2325
2326                         if (strstr(obd_name, fail_ost_name))
2327                                 RETURN(err_serious(-EIO));
2328                 } else {
2329                         RETURN(err_serious(-EIO));
2330                 }
2331         }
2332
2333         OBD_FAIL_TIMEOUT(OBD_FAIL_OST_BRW_PAUSE_BULK, cfs_fail_val > 0 ?
2334                          cfs_fail_val : (obd_timeout + 1) / 4);
2335
2336         /* Check if there is eviction in progress, and if so, wait for it to
2337          * finish */
2338         if (unlikely(atomic_read(&exp->exp_obd->obd_evict_inprogress))) {
2339                 /* We do not care how long it takes */
2340                 wait_event_idle(
2341                         exp->exp_obd->obd_evict_inprogress_waitq,
2342                         !atomic_read(&exp->exp_obd->obd_evict_inprogress));
2343         }
2344
2345         /* There must be big cache in current thread to process this request
2346          * if it is NULL then something went wrong and it wasn't allocated,
2347          * report -ENOMEM in that case */
2348         if (tbc == NULL)
2349                 RETURN(-ENOMEM);
2350
2351         body = tsi->tsi_ost_body;
2352         LASSERT(body != NULL);
2353
2354         if (body->oa.o_valid & OBD_MD_FLFLAGS &&
2355             body->oa.o_flags & OBD_FL_NORPC)
2356                 RETURN(0);
2357
2358         ioo = req_capsule_client_get(tsi->tsi_pill, &RMF_OBD_IOOBJ);
2359         LASSERT(ioo != NULL); /* must exists after tgt_ost_body_unpack */
2360
2361         remote_nb = req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE);
2362         LASSERT(remote_nb != NULL); /* must exists after tgt_ost_body_unpack */
2363
2364         local_nb = tbc->local;
2365
2366         rc = tgt_brw_lock(tsi->tsi_env, exp, &tsi->tsi_resid, ioo, remote_nb,
2367                           &lockh, LCK_PR);
2368         if (rc != 0)
2369                 RETURN(rc);
2370
2371         /*
2372          * If getting the lock took more time than
2373          * client was willing to wait, drop it. b=11330
2374          */
2375         if (ktime_get_real_seconds() > req->rq_deadline ||
2376             OBD_FAIL_CHECK(OBD_FAIL_OST_DROP_REQ)) {
2377                 no_reply = 1;
2378                 CERROR("Dropping timed-out read from %s because locking object " DOSTID " took %lld seconds (limit was %lld).\n",
2379                        libcfs_idstr(&req->rq_peer), POSTID(&ioo->ioo_oid),
2380                        ktime_get_real_seconds() - req->rq_arrival_time.tv_sec,
2381                        req->rq_deadline - req->rq_arrival_time.tv_sec);
2382                 GOTO(out_lock, rc = -ETIMEDOUT);
2383         }
2384
2385         /*
2386          * Because we already sync grant info with client when
2387          * reconnect, grant info will be cleared for resent req,
2388          * otherwise, outdated grant count in the rpc would de-sync
2389          * grant counters in case of shrink
2390          */
2391         if (lustre_msg_get_flags(req->rq_reqmsg) & (MSG_RESENT | MSG_REPLAY)) {
2392                 DEBUG_REQ(D_CACHE, req, "clear resent/replay req grant info");
2393                 body->oa.o_valid &= ~OBD_MD_FLGRANT;
2394         }
2395
2396         repbody = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
2397         repbody->oa = body->oa;
2398
2399         npages = PTLRPC_MAX_BRW_PAGES;
2400         kstart = ktime_get();
2401         rc = obd_preprw(tsi->tsi_env, OBD_BRW_READ, exp, &repbody->oa, 1,
2402                         ioo, remote_nb, &npages, local_nb);
2403         if (rc != 0)
2404                 GOTO(out_lock, rc);
2405
2406         if (body->oa.o_valid & OBD_MD_FLFLAGS &&
2407             body->oa.o_flags & OBD_FL_SHORT_IO) {
2408                 desc = NULL;
2409         } else {
2410                 desc = ptlrpc_prep_bulk_exp(req, npages, ioobj_max_brw_get(ioo),
2411                                             PTLRPC_BULK_PUT_SOURCE,
2412                                             OST_BULK_PORTAL,
2413                                             &ptlrpc_bulk_kiov_nopin_ops);
2414                 if (desc == NULL)
2415                         GOTO(out_commitrw, rc = -ENOMEM);
2416         }
2417
2418         npages_read = npages;
2419         for (i = 0; i < npages; i++) {
2420                 int page_rc = local_nb[i].lnb_rc;
2421
2422                 if (page_rc < 0) {
2423                         rc = page_rc;
2424                         npages_read = i;
2425                         break;
2426                 }
2427
2428                 nob += page_rc;
2429                 if (page_rc != 0 && desc != NULL) { /* some data! */
2430                         LASSERT(local_nb[i].lnb_page != NULL);
2431                         desc->bd_frag_ops->add_kiov_frag
2432                           (desc, local_nb[i].lnb_page,
2433                            local_nb[i].lnb_page_offset & ~PAGE_MASK,
2434                            page_rc);
2435                 }
2436
2437                 if (page_rc != local_nb[i].lnb_len) { /* short read */
2438                         local_nb[i].lnb_len = page_rc;
2439                         npages_read = i + (page_rc != 0 ? 1 : 0);
2440                         /* All subsequent pages should be 0 */
2441                         while (++i < npages)
2442                                 LASSERT(local_nb[i].lnb_rc == 0);
2443                         break;
2444                 }
2445         }
2446
2447         if (body->oa.o_valid & OBD_MD_FLCKSUM) {
2448                 u32 flag = body->oa.o_valid & OBD_MD_FLFLAGS ?
2449                            body->oa.o_flags : 0;
2450                 enum cksum_types cksum_type = obd_cksum_type_unpack(flag);
2451                 bool resend = (body->oa.o_valid & OBD_MD_FLFLAGS) &&
2452                         (body->oa.o_flags & OBD_FL_RECOV_RESEND);
2453
2454                 repbody->oa.o_flags = obd_cksum_type_pack(obd_name,
2455                                                           cksum_type);
2456                 repbody->oa.o_valid = OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
2457
2458                 rc = tgt_checksum_niobuf_rw(tsi->tsi_tgt, cksum_type,
2459                                             local_nb, npages_read, OST_READ,
2460                                             &repbody->oa.o_cksum, resend);
2461                 if (rc < 0)
2462                         GOTO(out_commitrw, rc);
2463                 CDEBUG(D_PAGE | (resend ? D_HA : 0),
2464                        "checksum at read origin: %x (%x)\n",
2465                        repbody->oa.o_cksum, cksum_type);
2466
2467                 /* if a resend it could be for a cksum error, so check Server
2468                  * cksum with returned Client cksum (this should even cover
2469                  * zero-cksum case) */
2470                 if (resend)
2471                         check_read_checksum(local_nb, npages_read, exp,
2472                                             &body->oa, &req->rq_peer,
2473                                             body->oa.o_cksum,
2474                                             repbody->oa.o_cksum, cksum_type);
2475         } else {
2476                 repbody->oa.o_valid = 0;
2477         }
2478         if (body->oa.o_valid & OBD_MD_FLGRANT)
2479                 repbody->oa.o_valid |= OBD_MD_FLGRANT;
2480         /* We're finishing using body->oa as an input variable */
2481
2482         /* Check if client was evicted while we were doing i/o before touching
2483          * network */
2484         if (rc == 0) {
2485                 if (body->oa.o_valid & OBD_MD_FLFLAGS &&
2486                     body->oa.o_flags & OBD_FL_SHORT_IO) {
2487                         unsigned char *short_io_buf;
2488                         int short_io_size;
2489
2490                         short_io_buf = req_capsule_server_get(&req->rq_pill,
2491                                                               &RMF_SHORT_IO);
2492                         short_io_size = req_capsule_get_size(&req->rq_pill,
2493                                                              &RMF_SHORT_IO,
2494                                                              RCL_SERVER);
2495                         rc = tgt_pages2shortio(local_nb, npages_read,
2496                                                short_io_buf, short_io_size);
2497                         if (rc >= 0)
2498                                 req_capsule_shrink(&req->rq_pill,
2499                                                    &RMF_SHORT_IO, rc,
2500                                                    RCL_SERVER);
2501                         rc = rc > 0 ? 0 : rc;
2502                 } else if (!CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2)) {
2503                         rc = target_bulk_io(exp, desc);
2504                 }
2505                 no_reply = rc != 0;
2506         } else {
2507                 if (body->oa.o_valid & OBD_MD_FLFLAGS &&
2508                     body->oa.o_flags & OBD_FL_SHORT_IO)
2509                         req_capsule_shrink(&req->rq_pill, &RMF_SHORT_IO, 0,
2510                                            RCL_SERVER);
2511         }
2512
2513 out_commitrw:
2514         /* Must commit after prep above in all cases */
2515         rc = obd_commitrw(tsi->tsi_env, OBD_BRW_READ, exp, &repbody->oa, 1, ioo,
2516                           remote_nb, npages, local_nb, rc, nob, kstart);
2517 out_lock:
2518         tgt_brw_unlock(exp, ioo, remote_nb, &lockh, LCK_PR);
2519
2520         if (desc && !CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2))
2521                 ptlrpc_free_bulk(desc);
2522
2523         LASSERT(rc <= 0);
2524         if (rc == 0) {
2525                 rc = nob;
2526                 ptlrpc_lprocfs_brw(req, nob);
2527         } else if (no_reply) {
2528                 req->rq_no_reply = 1;
2529                 /* reply out callback would free */
2530                 ptlrpc_req_drop_rs(req);
2531                 LCONSOLE_WARN("%s: Bulk IO read error with %s (at %s), "
2532                               "client will retry: rc %d\n",
2533                               obd_name,
2534                               obd_uuid2str(&exp->exp_client_uuid),
2535                               obd_export_nid2str(exp), rc);
2536         }
2537         /* send a bulk after reply to simulate a network delay or reordering
2538          * by a router - Note that !desc implies short io, so there is no bulk
2539          * to reorder. */
2540         if (unlikely(CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2)) &&
2541             desc) {
2542                 /* Calculate checksum before request transfer, original
2543                  * it is done by target_bulk_io() */
2544                 rc = sptlrpc_svc_wrap_bulk(req, desc);
2545                 if (OCD_HAS_FLAG(&exp->exp_connect_data, BULK_MBITS))
2546                         req->rq_mbits = lustre_msg_get_mbits(req->rq_reqmsg);
2547                 else /* old version, bulk matchbits is rq_xid */
2548                         req->rq_mbits = req->rq_xid;
2549
2550                 req->rq_status = rc;
2551                 target_committed_to_req(req);
2552                 target_send_reply(req, 0, 0);
2553
2554                 CDEBUG(D_INFO, "reorder BULK\n");
2555                 OBD_FAIL_TIMEOUT(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2,
2556                                  cfs_fail_val ? : 3);
2557
2558                 target_bulk_io(exp, desc);
2559                 ptlrpc_free_bulk(desc);
2560         }
2561
2562         RETURN(rc);
2563 }
2564 EXPORT_SYMBOL(tgt_brw_read);
2565
2566 static int tgt_shortio2pages(struct niobuf_local *local, int npages,
2567                              unsigned char *buf, unsigned int size)
2568 {
2569         int     i, off, len;
2570         char    *ptr;
2571
2572         for (i = 0; i < npages; i++) {
2573                 off = local[i].lnb_page_offset & ~PAGE_MASK;
2574                 len = local[i].lnb_len;
2575
2576                 if (len == 0)
2577                         continue;
2578
2579                 CDEBUG(D_PAGE, "index %d offset = %d len = %d left = %d\n",
2580                        i, off, len, size);
2581                 ptr = kmap_atomic(local[i].lnb_page);
2582                 if (ptr == NULL)
2583                         return -EINVAL;
2584                 memcpy(ptr + off, buf, len < size ? len : size);
2585                 kunmap_atomic(ptr);
2586                 buf += len;
2587                 size -= len;
2588         }
2589         return 0;
2590 }
2591
2592 static void tgt_warn_on_cksum(struct ptlrpc_request *req,
2593                               struct ptlrpc_bulk_desc *desc,
2594                               struct niobuf_local *local_nb, int npages,
2595                               u32 client_cksum, u32 server_cksum,
2596                               bool mmap)
2597 {
2598         struct obd_export *exp = req->rq_export;
2599         struct ost_body *body;
2600         char *router = "";
2601         char *via = "";
2602
2603         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
2604         LASSERT(body != NULL);
2605
2606         if (desc && !nid_same(&req->rq_peer.nid, &desc->bd_sender)) {
2607                 via = " via ";
2608                 router = libcfs_nidstr(&desc->bd_sender);
2609         }
2610
2611         if (exp->exp_obd->obd_checksum_dump)
2612                 dump_all_bulk_pages(&body->oa, npages, local_nb, server_cksum,
2613                                     client_cksum);
2614
2615         if (mmap) {
2616                 CDEBUG_LIMIT(D_INFO, "client csum %x, server csum %x\n",
2617                              client_cksum, server_cksum);
2618                 return;
2619         }
2620
2621         LCONSOLE_ERROR_MSG(0x168, "%s: BAD WRITE CHECKSUM: from %s%s%s inode "
2622                            DFID" object "DOSTID" extent [%llu-%llu"
2623                            "]: client csum %x, server csum %x\n",
2624                            exp->exp_obd->obd_name, libcfs_idstr(&req->rq_peer),
2625                            via, router,
2626                            body->oa.o_valid & OBD_MD_FLFID ?
2627                            body->oa.o_parent_seq : (__u64)0,
2628                            body->oa.o_valid & OBD_MD_FLFID ?
2629                            body->oa.o_parent_oid : 0,
2630                            body->oa.o_valid & OBD_MD_FLFID ?
2631                            body->oa.o_parent_ver : 0,
2632                            POSTID(&body->oa.o_oi),
2633                            local_nb[0].lnb_file_offset,
2634                            local_nb[npages-1].lnb_file_offset +
2635                            local_nb[npages - 1].lnb_len - 1,
2636                            client_cksum, server_cksum);
2637 }
2638
2639 int tgt_brw_write(struct tgt_session_info *tsi)
2640 {
2641         struct tgt_thread_info *tti = tgt_th_info(tsi->tsi_env);
2642         struct ptlrpc_request   *req = tgt_ses_req(tsi);
2643         struct ptlrpc_bulk_desc *desc = NULL;
2644         struct obd_export       *exp = req->rq_export;
2645         struct niobuf_remote    *remote_nb;
2646         struct niobuf_local     *local_nb;
2647         struct obd_ioobj        *ioo;
2648         struct ost_body         *body, *repbody;
2649         struct lustre_handle     lockh = {0};
2650         __u32                   *rcs;
2651         int                      objcount, niocount, npages;
2652         int                      rc = 0;
2653         int                      i, j;
2654         enum cksum_types cksum_type = OBD_CKSUM_CRC32;
2655         bool                     no_reply = false, mmap;
2656         struct tgt_thread_big_cache *tbc = req->rq_svc_thread->t_data;
2657         bool wait_sync = false;
2658         const char *obd_name = exp->exp_obd->obd_name;
2659         /* '1' for consistency with code that checks !mpflag to restore */
2660         unsigned int mpflags = 1;
2661         ktime_t kstart;
2662         int nob = 0;
2663
2664         ENTRY;
2665
2666         if (ptlrpc_req2svc(req)->srv_req_portal != OST_IO_PORTAL &&
2667             ptlrpc_req2svc(req)->srv_req_portal != MDS_IO_PORTAL) {
2668                 CERROR("%s: deny write request from %s to portal %u\n",
2669                        tgt_name(tsi->tsi_tgt),
2670                        obd_export_nid2str(req->rq_export),
2671                        ptlrpc_req2svc(req)->srv_req_portal);
2672                 RETURN(err_serious(-EPROTO));
2673         }
2674
2675         if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOSPC))
2676                 RETURN(err_serious(-ENOSPC));
2677         if (OBD_FAIL_TIMEOUT(OBD_FAIL_OST_EROFS, 1))
2678                 RETURN(err_serious(-EROFS));
2679
2680         req->rq_bulk_write = 1;
2681
2682         if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_WRITE_BULK))
2683                 rc = -EIO;
2684         if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_WRITE_BULK2))
2685                 rc = -EFAULT;
2686         if (rc < 0) {
2687                 /* optionally use cfs_fail_val - 1 to select a specific OST on
2688                  * this server to fail requests.
2689                  */
2690                 char fail_ost_name[MAX_OBD_NAME];
2691
2692                 if (cfs_fail_val > 0) {
2693                         snprintf(fail_ost_name, MAX_OBD_NAME, "OST%04X",
2694                                  cfs_fail_val - 1);
2695
2696                         if (strstr(obd_name, fail_ost_name))
2697                                 RETURN(err_serious(rc));
2698                 } else {
2699                         RETURN(err_serious(rc));
2700                 }
2701         }
2702
2703         /* pause before transaction has been started */
2704         CFS_FAIL_TIMEOUT(OBD_FAIL_OST_BRW_PAUSE_BULK, cfs_fail_val > 0 ?
2705                          cfs_fail_val : (obd_timeout + 1) / 4);
2706
2707         /* Delay write commit to show stale size information */
2708         CFS_FAIL_TIMEOUT(OBD_FAIL_OSC_NO_SIZE_DATA, cfs_fail_val);
2709
2710         /* There must be big cache in current thread to process this request
2711          * if it is NULL then something went wrong and it wasn't allocated,
2712          * report -ENOMEM in that case */
2713         if (tbc == NULL)
2714                 RETURN(-ENOMEM);
2715
2716         body = tsi->tsi_ost_body;
2717         LASSERT(body != NULL);
2718
2719         if (body->oa.o_valid & OBD_MD_FLFLAGS &&
2720             body->oa.o_flags & OBD_FL_NORPC)
2721                 RETURN(0);
2722
2723
2724         ioo = req_capsule_client_get(&req->rq_pill, &RMF_OBD_IOOBJ);
2725         LASSERT(ioo != NULL); /* must exists after tgt_ost_body_unpack */
2726
2727         objcount = req_capsule_get_size(&req->rq_pill, &RMF_OBD_IOOBJ,
2728                                         RCL_CLIENT) / sizeof(*ioo);
2729
2730         for (niocount = i = 0; i < objcount; i++)
2731                 niocount += ioo[i].ioo_bufcnt;
2732
2733         remote_nb = req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE);
2734         LASSERT(remote_nb != NULL); /* must exists after tgt_ost_body_unpack */
2735         if (niocount != req_capsule_get_size(&req->rq_pill,
2736                                              &RMF_NIOBUF_REMOTE, RCL_CLIENT) /
2737                         sizeof(*remote_nb))
2738                 RETURN(err_serious(-EPROTO));
2739
2740         if ((remote_nb[0].rnb_flags & OBD_BRW_MEMALLOC) &&
2741             ptlrpc_connection_is_local(exp->exp_connection))
2742                 mpflags = memalloc_noreclaim_save();
2743
2744         req_capsule_set_size(&req->rq_pill, &RMF_RCS, RCL_SERVER,
2745                              niocount * sizeof(*rcs));
2746         rc = req_capsule_server_pack(&req->rq_pill);
2747         if (rc != 0)
2748                 GOTO(out, rc = err_serious(rc));
2749
2750         CFS_FAIL_TIMEOUT(OBD_FAIL_OST_BRW_PAUSE_PACK, cfs_fail_val);
2751         rcs = req_capsule_server_get(&req->rq_pill, &RMF_RCS);
2752
2753         local_nb = tbc->local;
2754
2755         rc = tgt_brw_lock(tsi->tsi_env, exp, &tsi->tsi_resid, ioo, remote_nb,
2756                           &lockh, LCK_PW);
2757         if (rc != 0)
2758                 GOTO(out, rc);
2759
2760         /*
2761          * If getting the lock took more time than
2762          * client was willing to wait, drop it. b=11330
2763          */
2764         if (ktime_get_real_seconds() > req->rq_deadline ||
2765             OBD_FAIL_CHECK(OBD_FAIL_OST_DROP_REQ)) {
2766                 no_reply = true;
2767                 CERROR("%s: Dropping timed-out write from %s because locking object " DOSTID " took %lld seconds (limit was %lld).\n",
2768                        tgt_name(tsi->tsi_tgt), libcfs_idstr(&req->rq_peer),
2769                        POSTID(&ioo->ioo_oid),
2770                        ktime_get_real_seconds() - req->rq_arrival_time.tv_sec,
2771                        req->rq_deadline - req->rq_arrival_time.tv_sec);
2772                 GOTO(out_lock, rc = -ETIMEDOUT);
2773         }
2774
2775         /* Because we already sync grant info with client when reconnect,
2776          * grant info will be cleared for resent req, then fed_grant and
2777          * total_grant will not be modified in following preprw_write */
2778         if (lustre_msg_get_flags(req->rq_reqmsg) & (MSG_RESENT | MSG_REPLAY)) {
2779                 DEBUG_REQ(D_CACHE, req, "clear resent/replay req grant info");
2780                 body->oa.o_valid &= ~OBD_MD_FLGRANT;
2781         }
2782
2783         repbody = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
2784         if (repbody == NULL)
2785                 GOTO(out_lock, rc = -ENOMEM);
2786         repbody->oa = body->oa;
2787
2788         npages = PTLRPC_MAX_BRW_PAGES;
2789         kstart = ktime_get();
2790         rc = obd_preprw(tsi->tsi_env, OBD_BRW_WRITE, exp, &repbody->oa,
2791                         objcount, ioo, remote_nb, &npages, local_nb);
2792         if (rc < 0)
2793                 GOTO(out_lock, rc);
2794         if (body->oa.o_valid & OBD_MD_FLFLAGS &&
2795             body->oa.o_flags & OBD_FL_SHORT_IO) {
2796                 unsigned int short_io_size;
2797                 unsigned char *short_io_buf;
2798
2799                 short_io_size = req_capsule_get_size(&req->rq_pill,
2800                                                      &RMF_SHORT_IO,
2801                                                      RCL_CLIENT);
2802                 short_io_buf = req_capsule_client_get(&req->rq_pill,
2803                                                       &RMF_SHORT_IO);
2804                 CDEBUG(D_INFO, "Client use short io for data transfer,"
2805                                " size = %d\n", short_io_size);
2806
2807                 /* Copy short io buf to pages */
2808                 rc = tgt_shortio2pages(local_nb, npages, short_io_buf,
2809                                        short_io_size);
2810                 desc = NULL;
2811         } else {
2812                 desc = ptlrpc_prep_bulk_exp(req, npages, ioobj_max_brw_get(ioo),
2813                                             PTLRPC_BULK_GET_SINK,
2814                                             OST_BULK_PORTAL,
2815                                             &ptlrpc_bulk_kiov_nopin_ops);
2816                 if (desc == NULL)
2817                         GOTO(skip_transfer, rc = -ENOMEM);
2818
2819                 /* NB Having prepped, we must commit... */
2820                 for (i = 0; i < npages; i++)
2821                         desc->bd_frag_ops->add_kiov_frag(desc,
2822                                         local_nb[i].lnb_page,
2823                                         local_nb[i].lnb_page_offset & ~PAGE_MASK,
2824                                         local_nb[i].lnb_len);
2825
2826                 rc = sptlrpc_svc_prep_bulk(req, desc);
2827                 if (rc != 0)
2828                         GOTO(skip_transfer, rc);
2829
2830                 rc = target_bulk_io(exp, desc);
2831         }
2832
2833         no_reply = rc != 0;
2834
2835 skip_transfer:
2836         if (body->oa.o_valid & OBD_MD_FLCKSUM && rc == 0) {
2837                 static int cksum_counter;
2838
2839                 if (body->oa.o_valid & OBD_MD_FLFLAGS)
2840                         cksum_type = obd_cksum_type_unpack(body->oa.o_flags);
2841
2842                 repbody->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
2843                 repbody->oa.o_flags &= ~OBD_FL_CKSUM_ALL;
2844                 repbody->oa.o_flags |= obd_cksum_type_pack(obd_name,
2845                                                            cksum_type);
2846
2847                 rc = tgt_checksum_niobuf_rw(tsi->tsi_tgt, cksum_type,
2848                                             local_nb, npages, OST_WRITE,
2849                                             &repbody->oa.o_cksum, false);
2850                 if (rc < 0)
2851                         GOTO(out_commitrw, rc);
2852
2853                 cksum_counter++;
2854
2855                 if (unlikely(body->oa.o_cksum != repbody->oa.o_cksum)) {
2856                         mmap = (body->oa.o_valid & OBD_MD_FLFLAGS &&
2857                                 body->oa.o_flags & OBD_FL_MMAP);
2858
2859                         tgt_warn_on_cksum(req, desc, local_nb, npages,
2860                                           body->oa.o_cksum,
2861                                           repbody->oa.o_cksum, mmap);
2862                         cksum_counter = 0;
2863                 } else if ((cksum_counter & (-cksum_counter)) ==
2864                            cksum_counter) {
2865                         CDEBUG(D_INFO, "Checksum %u from %s OK: %x\n",
2866                                cksum_counter, libcfs_idstr(&req->rq_peer),
2867                                repbody->oa.o_cksum);
2868                 }
2869         }
2870
2871         OBD_FAIL_TIMEOUT(OBD_FAIL_OST_BRW_PAUSE_BULK2, cfs_fail_val);
2872
2873 out_commitrw:
2874         /* calculate the expected actual write bytes (nob) for OFD stats.
2875          * Technically, if commit fails this would be wrong, but that should be
2876          * very rare
2877          */
2878         for (i = 0; i < niocount; i++) {
2879                 int len = remote_nb[i].rnb_len;
2880
2881                 nob += len;
2882         }
2883
2884         /* multiple transactions can be assigned during write commit */
2885         tti->tti_mult_trans = 1;
2886
2887         /* Must commit after prep above in all cases */
2888         rc = obd_commitrw(tsi->tsi_env, OBD_BRW_WRITE, exp, &repbody->oa,
2889                           objcount, ioo, remote_nb, npages, local_nb, rc, nob,
2890                           kstart);
2891         if (rc == -ENOTCONN)
2892                 /* quota acquire process has been given up because
2893                  * either the client has been evicted or the client
2894                  * has timed out the request already
2895                  */
2896                 no_reply = true;
2897
2898         for (i = 0; i < niocount; i++) {
2899                 if (!(local_nb[i].lnb_flags & OBD_BRW_ASYNC)) {
2900                         wait_sync = true;
2901                         break;
2902                 }
2903         }
2904         /*
2905          * Disable sending mtime back to the client. If the client locked the
2906          * whole object, then it has already updated the mtime on its side,
2907          * otherwise it will have to glimpse anyway (see bug 21489, comment 32)
2908          */
2909         repbody->oa.o_valid &= ~(OBD_MD_FLMTIME | OBD_MD_FLATIME);
2910
2911         if (rc == 0) {
2912                 /* set per-requested niobuf return codes */
2913                 for (i = j = 0; i < niocount; i++) {
2914                         int len = remote_nb[i].rnb_len;
2915
2916                         rcs[i] = 0;
2917                         do {
2918                                 LASSERT(j < npages);
2919                                 if (local_nb[j].lnb_rc < 0)
2920                                         rcs[i] = local_nb[j].lnb_rc;
2921                                 len -= local_nb[j].lnb_len;
2922                                 j++;
2923                         } while (len > 0);
2924                         LASSERT(len == 0);
2925                 }
2926                 LASSERT(j == npages);
2927                 ptlrpc_lprocfs_brw(req, nob);
2928         }
2929 out_lock:
2930         tgt_brw_unlock(exp, ioo, remote_nb, &lockh, LCK_PW);
2931         if (desc)
2932                 ptlrpc_free_bulk(desc);
2933 out:
2934         if (unlikely(no_reply || (exp->exp_obd->obd_no_transno && wait_sync))) {
2935                 req->rq_no_reply = 1;
2936                 /* reply out callback would free */
2937                 ptlrpc_req_drop_rs(req);
2938                 if (!exp->exp_obd->obd_no_transno)
2939                         LCONSOLE_WARN("%s: Bulk IO write error with %s (at %s),"
2940                                       " client will retry: rc = %d\n",
2941                                       obd_name,
2942                                       obd_uuid2str(&exp->exp_client_uuid),
2943                                       obd_export_nid2str(exp), rc);
2944         }
2945
2946         if (mpflags)
2947                 memalloc_noreclaim_restore(mpflags);
2948
2949         RETURN(rc);
2950 }
2951 EXPORT_SYMBOL(tgt_brw_write);
2952
2953 /**
2954  * Common request handler for OST_SEEK RPC.
2955  *
2956  * Unified request handling for OST_SEEK RPC.
2957  * It takes object by its FID, does needed lseek and packs result
2958  * into reply. Only SEEK_HOLE and SEEK_DATA are supported.
2959  *
2960  * \param[in] tsi       target session environment for this request
2961  *
2962  * \retval              0 if successful
2963  * \retval              negative value on error
2964  */
2965 int tgt_lseek(struct tgt_session_info *tsi)
2966 {
2967         struct lustre_handle lh = { 0 };
2968         struct dt_object *dob;
2969         struct ost_body *repbody;
2970         loff_t offset = tsi->tsi_ost_body->oa.o_size;
2971         int whence = tsi->tsi_ost_body->oa.o_mode;
2972         bool srvlock;
2973         int rc = 0;
2974
2975         ENTRY;
2976
2977         if (whence != SEEK_HOLE && whence != SEEK_DATA)
2978                 RETURN(-EPROTO);
2979
2980         /* Negative offset is prohibited on wire and must be handled on client
2981          * prior sending RPC.
2982          */
2983         if (offset < 0)
2984                 RETURN(-EPROTO);
2985
2986         repbody = req_capsule_server_get(tsi->tsi_pill, &RMF_OST_BODY);
2987         if (repbody == NULL)
2988                 RETURN(-ENOMEM);
2989         repbody->oa = tsi->tsi_ost_body->oa;
2990
2991         srvlock = tsi->tsi_ost_body->oa.o_valid & OBD_MD_FLFLAGS &&
2992                   tsi->tsi_ost_body->oa.o_flags & OBD_FL_SRVLOCK;
2993         if (srvlock) {
2994                 rc = tgt_data_lock(tsi->tsi_env, tsi->tsi_exp, &tsi->tsi_resid,
2995                                    offset, OBD_OBJECT_EOF, &lh, LCK_PR);
2996                 if (rc)
2997                         RETURN(rc);
2998         }
2999
3000         dob = dt_locate(tsi->tsi_env, tsi->tsi_tgt->lut_bottom, &tsi->tsi_fid);
3001         if (IS_ERR(dob))
3002                 GOTO(out, rc = PTR_ERR(dob));
3003
3004         if (!dt_object_exists(dob))
3005                 GOTO(obj_put, rc = -ENOENT);
3006
3007         repbody->oa.o_size = dt_lseek(tsi->tsi_env, dob, offset, whence);
3008         rc = 0;
3009 obj_put:
3010         dt_object_put(tsi->tsi_env, dob);
3011 out:
3012         if (srvlock)
3013                 tgt_data_unlock(&lh, LCK_PR);
3014
3015         RETURN(rc);
3016 }
3017 EXPORT_SYMBOL(tgt_lseek);
3018
3019 /* Check if request can be reconstructed from saved reply data
3020  * A copy of the reply data is returned in @trd if the pointer is not NULL
3021  */
3022 int req_can_reconstruct(struct ptlrpc_request *req,
3023                          struct tg_reply_data *trd)
3024 {
3025         struct tg_export_data *ted = &req->rq_export->exp_target_data;
3026         struct lsd_client_data *lcd = ted->ted_lcd;
3027         int found;
3028
3029         if (tgt_is_multimodrpcs_client(req->rq_export))
3030                 return tgt_lookup_reply(req, trd);
3031
3032         mutex_lock(&ted->ted_lcd_lock);
3033         found = req->rq_xid == lcd->lcd_last_xid ||
3034                 req->rq_xid == lcd->lcd_last_close_xid;
3035
3036         if (found && trd != NULL) {
3037                 if (lustre_msg_get_opc(req->rq_reqmsg) == MDS_CLOSE) {
3038                         trd->trd_reply.lrd_xid = lcd->lcd_last_close_xid;
3039                         trd->trd_reply.lrd_transno =
3040                                                 lcd->lcd_last_close_transno;
3041                         trd->trd_reply.lrd_result = lcd->lcd_last_close_result;
3042                 } else {
3043                         trd->trd_reply.lrd_xid = lcd->lcd_last_xid;
3044                         trd->trd_reply.lrd_transno = lcd->lcd_last_transno;
3045                         trd->trd_reply.lrd_result = lcd->lcd_last_result;
3046                         trd->trd_reply.lrd_data = lcd->lcd_last_data;
3047                         trd->trd_pre_versions[0] = lcd->lcd_pre_versions[0];
3048                         trd->trd_pre_versions[1] = lcd->lcd_pre_versions[1];
3049                         trd->trd_pre_versions[2] = lcd->lcd_pre_versions[2];
3050                         trd->trd_pre_versions[3] = lcd->lcd_pre_versions[3];
3051                 }
3052         }
3053         mutex_unlock(&ted->ted_lcd_lock);
3054
3055         return found;
3056 }
3057 EXPORT_SYMBOL(req_can_reconstruct);
3058