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