lustre/osc/osc_request.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, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  */
  31
  32 #define DEBUG_SUBSYSTEM S_OSC
  33
  34 #include <linux/workqueue.h>
  35 #include <libcfs/libcfs.h>
  36 #include <linux/falloc.h>
  37 #include <lprocfs_status.h>
  38 #include <lustre_dlm.h>
  39 #include <lustre_fid.h>
  40 #include <lustre_ha.h>
  41 #include <uapi/linux/lustre/lustre_ioctl.h>
  42 #include <lustre_net.h>
  43 #include <lustre_obdo.h>
  44 #include <obd.h>
  45 #include <obd_cksum.h>
  46 #include <obd_class.h>
  47 #include <lustre_osc.h>
  48 #include <linux/falloc.h>
  49
  50 #include "osc_internal.h"
  51 #include <lnet/lnet_rdma.h>
  52
  53 atomic_t osc_pool_req_count;
  54 unsigned int osc_reqpool_maxreqcount;
  55 struct ptlrpc_request_pool *osc_rq_pool;
  56
  57 /* max memory used for request pool, unit is MB */
  58 static unsigned int osc_reqpool_mem_max = 5;
  59 module_param(osc_reqpool_mem_max, uint, 0444);
  60
  61 static int osc_idle_timeout = 20;
  62 module_param(osc_idle_timeout, uint, 0644);
  63
  64 #define osc_grant_args osc_brw_async_args
  65
  66 struct osc_setattr_args {
  67         struct obdo             *sa_oa;
  68         obd_enqueue_update_f     sa_upcall;
  69         void                    *sa_cookie;
  70 };
  71
  72 struct osc_fsync_args {
  73         struct osc_object       *fa_obj;
  74         struct obdo             *fa_oa;
  75         obd_enqueue_update_f    fa_upcall;
  76         void                    *fa_cookie;
  77 };
  78
  79 struct osc_ladvise_args {
  80         struct obdo             *la_oa;
  81         obd_enqueue_update_f     la_upcall;
  82         void                    *la_cookie;
  83 };
  84
  85 static void osc_release_ppga(struct brw_page **ppga, size_t count);
  86 static int brw_interpret(const struct lu_env *env, struct ptlrpc_request *req,
  87                          void *data, int rc);
  88
  89 void osc_pack_req_body(struct ptlrpc_request *req, struct obdo *oa)
  90 {
  91         struct ost_body *body;
  92
  93         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
  94         LASSERT(body);
  95
  96         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
  97 }
  98
  99 static int osc_getattr(const struct lu_env *env, struct obd_export *exp,
 100                        struct obdo *oa)
 101 {
 102         struct ptlrpc_request   *req;
 103         struct ost_body         *body;
 104         int                      rc;
 105
 106         ENTRY;
 107         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_GETATTR);
 108         if (req == NULL)
 109                 RETURN(-ENOMEM);
 110
 111         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GETATTR);
 112         if (rc) {
 113                 ptlrpc_request_free(req);
 114                 RETURN(rc);
 115         }
 116
 117         osc_pack_req_body(req, oa);
 118
 119         ptlrpc_request_set_replen(req);
 120
 121         rc = ptlrpc_queue_wait(req);
 122         if (rc)
 123                 GOTO(out, rc);
 124
 125         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 126         if (body == NULL)
 127                 GOTO(out, rc = -EPROTO);
 128
 129         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
 130         lustre_get_wire_obdo(&req->rq_import->imp_connect_data, oa, &body->oa);
 131
 132         oa->o_blksize = cli_brw_size(exp->exp_obd);
 133         oa->o_valid |= OBD_MD_FLBLKSZ;
 134
 135         EXIT;
 136 out:
 137         ptlrpc_req_finished(req);
 138
 139         return rc;
 140 }
 141
 142 static int osc_setattr(const struct lu_env *env, struct obd_export *exp,
 143                        struct obdo *oa)
 144 {
 145         struct ptlrpc_request   *req;
 146         struct ost_body         *body;
 147         int                      rc;
 148
 149         ENTRY;
 150         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
 151
 152         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SETATTR);
 153         if (req == NULL)
 154                 RETURN(-ENOMEM);
 155
 156         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SETATTR);
 157         if (rc) {
 158                 ptlrpc_request_free(req);
 159                 RETURN(rc);
 160         }
 161
 162         osc_pack_req_body(req, oa);
 163
 164         ptlrpc_request_set_replen(req);
 165
 166         rc = ptlrpc_queue_wait(req);
 167         if (rc)
 168                 GOTO(out, rc);
 169
 170         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 171         if (body == NULL)
 172                 GOTO(out, rc = -EPROTO);
 173
 174         lustre_get_wire_obdo(&req->rq_import->imp_connect_data, oa, &body->oa);
 175
 176         EXIT;
 177 out:
 178         ptlrpc_req_finished(req);
 179
 180         RETURN(rc);
 181 }
 182
 183 static int osc_setattr_interpret(const struct lu_env *env,
 184                                  struct ptlrpc_request *req, void *args, int rc)
 185 {
 186         struct osc_setattr_args *sa = args;
 187         struct ost_body *body;
 188
 189         ENTRY;
 190
 191         if (rc != 0)
 192                 GOTO(out, rc);
 193
 194         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 195         if (body == NULL)
 196                 GOTO(out, rc = -EPROTO);
 197
 198         lustre_get_wire_obdo(&req->rq_import->imp_connect_data, sa->sa_oa,
 199                              &body->oa);
 200 out:
 201         rc = sa->sa_upcall(sa->sa_cookie, rc);
 202         RETURN(rc);
 203 }
 204
 205 int osc_setattr_async(struct obd_export *exp, struct obdo *oa,
 206                       obd_enqueue_update_f upcall, void *cookie,
 207                       struct ptlrpc_request_set *rqset)
 208 {
 209         struct ptlrpc_request   *req;
 210         struct osc_setattr_args *sa;
 211         int                      rc;
 212
 213         ENTRY;
 214
 215         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SETATTR);
 216         if (req == NULL)
 217                 RETURN(-ENOMEM);
 218
 219         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SETATTR);
 220         if (rc) {
 221                 ptlrpc_request_free(req);
 222                 RETURN(rc);
 223         }
 224
 225         osc_pack_req_body(req, oa);
 226
 227         ptlrpc_request_set_replen(req);
 228
 229         /* do mds to ost setattr asynchronously */
 230         if (!rqset) {
 231                 /* Do not wait for response. */
 232                 ptlrpcd_add_req(req);
 233         } else {
 234                 req->rq_interpret_reply = osc_setattr_interpret;
 235
 236                 sa = ptlrpc_req_async_args(sa, req);
 237                 sa->sa_oa = oa;
 238                 sa->sa_upcall = upcall;
 239                 sa->sa_cookie = cookie;
 240
 241                 ptlrpc_set_add_req(rqset, req);
 242         }
 243
 244         RETURN(0);
 245 }
 246
 247 static int osc_ladvise_interpret(const struct lu_env *env,
 248                                  struct ptlrpc_request *req,
 249                                  void *arg, int rc)
 250 {
 251         struct osc_ladvise_args *la = arg;
 252         struct ost_body *body;
 253         ENTRY;
 254
 255         if (rc != 0)
 256                 GOTO(out, rc);
 257
 258         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 259         if (body == NULL)
 260                 GOTO(out, rc = -EPROTO);
 261
 262         *la->la_oa = body->oa;
 263 out:
 264         rc = la->la_upcall(la->la_cookie, rc);
 265         RETURN(rc);
 266 }
 267
 268 /**
 269  * If rqset is NULL, do not wait for response. Upcall and cookie could also
 270  * be NULL in this case
 271  */
 272 int osc_ladvise_base(struct obd_export *exp, struct obdo *oa,
 273                      struct ladvise_hdr *ladvise_hdr,
 274                      obd_enqueue_update_f upcall, void *cookie,
 275                      struct ptlrpc_request_set *rqset)
 276 {
 277         struct ptlrpc_request   *req;
 278         struct ost_body         *body;
 279         struct osc_ladvise_args *la;
 280         int                      rc;
 281         struct lu_ladvise       *req_ladvise;
 282         struct lu_ladvise       *ladvise = ladvise_hdr->lah_advise;
 283         int                      num_advise = ladvise_hdr->lah_count;
 284         struct ladvise_hdr      *req_ladvise_hdr;
 285         ENTRY;
 286
 287         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_LADVISE);
 288         if (req == NULL)
 289                 RETURN(-ENOMEM);
 290
 291         req_capsule_set_size(&req->rq_pill, &RMF_OST_LADVISE, RCL_CLIENT,
 292                              num_advise * sizeof(*ladvise));
 293         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_LADVISE);
 294         if (rc != 0) {
 295                 ptlrpc_request_free(req);
 296                 RETURN(rc);
 297         }
 298         req->rq_request_portal = OST_IO_PORTAL;
 299         ptlrpc_at_set_req_timeout(req);
 300
 301         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 302         LASSERT(body);
 303         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa,
 304                              oa);
 305
 306         req_ladvise_hdr = req_capsule_client_get(&req->rq_pill,
 307                                                  &RMF_OST_LADVISE_HDR);
 308         memcpy(req_ladvise_hdr, ladvise_hdr, sizeof(*ladvise_hdr));
 309
 310         req_ladvise = req_capsule_client_get(&req->rq_pill, &RMF_OST_LADVISE);
 311         memcpy(req_ladvise, ladvise, sizeof(*ladvise) * num_advise);
 312         ptlrpc_request_set_replen(req);
 313
 314         if (rqset == NULL) {
 315                 /* Do not wait for response. */
 316                 ptlrpcd_add_req(req);
 317                 RETURN(0);
 318         }
 319
 320         req->rq_interpret_reply = osc_ladvise_interpret;
 321         la = ptlrpc_req_async_args(la, req);
 322         la->la_oa = oa;
 323         la->la_upcall = upcall;
 324         la->la_cookie = cookie;
 325
 326         ptlrpc_set_add_req(rqset, req);
 327
 328         RETURN(0);
 329 }
 330
 331 static int osc_create(const struct lu_env *env, struct obd_export *exp,
 332                       struct obdo *oa)
 333 {
 334         struct ptlrpc_request *req;
 335         struct ost_body       *body;
 336         int                    rc;
 337         ENTRY;
 338
 339         LASSERT(oa != NULL);
 340         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
 341         LASSERT(fid_seq_is_echo(ostid_seq(&oa->o_oi)));
 342
 343         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_CREATE);
 344         if (req == NULL)
 345                 GOTO(out, rc = -ENOMEM);
 346
 347         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_CREATE);
 348         if (rc) {
 349                 ptlrpc_request_free(req);
 350                 GOTO(out, rc);
 351         }
 352
 353         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 354         LASSERT(body);
 355
 356         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
 357
 358         ptlrpc_request_set_replen(req);
 359
 360         rc = ptlrpc_queue_wait(req);
 361         if (rc)
 362                 GOTO(out_req, rc);
 363
 364         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 365         if (body == NULL)
 366                 GOTO(out_req, rc = -EPROTO);
 367
 368         CDEBUG(D_INFO, "oa flags %x\n", oa->o_flags);
 369         lustre_get_wire_obdo(&req->rq_import->imp_connect_data, oa, &body->oa);
 370
 371         oa->o_blksize = cli_brw_size(exp->exp_obd);
 372         oa->o_valid |= OBD_MD_FLBLKSZ;
 373
 374         CDEBUG(D_HA, "transno: %lld\n",
 375                lustre_msg_get_transno(req->rq_repmsg));
 376 out_req:
 377         ptlrpc_req_finished(req);
 378 out:
 379         RETURN(rc);
 380 }
 381
 382 int osc_punch_send(struct obd_export *exp, struct obdo *oa,
 383                    obd_enqueue_update_f upcall, void *cookie)
 384 {
 385         struct ptlrpc_request *req;
 386         struct osc_setattr_args *sa;
 387         struct obd_import *imp = class_exp2cliimp(exp);
 388         struct ost_body *body;
 389         int rc;
 390
 391         ENTRY;
 392
 393         req = ptlrpc_request_alloc(imp, &RQF_OST_PUNCH);
 394         if (req == NULL)
 395                 RETURN(-ENOMEM);
 396
 397         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_PUNCH);
 398         if (rc < 0) {
 399                 ptlrpc_request_free(req);
 400                 RETURN(rc);
 401         }
 402
 403         osc_set_io_portal(req);
 404
 405         ptlrpc_at_set_req_timeout(req);
 406
 407         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 408
 409         lustre_set_wire_obdo(&imp->imp_connect_data, &body->oa, oa);
 410
 411         ptlrpc_request_set_replen(req);
 412
 413         req->rq_interpret_reply = osc_setattr_interpret;
 414         sa = ptlrpc_req_async_args(sa, req);
 415         sa->sa_oa = oa;
 416         sa->sa_upcall = upcall;
 417         sa->sa_cookie = cookie;
 418
 419         ptlrpcd_add_req(req);
 420
 421         RETURN(0);
 422 }
 423 EXPORT_SYMBOL(osc_punch_send);
 424
 425 /**
 426  * osc_fallocate_base() - Handles fallocate request.
 427  *
 428  * @exp:        Export structure
 429  * @oa:         Attributes passed to OSS from client (obdo structure)
 430  * @upcall:     Primary & supplementary group information
 431  * @cookie:     Exclusive identifier
 432  * @rqset:      Request list.
 433  * @mode:       Operation done on given range.
 434  *
 435  * osc_fallocate_base() - Handles fallocate requests only. Only block
 436  * allocation or standard preallocate operation is supported currently.
 437  * Other mode flags is not supported yet. ftruncate(2) or truncate(2)
 438  * is supported via SETATTR request.
 439  *
 440  * Return: Non-zero on failure and O on success.
 441  */
 442 int osc_fallocate_base(struct obd_export *exp, struct obdo *oa,
 443                        obd_enqueue_update_f upcall, void *cookie, int mode)
 444 {
 445         struct ptlrpc_request *req;
 446         struct osc_setattr_args *sa;
 447         struct ost_body *body;
 448         struct obd_import *imp = class_exp2cliimp(exp);
 449         int rc;
 450         ENTRY;
 451
 452         oa->o_falloc_mode = mode;
 453         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
 454                                    &RQF_OST_FALLOCATE);
 455         if (req == NULL)
 456                 RETURN(-ENOMEM);
 457
 458         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_FALLOCATE);
 459         if (rc != 0) {
 460                 ptlrpc_request_free(req);
 461                 RETURN(rc);
 462         }
 463
 464         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 465         LASSERT(body);
 466
 467         lustre_set_wire_obdo(&imp->imp_connect_data, &body->oa, oa);
 468
 469         ptlrpc_request_set_replen(req);
 470
 471         req->rq_interpret_reply = osc_setattr_interpret;
 472         BUILD_BUG_ON(sizeof(*sa) > sizeof(req->rq_async_args));
 473         sa = ptlrpc_req_async_args(sa, req);
 474         sa->sa_oa = oa;
 475         sa->sa_upcall = upcall;
 476         sa->sa_cookie = cookie;
 477
 478         ptlrpcd_add_req(req);
 479
 480         RETURN(0);
 481 }
 482 EXPORT_SYMBOL(osc_fallocate_base);
 483
 484 static int osc_sync_interpret(const struct lu_env *env,
 485                               struct ptlrpc_request *req, void *args, int rc)
 486 {
 487         struct osc_fsync_args *fa = args;
 488         struct ost_body *body;
 489         struct cl_attr *attr = &osc_env_info(env)->oti_attr;
 490         unsigned long valid = 0;
 491         struct cl_object *obj;
 492         ENTRY;
 493
 494         if (rc != 0)
 495                 GOTO(out, rc);
 496
 497         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 498         if (body == NULL) {
 499                 CERROR("can't unpack ost_body\n");
 500                 GOTO(out, rc = -EPROTO);
 501         }
 502
 503         *fa->fa_oa = body->oa;
 504         obj = osc2cl(fa->fa_obj);
 505
 506         /* Update osc object's blocks attribute */
 507         cl_object_attr_lock(obj);
 508         if (body->oa.o_valid & OBD_MD_FLBLOCKS) {
 509                 attr->cat_blocks = body->oa.o_blocks;
 510                 valid |= CAT_BLOCKS;
 511         }
 512
 513         if (valid != 0)
 514                 cl_object_attr_update(env, obj, attr, valid);
 515         cl_object_attr_unlock(obj);
 516
 517 out:
 518         rc = fa->fa_upcall(fa->fa_cookie, rc);
 519         RETURN(rc);
 520 }
 521
 522 int osc_sync_base(struct osc_object *obj, struct obdo *oa,
 523                   obd_enqueue_update_f upcall, void *cookie,
 524                   struct ptlrpc_request_set *rqset)
 525 {
 526         struct obd_export     *exp = osc_export(obj);
 527         struct ptlrpc_request *req;
 528         struct ost_body       *body;
 529         struct osc_fsync_args *fa;
 530         int                    rc;
 531         ENTRY;
 532
 533         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SYNC);
 534         if (req == NULL)
 535                 RETURN(-ENOMEM);
 536
 537         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SYNC);
 538         if (rc) {
 539                 ptlrpc_request_free(req);
 540                 RETURN(rc);
 541         }
 542
 543         /* overload the size and blocks fields in the oa with start/end */
 544         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 545         LASSERT(body);
 546         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
 547
 548         ptlrpc_request_set_replen(req);
 549         req->rq_interpret_reply = osc_sync_interpret;
 550
 551         fa = ptlrpc_req_async_args(fa, req);
 552         fa->fa_obj = obj;
 553         fa->fa_oa = oa;
 554         fa->fa_upcall = upcall;
 555         fa->fa_cookie = cookie;
 556
 557         ptlrpc_set_add_req(rqset, req);
 558
 559         RETURN (0);
 560 }
 561
 562 /* Find and cancel locally locks matched by @mode in the resource found by
 563  * @objid. Found locks are added into @cancel list. Returns the amount of
 564  * locks added to @cancels list. */
 565 static int osc_resource_get_unused(struct obd_export *exp, struct obdo *oa,
 566                                    struct list_head *cancels,
 567                                    enum ldlm_mode mode, __u64 lock_flags)
 568 {
 569         struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
 570         struct ldlm_res_id res_id;
 571         struct ldlm_resource *res;
 572         int count;
 573         ENTRY;
 574
 575         /* Return, i.e. cancel nothing, only if ELC is supported (flag in
 576          * export) but disabled through procfs (flag in NS).
 577          *
 578          * This distinguishes from a case when ELC is not supported originally,
 579          * when we still want to cancel locks in advance and just cancel them
 580          * locally, without sending any RPC. */
 581         if (exp_connect_cancelset(exp) && !ns_connect_cancelset(ns))
 582                 RETURN(0);
 583
 584         ostid_build_res_name(&oa->o_oi, &res_id);
 585         res = ldlm_resource_get(ns, NULL, &res_id, 0, 0);
 586         if (IS_ERR(res))
 587                 RETURN(0);
 588
 589         LDLM_RESOURCE_ADDREF(res);
 590         count = ldlm_cancel_resource_local(res, cancels, NULL, mode,
 591                                            lock_flags, 0, NULL);
 592         LDLM_RESOURCE_DELREF(res);
 593         ldlm_resource_putref(res);
 594         RETURN(count);
 595 }
 596
 597 static int osc_destroy_interpret(const struct lu_env *env,
 598                                  struct ptlrpc_request *req, void *args, int rc)
 599 {
 600         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
 601
 602         atomic_dec(&cli->cl_destroy_in_flight);
 603         wake_up(&cli->cl_destroy_waitq);
 604
 605         return 0;
 606 }
 607
 608 static int osc_can_send_destroy(struct client_obd *cli)
 609 {
 610         if (atomic_inc_return(&cli->cl_destroy_in_flight) <=
 611             cli->cl_max_rpcs_in_flight) {
 612                 /* The destroy request can be sent */
 613                 return 1;
 614         }
 615         if (atomic_dec_return(&cli->cl_destroy_in_flight) <
 616             cli->cl_max_rpcs_in_flight) {
 617                 /*
 618                  * The counter has been modified between the two atomic
 619                  * operations.
 620                  */
 621                 wake_up(&cli->cl_destroy_waitq);
 622         }
 623         return 0;
 624 }
 625
 626 static int osc_destroy(const struct lu_env *env, struct obd_export *exp,
 627                        struct obdo *oa)
 628 {
 629         struct client_obd     *cli = &exp->exp_obd->u.cli;
 630         struct ptlrpc_request *req;
 631         struct ost_body       *body;
 632         LIST_HEAD(cancels);
 633         int rc, count;
 634         ENTRY;
 635
 636         if (!oa) {
 637                 CDEBUG(D_INFO, "oa NULL\n");
 638                 RETURN(-EINVAL);
 639         }
 640
 641         count = osc_resource_get_unused(exp, oa, &cancels, LCK_PW,
 642                                         LDLM_FL_DISCARD_DATA);
 643
 644         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_DESTROY);
 645         if (req == NULL) {
 646                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 647                 RETURN(-ENOMEM);
 648         }
 649
 650         rc = ldlm_prep_elc_req(exp, req, LUSTRE_OST_VERSION, OST_DESTROY,
 651                                0, &cancels, count);
 652         if (rc) {
 653                 ptlrpc_request_free(req);
 654                 RETURN(rc);
 655         }
 656
 657         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
 658         ptlrpc_at_set_req_timeout(req);
 659
 660         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 661         LASSERT(body);
 662         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
 663
 664         ptlrpc_request_set_replen(req);
 665
 666         req->rq_interpret_reply = osc_destroy_interpret;
 667         if (!osc_can_send_destroy(cli)) {
 668                 /*
 669                  * Wait until the number of on-going destroy RPCs drops
 670                  * under max_rpc_in_flight
 671                  */
 672                 rc = l_wait_event_abortable_exclusive(
 673                         cli->cl_destroy_waitq,
 674                         osc_can_send_destroy(cli));
 675                 if (rc) {
 676                         ptlrpc_req_finished(req);
 677                         RETURN(-EINTR);
 678                 }
 679         }
 680
 681         /* Do not wait for response */
 682         ptlrpcd_add_req(req);
 683         RETURN(0);
 684 }
 685
 686 static void osc_announce_cached(struct client_obd *cli, struct obdo *oa,
 687                                 long writing_bytes)
 688 {
 689         u64 bits = OBD_MD_FLBLOCKS | OBD_MD_FLGRANT;
 690
 691         LASSERT(!(oa->o_valid & bits));
 692
 693         oa->o_valid |= bits;
 694         spin_lock(&cli->cl_loi_list_lock);
 695         if (cli->cl_ocd_grant_param)
 696                 oa->o_dirty = cli->cl_dirty_grant;
 697         else
 698                 oa->o_dirty = cli->cl_dirty_pages << PAGE_SHIFT;
 699         if (unlikely(cli->cl_dirty_pages > cli->cl_dirty_max_pages)) {
 700                 CERROR("dirty %lu > dirty_max %lu\n",
 701                        cli->cl_dirty_pages,
 702                        cli->cl_dirty_max_pages);
 703                 oa->o_undirty = 0;
 704         } else if (unlikely(atomic_long_read(&obd_dirty_pages) >
 705                             (long)(obd_max_dirty_pages + 1))) {
 706                 /* The atomic_read() allowing the atomic_inc() are
 707                  * not covered by a lock thus they may safely race and trip
 708                  * this CERROR() unless we add in a small fudge factor (+1). */
 709                 CERROR("%s: dirty %ld > system dirty_max %ld\n",
 710                        cli_name(cli), atomic_long_read(&obd_dirty_pages),
 711                        obd_max_dirty_pages);
 712                 oa->o_undirty = 0;
 713         } else if (unlikely(cli->cl_dirty_max_pages - cli->cl_dirty_pages >
 714                             0x7fffffff)) {
 715                 CERROR("dirty %lu - dirty_max %lu too big???\n",
 716                        cli->cl_dirty_pages, cli->cl_dirty_max_pages);
 717                 oa->o_undirty = 0;
 718         } else {
 719                 unsigned long nrpages;
 720                 unsigned long undirty;
 721
 722                 nrpages = cli->cl_max_pages_per_rpc;
 723                 nrpages *= cli->cl_max_rpcs_in_flight + 1;
 724                 nrpages = max(nrpages, cli->cl_dirty_max_pages);
 725                 undirty = nrpages << PAGE_SHIFT;
 726                 if (cli->cl_ocd_grant_param) {
 727                         int nrextents;
 728
 729                         /* take extent tax into account when asking for more
 730                          * grant space */
 731                         nrextents = (nrpages + cli->cl_max_extent_pages - 1) /
 732                                      cli->cl_max_extent_pages;
 733                         undirty += nrextents * cli->cl_grant_extent_tax;
 734                 }
 735                 /* Do not ask for more than OBD_MAX_GRANT - a margin for server
 736                  * to add extent tax, etc.
 737                  */
 738                 oa->o_undirty = min(undirty, OBD_MAX_GRANT &
 739                                     ~(PTLRPC_MAX_BRW_SIZE * 4UL));
 740         }
 741         oa->o_grant = cli->cl_avail_grant + cli->cl_reserved_grant;
 742         /* o_dropped AKA o_misc is 32 bits, but cl_lost_grant is 64 bits */
 743         if (cli->cl_lost_grant > INT_MAX) {
 744                 CDEBUG(D_CACHE,
 745                       "%s: avoided o_dropped overflow: cl_lost_grant %lu\n",
 746                       cli_name(cli), cli->cl_lost_grant);
 747                 oa->o_dropped = INT_MAX;
 748         } else {
 749                 oa->o_dropped = cli->cl_lost_grant;
 750         }
 751         cli->cl_lost_grant -= oa->o_dropped;
 752         spin_unlock(&cli->cl_loi_list_lock);
 753         CDEBUG(D_CACHE, "%s: dirty: %llu undirty: %u dropped %u grant: %llu"
 754                " cl_lost_grant %lu\n", cli_name(cli), oa->o_dirty,
 755                oa->o_undirty, oa->o_dropped, oa->o_grant, cli->cl_lost_grant);
 756 }
 757
 758 void osc_update_next_shrink(struct client_obd *cli)
 759 {
 760         cli->cl_next_shrink_grant = ktime_get_seconds() +
 761                                     cli->cl_grant_shrink_interval;
 762
 763         CDEBUG(D_CACHE, "next time %lld to shrink grant\n",
 764                cli->cl_next_shrink_grant);
 765 }
 766
 767 static void __osc_update_grant(struct client_obd *cli, u64 grant)
 768 {
 769         spin_lock(&cli->cl_loi_list_lock);
 770         cli->cl_avail_grant += grant;
 771         spin_unlock(&cli->cl_loi_list_lock);
 772 }
 773
 774 static void osc_update_grant(struct client_obd *cli, struct ost_body *body)
 775 {
 776         if (body->oa.o_valid & OBD_MD_FLGRANT) {
 777                 CDEBUG(D_CACHE, "got %llu extra grant\n", body->oa.o_grant);
 778                 __osc_update_grant(cli, body->oa.o_grant);
 779         }
 780 }
 781
 782 /**
 783  * grant thread data for shrinking space.
 784  */
 785 struct grant_thread_data {
 786         struct list_head        gtd_clients;
 787         struct mutex            gtd_mutex;
 788         unsigned long           gtd_stopped:1;
 789 };
 790 static struct grant_thread_data client_gtd;
 791
 792 static int osc_shrink_grant_interpret(const struct lu_env *env,
 793                                       struct ptlrpc_request *req,
 794                                       void *args, int rc)
 795 {
 796         struct osc_grant_args *aa = args;
 797         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
 798         struct ost_body *body;
 799
 800         if (rc != 0) {
 801                 __osc_update_grant(cli, aa->aa_oa->o_grant);
 802                 GOTO(out, rc);
 803         }
 804
 805         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 806         LASSERT(body);
 807         osc_update_grant(cli, body);
 808 out:
 809         OBD_SLAB_FREE_PTR(aa->aa_oa, osc_obdo_kmem);
 810         aa->aa_oa = NULL;
 811
 812         return rc;
 813 }
 814
 815 static void osc_shrink_grant_local(struct client_obd *cli, struct obdo *oa)
 816 {
 817         spin_lock(&cli->cl_loi_list_lock);
 818         oa->o_grant = cli->cl_avail_grant / 4;
 819         cli->cl_avail_grant -= oa->o_grant;
 820         spin_unlock(&cli->cl_loi_list_lock);
 821         if (!(oa->o_valid & OBD_MD_FLFLAGS)) {
 822                 oa->o_valid |= OBD_MD_FLFLAGS;
 823                 oa->o_flags = 0;
 824         }
 825         oa->o_flags |= OBD_FL_SHRINK_GRANT;
 826         osc_update_next_shrink(cli);
 827 }
 828
 829 /* Shrink the current grant, either from some large amount to enough for a
 830  * full set of in-flight RPCs, or if we have already shrunk to that limit
 831  * then to enough for a single RPC.  This avoids keeping more grant than
 832  * needed, and avoids shrinking the grant piecemeal. */
 833 static int osc_shrink_grant(struct client_obd *cli)
 834 {
 835         __u64 target_bytes = (cli->cl_max_rpcs_in_flight + 1) *
 836                              (cli->cl_max_pages_per_rpc << PAGE_SHIFT);
 837
 838         spin_lock(&cli->cl_loi_list_lock);
 839         if (cli->cl_avail_grant <= target_bytes)
 840                 target_bytes = cli->cl_max_pages_per_rpc << PAGE_SHIFT;
 841         spin_unlock(&cli->cl_loi_list_lock);
 842
 843         return osc_shrink_grant_to_target(cli, target_bytes);
 844 }
 845
 846 int osc_shrink_grant_to_target(struct client_obd *cli, __u64 target_bytes)
 847 {
 848         int                     rc = 0;
 849         struct ost_body        *body;
 850         ENTRY;
 851
 852         spin_lock(&cli->cl_loi_list_lock);
 853         /* Don't shrink if we are already above or below the desired limit
 854          * We don't want to shrink below a single RPC, as that will negatively
 855          * impact block allocation and long-term performance. */
 856         if (target_bytes < cli->cl_max_pages_per_rpc << PAGE_SHIFT)
 857                 target_bytes = cli->cl_max_pages_per_rpc << PAGE_SHIFT;
 858
 859         if (target_bytes >= cli->cl_avail_grant) {
 860                 spin_unlock(&cli->cl_loi_list_lock);
 861                 RETURN(0);
 862         }
 863         spin_unlock(&cli->cl_loi_list_lock);
 864
 865         OBD_ALLOC_PTR(body);
 866         if (!body)
 867                 RETURN(-ENOMEM);
 868
 869         osc_announce_cached(cli, &body->oa, 0);
 870
 871         spin_lock(&cli->cl_loi_list_lock);
 872         if (target_bytes >= cli->cl_avail_grant) {
 873                 /* available grant has changed since target calculation */
 874                 spin_unlock(&cli->cl_loi_list_lock);
 875                 GOTO(out_free, rc = 0);
 876         }
 877         body->oa.o_grant = cli->cl_avail_grant - target_bytes;
 878         cli->cl_avail_grant = target_bytes;
 879         spin_unlock(&cli->cl_loi_list_lock);
 880         if (!(body->oa.o_valid & OBD_MD_FLFLAGS)) {
 881                 body->oa.o_valid |= OBD_MD_FLFLAGS;
 882                 body->oa.o_flags = 0;
 883         }
 884         body->oa.o_flags |= OBD_FL_SHRINK_GRANT;
 885         osc_update_next_shrink(cli);
 886
 887         rc = osc_set_info_async(NULL, cli->cl_import->imp_obd->obd_self_export,
 888                                 sizeof(KEY_GRANT_SHRINK), KEY_GRANT_SHRINK,
 889                                 sizeof(*body), body, NULL);
 890         if (rc != 0)
 891                 __osc_update_grant(cli, body->oa.o_grant);
 892 out_free:
 893         OBD_FREE_PTR(body);
 894         RETURN(rc);
 895 }
 896
 897 static int osc_should_shrink_grant(struct client_obd *client)
 898 {
 899         time64_t next_shrink = client->cl_next_shrink_grant;
 900
 901         if (client->cl_import == NULL)
 902                 return 0;
 903
 904         if (!OCD_HAS_FLAG(&client->cl_import->imp_connect_data, GRANT_SHRINK) ||
 905             client->cl_import->imp_grant_shrink_disabled) {
 906                 osc_update_next_shrink(client);
 907                 return 0;
 908         }
 909
 910         if (ktime_get_seconds() >= next_shrink - 5) {
 911                 /* Get the current RPC size directly, instead of going via:
 912                  * cli_brw_size(obd->u.cli.cl_import->imp_obd->obd_self_export)
 913                  * Keep comment here so that it can be found by searching. */
 914                 int brw_size = client->cl_max_pages_per_rpc << PAGE_SHIFT;
 915
 916                 if (client->cl_import->imp_state == LUSTRE_IMP_FULL &&
 917                     client->cl_avail_grant > brw_size)
 918                         return 1;
 919                 else
 920                         osc_update_next_shrink(client);
 921         }
 922         return 0;
 923 }
 924
 925 #define GRANT_SHRINK_RPC_BATCH  100
 926
 927 static struct delayed_work work;
 928
 929 static void osc_grant_work_handler(struct work_struct *data)
 930 {
 931         struct client_obd *cli;
 932         int rpc_sent;
 933         bool init_next_shrink = true;
 934         time64_t next_shrink = ktime_get_seconds() + GRANT_SHRINK_INTERVAL;
 935
 936         rpc_sent = 0;
 937         mutex_lock(&client_gtd.gtd_mutex);
 938         list_for_each_entry(cli, &client_gtd.gtd_clients,
 939                             cl_grant_chain) {
 940                 if (rpc_sent < GRANT_SHRINK_RPC_BATCH &&
 941                     osc_should_shrink_grant(cli)) {
 942                         osc_shrink_grant(cli);
 943                         rpc_sent++;
 944                 }
 945
 946                 if (!init_next_shrink) {
 947                         if (cli->cl_next_shrink_grant < next_shrink &&
 948                             cli->cl_next_shrink_grant > ktime_get_seconds())
 949                                 next_shrink = cli->cl_next_shrink_grant;
 950                 } else {
 951                         init_next_shrink = false;
 952                         next_shrink = cli->cl_next_shrink_grant;
 953                 }
 954         }
 955         mutex_unlock(&client_gtd.gtd_mutex);
 956
 957         if (client_gtd.gtd_stopped == 1)
 958                 return;
 959
 960         if (next_shrink > ktime_get_seconds()) {
 961                 time64_t delay = next_shrink - ktime_get_seconds();
 962
 963                 schedule_delayed_work(&work, cfs_time_seconds(delay));
 964         } else {
 965                 schedule_work(&work.work);
 966         }
 967 }
 968
 969 void osc_schedule_grant_work(void)
 970 {
 971         cancel_delayed_work_sync(&work);
 972         schedule_work(&work.work);
 973 }
 974
 975 /**
 976  * Start grant thread for returing grant to server for idle clients.
 977  */
 978 static int osc_start_grant_work(void)
 979 {
 980         client_gtd.gtd_stopped = 0;
 981         mutex_init(&client_gtd.gtd_mutex);
 982         INIT_LIST_HEAD(&client_gtd.gtd_clients);
 983
 984         INIT_DELAYED_WORK(&work, osc_grant_work_handler);
 985         schedule_work(&work.work);
 986
 987         return 0;
 988 }
 989
 990 static void osc_stop_grant_work(void)
 991 {
 992         client_gtd.gtd_stopped = 1;
 993         cancel_delayed_work_sync(&work);
 994 }
 995
 996 static void osc_add_grant_list(struct client_obd *client)
 997 {
 998         mutex_lock(&client_gtd.gtd_mutex);
 999         list_add(&client->cl_grant_chain, &client_gtd.gtd_clients);
1000         mutex_unlock(&client_gtd.gtd_mutex);
1001 }
1002
1003 static void osc_del_grant_list(struct client_obd *client)
1004 {
1005         if (list_empty(&client->cl_grant_chain))
1006                 return;
1007
1008         mutex_lock(&client_gtd.gtd_mutex);
1009         list_del_init(&client->cl_grant_chain);
1010         mutex_unlock(&client_gtd.gtd_mutex);
1011 }
1012
1013 void osc_init_grant(struct client_obd *cli, struct obd_connect_data *ocd)
1014 {
1015         /*
1016          * ocd_grant is the total grant amount we're expect to hold: if we've
1017          * been evicted, it's the new avail_grant amount, cl_dirty_pages will
1018          * drop to 0 as inflight RPCs fail out; otherwise, it's avail_grant +
1019          * dirty.
1020          *
1021          * race is tolerable here: if we're evicted, but imp_state already
1022          * left EVICTED state, then cl_dirty_pages must be 0 already.
1023          */
1024         spin_lock(&cli->cl_loi_list_lock);
1025         cli->cl_avail_grant = ocd->ocd_grant;
1026         if (cli->cl_import->imp_state != LUSTRE_IMP_EVICTED) {
1027                 unsigned long consumed = cli->cl_reserved_grant;
1028
1029                 if (OCD_HAS_FLAG(ocd, GRANT_PARAM))
1030                         consumed += cli->cl_dirty_grant;
1031                 else
1032                         consumed += cli->cl_dirty_pages << PAGE_SHIFT;
1033                 if (cli->cl_avail_grant < consumed) {
1034                         CERROR("%s: granted %ld but already consumed %ld\n",
1035                                cli_name(cli), cli->cl_avail_grant, consumed);
1036                         cli->cl_avail_grant = 0;
1037                 } else {
1038                         cli->cl_avail_grant -= consumed;
1039                 }
1040         }
1041
1042         if (OCD_HAS_FLAG(ocd, GRANT_PARAM)) {
1043                 u64 size;
1044                 int chunk_mask;
1045
1046                 /* overhead for each extent insertion */
1047                 cli->cl_grant_extent_tax = ocd->ocd_grant_tax_kb << 10;
1048                 /* determine the appropriate chunk size used by osc_extent. */
1049                 cli->cl_chunkbits = max_t(int, PAGE_SHIFT,
1050                                           ocd->ocd_grant_blkbits);
1051                 /* max_pages_per_rpc must be chunk aligned */
1052                 chunk_mask = ~((1 << (cli->cl_chunkbits - PAGE_SHIFT)) - 1);
1053                 cli->cl_max_pages_per_rpc = (cli->cl_max_pages_per_rpc +
1054                                              ~chunk_mask) & chunk_mask;
1055                 /* determine maximum extent size, in #pages */
1056                 size = (u64)ocd->ocd_grant_max_blks << ocd->ocd_grant_blkbits;
1057                 cli->cl_max_extent_pages = (size >> PAGE_SHIFT) ?: 1;
1058                 cli->cl_ocd_grant_param = 1;
1059         } else {
1060                 cli->cl_ocd_grant_param = 0;
1061                 cli->cl_grant_extent_tax = 0;
1062                 cli->cl_chunkbits = PAGE_SHIFT;
1063                 cli->cl_max_extent_pages = DT_MAX_BRW_PAGES;
1064         }
1065         spin_unlock(&cli->cl_loi_list_lock);
1066
1067         CDEBUG(D_CACHE,
1068                "%s, setting cl_avail_grant: %ld cl_lost_grant: %ld. chunk bits: %d cl_max_extent_pages: %d\n",
1069                cli_name(cli),
1070                cli->cl_avail_grant, cli->cl_lost_grant, cli->cl_chunkbits,
1071                cli->cl_max_extent_pages);
1072
1073         if (OCD_HAS_FLAG(ocd, GRANT_SHRINK) && list_empty(&cli->cl_grant_chain))
1074                 osc_add_grant_list(cli);
1075 }
1076 EXPORT_SYMBOL(osc_init_grant);
1077
1078 /* We assume that the reason this OSC got a short read is because it read
1079  * beyond the end of a stripe file; i.e. lustre is reading a sparse file
1080  * via the LOV, and it _knows_ it's reading inside the file, it's just that
1081  * this stripe never got written at or beyond this stripe offset yet. */
1082 static void handle_short_read(int nob_read, size_t page_count,
1083                               struct brw_page **pga)
1084 {
1085         char *ptr;
1086         int i = 0;
1087
1088         /* skip bytes read OK */
1089         while (nob_read > 0) {
1090                 LASSERT (page_count > 0);
1091
1092                 if (pga[i]->count > nob_read) {
1093                         /* EOF inside this page */
1094                         ptr = kmap(pga[i]->pg) +
1095                                 (pga[i]->off & ~PAGE_MASK);
1096                         memset(ptr + nob_read, 0, pga[i]->count - nob_read);
1097                         kunmap(pga[i]->pg);
1098                         page_count--;
1099                         i++;
1100                         break;
1101                 }
1102
1103                 nob_read -= pga[i]->count;
1104                 page_count--;
1105                 i++;
1106         }
1107
1108         /* zero remaining pages */
1109         while (page_count-- > 0) {
1110                 ptr = kmap(pga[i]->pg) + (pga[i]->off & ~PAGE_MASK);
1111                 memset(ptr, 0, pga[i]->count);
1112                 kunmap(pga[i]->pg);
1113                 i++;
1114         }
1115 }
1116
1117 static int check_write_rcs(struct ptlrpc_request *req,
1118                            int requested_nob, int niocount,
1119                            size_t page_count, struct brw_page **pga)
1120 {
1121         int     i;
1122         __u32   *remote_rcs;
1123
1124         remote_rcs = req_capsule_server_sized_get(&req->rq_pill, &RMF_RCS,
1125                                                   sizeof(*remote_rcs) *
1126                                                   niocount);
1127         if (remote_rcs == NULL) {
1128                 CDEBUG(D_INFO, "Missing/short RC vector on BRW_WRITE reply\n");
1129                 return(-EPROTO);
1130         }
1131
1132         /* return error if any niobuf was in error */
1133         for (i = 0; i < niocount; i++) {
1134                 if ((int)remote_rcs[i] < 0) {
1135                         CDEBUG(D_INFO, "rc[%d]: %d req %p\n",
1136                                i, remote_rcs[i], req);
1137                         return remote_rcs[i];
1138                 }
1139
1140                 if (remote_rcs[i] != 0) {
1141                         CDEBUG(D_INFO, "rc[%d] invalid (%d) req %p\n",
1142                                 i, remote_rcs[i], req);
1143                         return(-EPROTO);
1144                 }
1145         }
1146         if (req->rq_bulk != NULL &&
1147             req->rq_bulk->bd_nob_transferred != requested_nob) {
1148                 CERROR("Unexpected # bytes transferred: %d (requested %d)\n",
1149                        req->rq_bulk->bd_nob_transferred, requested_nob);
1150                 return(-EPROTO);
1151         }
1152
1153         return (0);
1154 }
1155
1156 static inline int can_merge_pages(struct brw_page *p1, struct brw_page *p2)
1157 {
1158         if (p1->flag != p2->flag) {
1159                 unsigned mask = ~(OBD_BRW_FROM_GRANT | OBD_BRW_NOCACHE |
1160                                   OBD_BRW_SYNC       | OBD_BRW_ASYNC   |
1161                                   OBD_BRW_NOQUOTA    | OBD_BRW_SOFT_SYNC);
1162
1163                 /* warn if we try to combine flags that we don't know to be
1164                  * safe to combine */
1165                 if (unlikely((p1->flag & mask) != (p2->flag & mask))) {
1166                         CWARN("Saw flags 0x%x and 0x%x in the same brw, please "
1167                               "report this at https://jira.whamcloud.com/\n",
1168                               p1->flag, p2->flag);
1169                 }
1170                 return 0;
1171         }
1172
1173         return (p1->off + p1->count == p2->off);
1174 }
1175
1176 #if IS_ENABLED(CONFIG_CRC_T10DIF)
1177 static int osc_checksum_bulk_t10pi(const char *obd_name, int nob,
1178                                    size_t pg_count, struct brw_page **pga,
1179                                    int opc, obd_dif_csum_fn *fn,
1180                                    int sector_size,
1181                                    u32 *check_sum, bool resend)
1182 {
1183         struct ahash_request *req;
1184         /* Used Adler as the default checksum type on top of DIF tags */
1185         unsigned char cfs_alg = cksum_obd2cfs(OBD_CKSUM_T10_TOP);
1186         struct page *__page;
1187         unsigned char *buffer;
1188         __u16 *guard_start;
1189         unsigned int bufsize;
1190         int guard_number;
1191         int used_number = 0;
1192         int used;
1193         u32 cksum;
1194         int rc = 0;
1195         int i = 0;
1196
1197         LASSERT(pg_count > 0);
1198
1199         __page = alloc_page(GFP_KERNEL);
1200         if (__page == NULL)
1201                 return -ENOMEM;
1202
1203         req = cfs_crypto_hash_init(cfs_alg, NULL, 0);
1204         if (IS_ERR(req)) {
1205                 rc = PTR_ERR(req);
1206                 CERROR("%s: unable to initialize checksum hash %s: rc = %d\n",
1207                        obd_name, cfs_crypto_hash_name(cfs_alg), rc);
1208                 GOTO(out, rc);
1209         }
1210
1211         buffer = kmap(__page);
1212         guard_start = (__u16 *)buffer;
1213         guard_number = PAGE_SIZE / sizeof(*guard_start);
1214         CDEBUG(D_PAGE | (resend ? D_HA : 0),
1215                "GRD tags per page=%u, resend=%u, bytes=%u, pages=%zu\n",
1216                guard_number, resend, nob, pg_count);
1217
1218         while (nob > 0 && pg_count > 0) {
1219                 unsigned int count = pga[i]->count > nob ? nob : pga[i]->count;
1220
1221                 /* corrupt the data before we compute the checksum, to
1222                  * simulate an OST->client data error */
1223                 if (unlikely(i == 0 && opc == OST_READ &&
1224                              OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_RECEIVE))) {
1225                         unsigned char *ptr = kmap(pga[i]->pg);
1226                         int off = pga[i]->off & ~PAGE_MASK;
1227
1228                         memcpy(ptr + off, "bad1", min_t(typeof(nob), 4, nob));
1229                         kunmap(pga[i]->pg);
1230                 }
1231
1232                 /*
1233                  * The left guard number should be able to hold checksums of a
1234                  * whole page
1235                  */
1236                 rc = obd_page_dif_generate_buffer(obd_name, pga[i]->pg,
1237                                                   pga[i]->off & ~PAGE_MASK,
1238                                                   count,
1239                                                   guard_start + used_number,
1240                                                   guard_number - used_number,
1241                                                   &used, sector_size,
1242                                                   fn);
1243                 if (unlikely(resend))
1244                         CDEBUG(D_PAGE | D_HA,
1245                                "pga[%u]: used %u off %llu+%u gen checksum: %*phN\n",
1246                                i, used, pga[i]->off & ~PAGE_MASK, count,
1247                                (int)(used * sizeof(*guard_start)),
1248                                guard_start + used_number);
1249                 if (rc)
1250                         break;
1251
1252                 used_number += used;
1253                 if (used_number == guard_number) {
1254                         cfs_crypto_hash_update_page(req, __page, 0,
1255                                 used_number * sizeof(*guard_start));
1256                         used_number = 0;
1257                 }
1258
1259                 nob -= pga[i]->count;
1260                 pg_count--;
1261                 i++;
1262         }
1263         kunmap(__page);
1264         if (rc)
1265                 GOTO(out, rc);
1266
1267         if (used_number != 0)
1268                 cfs_crypto_hash_update_page(req, __page, 0,
1269                         used_number * sizeof(*guard_start));
1270
1271         bufsize = sizeof(cksum);
1272         cfs_crypto_hash_final(req, (unsigned char *)&cksum, &bufsize);
1273
1274         /* For sending we only compute the wrong checksum instead
1275          * of corrupting the data so it is still correct on a redo */
1276         if (opc == OST_WRITE && OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_SEND))
1277                 cksum++;
1278
1279         *check_sum = cksum;
1280 out:
1281         __free_page(__page);
1282         return rc;
1283 }
1284 #else /* !CONFIG_CRC_T10DIF */
1285 #define obd_dif_ip_fn NULL
1286 #define obd_dif_crc_fn NULL
1287 #define osc_checksum_bulk_t10pi(name, nob, pgc, pga, opc, fn, ssize, csum, re) \
1288         -EOPNOTSUPP
1289 #endif /* CONFIG_CRC_T10DIF */
1290
1291 static int osc_checksum_bulk(int nob, size_t pg_count,
1292                              struct brw_page **pga, int opc,
1293                              enum cksum_types cksum_type,
1294                              u32 *cksum)
1295 {
1296         int                             i = 0;
1297         struct ahash_request           *req;
1298         unsigned int                    bufsize;
1299         unsigned char                   cfs_alg = cksum_obd2cfs(cksum_type);
1300
1301         LASSERT(pg_count > 0);
1302
1303         req = cfs_crypto_hash_init(cfs_alg, NULL, 0);
1304         if (IS_ERR(req)) {
1305                 CERROR("Unable to initialize checksum hash %s\n",
1306                        cfs_crypto_hash_name(cfs_alg));
1307                 return PTR_ERR(req);
1308         }
1309
1310         while (nob > 0 && pg_count > 0) {
1311                 unsigned int count = pga[i]->count > nob ? nob : pga[i]->count;
1312
1313                 /* corrupt the data before we compute the checksum, to
1314                  * simulate an OST->client data error */
1315                 if (i == 0 && opc == OST_READ &&
1316                     OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_RECEIVE)) {
1317                         unsigned char *ptr = kmap(pga[i]->pg);
1318                         int off = pga[i]->off & ~PAGE_MASK;
1319
1320                         memcpy(ptr + off, "bad1", min_t(typeof(nob), 4, nob));
1321                         kunmap(pga[i]->pg);
1322                 }
1323                 cfs_crypto_hash_update_page(req, pga[i]->pg,
1324                                             pga[i]->off & ~PAGE_MASK,
1325                                             count);
1326                 LL_CDEBUG_PAGE(D_PAGE, pga[i]->pg, "off %d\n",
1327                                (int)(pga[i]->off & ~PAGE_MASK));
1328
1329                 nob -= pga[i]->count;
1330                 pg_count--;
1331                 i++;
1332         }
1333
1334         bufsize = sizeof(*cksum);
1335         cfs_crypto_hash_final(req, (unsigned char *)cksum, &bufsize);
1336
1337         /* For sending we only compute the wrong checksum instead
1338          * of corrupting the data so it is still correct on a redo */
1339         if (opc == OST_WRITE && OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_SEND))
1340                 (*cksum)++;
1341
1342         return 0;
1343 }
1344
1345 static int osc_checksum_bulk_rw(const char *obd_name,
1346                                 enum cksum_types cksum_type,
1347                                 int nob, size_t pg_count,
1348                                 struct brw_page **pga, int opc,
1349                                 u32 *check_sum, bool resend)
1350 {
1351         obd_dif_csum_fn *fn = NULL;
1352         int sector_size = 0;
1353         int rc;
1354
1355         ENTRY;
1356         obd_t10_cksum2dif(cksum_type, &fn, &sector_size);
1357
1358         if (fn)
1359                 rc = osc_checksum_bulk_t10pi(obd_name, nob, pg_count, pga,
1360                                              opc, fn, sector_size, check_sum,
1361                                              resend);
1362         else
1363                 rc = osc_checksum_bulk(nob, pg_count, pga, opc, cksum_type,
1364                                        check_sum);
1365
1366         RETURN(rc);
1367 }
1368
1369 static inline void osc_release_bounce_pages(struct brw_page **pga,
1370                                             u32 page_count)
1371 {
1372 #ifdef HAVE_LUSTRE_CRYPTO
1373         int i;
1374
1375         for (i = 0; i < page_count; i++) {
1376                 /* Bounce pages allocated by a call to
1377                  * llcrypt_encrypt_pagecache_blocks() in osc_brw_prep_request()
1378                  * are identified thanks to the PageChecked flag.
1379                  */
1380                 if (PageChecked(pga[i]->pg))
1381                         llcrypt_finalize_bounce_page(&pga[i]->pg);
1382                 pga[i]->count -= pga[i]->bp_count_diff;
1383                 pga[i]->off += pga[i]->bp_off_diff;
1384         }
1385 #endif
1386 }
1387
1388 static int
1389 osc_brw_prep_request(int cmd, struct client_obd *cli, struct obdo *oa,
1390                      u32 page_count, struct brw_page **pga,
1391                      struct ptlrpc_request **reqp, int resend)
1392 {
1393         struct ptlrpc_request *req;
1394         struct ptlrpc_bulk_desc *desc;
1395         struct ost_body *body;
1396         struct obd_ioobj *ioobj;
1397         struct niobuf_remote *niobuf;
1398         int niocount, i, requested_nob, opc, rc, short_io_size = 0;
1399         struct osc_brw_async_args *aa;
1400         struct req_capsule *pill;
1401         struct brw_page *pg_prev;
1402         void *short_io_buf;
1403         const char *obd_name = cli->cl_import->imp_obd->obd_name;
1404         struct inode *inode = NULL;
1405         bool directio = false;
1406         bool enable_checksum = true;
1407
1408         ENTRY;
1409         if (pga[0]->pg) {
1410                 inode = page2inode(pga[0]->pg);
1411                 if (inode == NULL) {
1412                         /* Try to get reference to inode from cl_page if we are
1413                          * dealing with direct IO, as handled pages are not
1414                          * actual page cache pages.
1415                          */
1416                         struct osc_async_page *oap = brw_page2oap(pga[0]);
1417                         struct cl_page *clpage = oap2cl_page(oap);
1418
1419                         inode = clpage->cp_inode;
1420                         if (inode)
1421                                 directio = true;
1422                 }
1423         }
1424         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_PREP_REQ))
1425                 RETURN(-ENOMEM); /* Recoverable */
1426         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_PREP_REQ2))
1427                 RETURN(-EINVAL); /* Fatal */
1428
1429         if ((cmd & OBD_BRW_WRITE) != 0) {
1430                 opc = OST_WRITE;
1431                 req = ptlrpc_request_alloc_pool(cli->cl_import,
1432                                                 osc_rq_pool,
1433                                                 &RQF_OST_BRW_WRITE);
1434         } else {
1435                 opc = OST_READ;
1436                 req = ptlrpc_request_alloc(cli->cl_import, &RQF_OST_BRW_READ);
1437         }
1438         if (req == NULL)
1439                 RETURN(-ENOMEM);
1440
1441         if (opc == OST_WRITE && inode && IS_ENCRYPTED(inode)) {
1442                 for (i = 0; i < page_count; i++) {
1443                         struct brw_page *pg = pga[i];
1444                         struct page *data_page = NULL;
1445                         bool retried = false;
1446                         bool lockedbymyself;
1447                         u32 nunits = (pg->off & ~PAGE_MASK) + pg->count;
1448                         struct address_space *map_orig = NULL;
1449                         pgoff_t index_orig;
1450
1451 retry_encrypt:
1452                         if (nunits & ~LUSTRE_ENCRYPTION_MASK)
1453                                 nunits = (nunits & LUSTRE_ENCRYPTION_MASK) +
1454                                         LUSTRE_ENCRYPTION_UNIT_SIZE;
1455                         /* The page can already be locked when we arrive here.
1456                          * This is possible when cl_page_assume/vvp_page_assume
1457                          * is stuck on wait_on_page_writeback with page lock
1458                          * held. In this case there is no risk for the lock to
1459                          * be released while we are doing our encryption
1460                          * processing, because writeback against that page will
1461                          * end in vvp_page_completion_write/cl_page_completion,
1462                          * which means only once the page is fully processed.
1463                          */
1464                         lockedbymyself = trylock_page(pg->pg);
1465                         if (directio) {
1466                                 map_orig = pg->pg->mapping;
1467                                 pg->pg->mapping = inode->i_mapping;
1468                                 index_orig = pg->pg->index;
1469                                 pg->pg->index = pg->off >> PAGE_SHIFT;
1470                         }
1471                         data_page =
1472                                 llcrypt_encrypt_pagecache_blocks(pg->pg,
1473                                                                  nunits, 0,
1474                                                                  GFP_NOFS);
1475                         if (directio) {
1476                                 pg->pg->mapping = map_orig;
1477                                 pg->pg->index = index_orig;
1478                         }
1479                         if (lockedbymyself)
1480                                 unlock_page(pg->pg);
1481                         if (IS_ERR(data_page)) {
1482                                 rc = PTR_ERR(data_page);
1483                                 if (rc == -ENOMEM && !retried) {
1484                                         retried = true;
1485                                         rc = 0;
1486                                         goto retry_encrypt;
1487                                 }
1488                                 ptlrpc_request_free(req);
1489                                 RETURN(rc);
1490                         }
1491                         /* Set PageChecked flag on bounce page for
1492                          * disambiguation in osc_release_bounce_pages().
1493                          */
1494                         SetPageChecked(data_page);
1495                         pg->pg = data_page;
1496                         /* there should be no gap in the middle of page array */
1497                         if (i == page_count - 1) {
1498                                 struct osc_async_page *oap = brw_page2oap(pg);
1499
1500                                 oa->o_size = oap->oap_count +
1501                                         oap->oap_obj_off + oap->oap_page_off;
1502                         }
1503                         /* len is forced to nunits, and relative offset to 0
1504                          * so store the old, clear text info
1505                          */
1506                         pg->bp_count_diff = nunits - pg->count;
1507                         pg->count = nunits;
1508                         pg->bp_off_diff = pg->off & ~PAGE_MASK;
1509                         pg->off = pg->off & PAGE_MASK;
1510                 }
1511         } else if (opc == OST_READ && inode && IS_ENCRYPTED(inode)) {
1512                 for (i = 0; i < page_count; i++) {
1513                         struct brw_page *pg = pga[i];
1514                         u32 nunits = (pg->off & ~PAGE_MASK) + pg->count;
1515
1516                         if (nunits & ~LUSTRE_ENCRYPTION_MASK)
1517                                 nunits = (nunits & LUSTRE_ENCRYPTION_MASK) +
1518                                         LUSTRE_ENCRYPTION_UNIT_SIZE;
1519                         /* count/off are forced to cover the whole encryption
1520                          * unit size so that all encrypted data is stored on the
1521                          * OST, so adjust bp_{count,off}_diff for the size of
1522                          * the clear text.
1523                          */
1524                         pg->bp_count_diff = nunits - pg->count;
1525                         pg->count = nunits;
1526                         pg->bp_off_diff = pg->off & ~PAGE_MASK;
1527                         pg->off = pg->off & PAGE_MASK;
1528                 }
1529         }
1530
1531         for (niocount = i = 1; i < page_count; i++) {
1532                 if (!can_merge_pages(pga[i - 1], pga[i]))
1533                         niocount++;
1534         }
1535
1536         pill = &req->rq_pill;
1537         req_capsule_set_size(pill, &RMF_OBD_IOOBJ, RCL_CLIENT,
1538                              sizeof(*ioobj));
1539         req_capsule_set_size(pill, &RMF_NIOBUF_REMOTE, RCL_CLIENT,
1540                              niocount * sizeof(*niobuf));
1541
1542         for (i = 0; i < page_count; i++) {
1543                 short_io_size += pga[i]->count;
1544                 if (!inode || !IS_ENCRYPTED(inode)) {
1545                         pga[i]->bp_count_diff = 0;
1546                         pga[i]->bp_off_diff = 0;
1547                 }
1548         }
1549
1550         if (lnet_is_rdma_only_page(pga[0]->pg)) {
1551                 enable_checksum = false;
1552                 short_io_size = 0;
1553         }
1554
1555         /* Check if read/write is small enough to be a short io. */
1556         if (short_io_size > cli->cl_max_short_io_bytes || niocount > 1 ||
1557             !imp_connect_shortio(cli->cl_import))
1558                 short_io_size = 0;
1559
1560         req_capsule_set_size(pill, &RMF_SHORT_IO, RCL_CLIENT,
1561                              opc == OST_READ ? 0 : short_io_size);
1562         if (opc == OST_READ)
1563                 req_capsule_set_size(pill, &RMF_SHORT_IO, RCL_SERVER,
1564                                      short_io_size);
1565
1566         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, opc);
1567         if (rc) {
1568                 ptlrpc_request_free(req);
1569                 RETURN(rc);
1570         }
1571         osc_set_io_portal(req);
1572
1573         ptlrpc_at_set_req_timeout(req);
1574         /* ask ptlrpc not to resend on EINPROGRESS since BRWs have their own
1575          * retry logic */
1576         req->rq_no_retry_einprogress = 1;
1577
1578         if (short_io_size != 0) {
1579                 desc = NULL;
1580                 short_io_buf = NULL;
1581                 goto no_bulk;
1582         }
1583
1584         desc = ptlrpc_prep_bulk_imp(req, page_count,
1585                 cli->cl_import->imp_connect_data.ocd_brw_size >> LNET_MTU_BITS,
1586                 (opc == OST_WRITE ? PTLRPC_BULK_GET_SOURCE :
1587                         PTLRPC_BULK_PUT_SINK),
1588                 OST_BULK_PORTAL,
1589                 &ptlrpc_bulk_kiov_pin_ops);
1590
1591         if (desc == NULL)
1592                 GOTO(out, rc = -ENOMEM);
1593         /* NB request now owns desc and will free it when it gets freed */
1594 no_bulk:
1595         body = req_capsule_client_get(pill, &RMF_OST_BODY);
1596         ioobj = req_capsule_client_get(pill, &RMF_OBD_IOOBJ);
1597         niobuf = req_capsule_client_get(pill, &RMF_NIOBUF_REMOTE);
1598         LASSERT(body != NULL && ioobj != NULL && niobuf != NULL);
1599
1600         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
1601
1602         /* For READ and WRITE, we can't fill o_uid and o_gid using from_kuid()
1603          * and from_kgid(), because they are asynchronous. Fortunately, variable
1604          * oa contains valid o_uid and o_gid in these two operations.
1605          * Besides, filling o_uid and o_gid is enough for nrs-tbf, see LU-9658.
1606          * OBD_MD_FLUID and OBD_MD_FLUID is not set in order to avoid breaking
1607          * other process logic */
1608         body->oa.o_uid = oa->o_uid;
1609         body->oa.o_gid = oa->o_gid;
1610
1611         obdo_to_ioobj(oa, ioobj);
1612         ioobj->ioo_bufcnt = niocount;
1613         /* The high bits of ioo_max_brw tells server _maximum_ number of bulks
1614          * that might be send for this request.  The actual number is decided
1615          * when the RPC is finally sent in ptlrpc_register_bulk(). It sends
1616          * "max - 1" for old client compatibility sending "0", and also so the
1617          * the actual maximum is a power-of-two number, not one less. LU-1431 */
1618         if (desc != NULL)
1619                 ioobj_max_brw_set(ioobj, desc->bd_md_max_brw);
1620         else /* short io */
1621                 ioobj_max_brw_set(ioobj, 0);
1622
1623         if (short_io_size != 0) {
1624                 if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0) {
1625                         body->oa.o_valid |= OBD_MD_FLFLAGS;
1626                         body->oa.o_flags = 0;
1627                 }
1628                 body->oa.o_flags |= OBD_FL_SHORT_IO;
1629                 CDEBUG(D_CACHE, "Using short io for data transfer, size = %d\n",
1630                        short_io_size);
1631                 if (opc == OST_WRITE) {
1632                         short_io_buf = req_capsule_client_get(pill,
1633                                                               &RMF_SHORT_IO);
1634                         LASSERT(short_io_buf != NULL);
1635                 }
1636         }
1637
1638         LASSERT(page_count > 0);
1639         pg_prev = pga[0];
1640         for (requested_nob = i = 0; i < page_count; i++, niobuf++) {
1641                 struct brw_page *pg = pga[i];
1642                 int poff = pg->off & ~PAGE_MASK;
1643
1644                 LASSERT(pg->count > 0);
1645                 /* make sure there is no gap in the middle of page array */
1646                 LASSERTF(page_count == 1 ||
1647                          (ergo(i == 0, poff + pg->count == PAGE_SIZE) &&
1648                           ergo(i > 0 && i < page_count - 1,
1649                                poff == 0 && pg->count == PAGE_SIZE)   &&
1650                           ergo(i == page_count - 1, poff == 0)),
1651                          "i: %d/%d pg: %p off: %llu, count: %u\n",
1652                          i, page_count, pg, pg->off, pg->count);
1653                 LASSERTF(i == 0 || pg->off > pg_prev->off,
1654                          "i %d p_c %u pg %p [pri %lu ind %lu] off %llu"
1655                          " prev_pg %p [pri %lu ind %lu] off %llu\n",
1656                          i, page_count,
1657                          pg->pg, page_private(pg->pg), pg->pg->index, pg->off,
1658                          pg_prev->pg, page_private(pg_prev->pg),
1659                          pg_prev->pg->index, pg_prev->off);
1660                 LASSERT((pga[0]->flag & OBD_BRW_SRVLOCK) ==
1661                         (pg->flag & OBD_BRW_SRVLOCK));
1662                 if (short_io_size != 0 && opc == OST_WRITE) {
1663                         unsigned char *ptr = kmap_atomic(pg->pg);
1664
1665                         LASSERT(short_io_size >= requested_nob + pg->count);
1666                         memcpy(short_io_buf + requested_nob,
1667                                ptr + poff,
1668                                pg->count);
1669                         kunmap_atomic(ptr);
1670                 } else if (short_io_size == 0) {
1671                         desc->bd_frag_ops->add_kiov_frag(desc, pg->pg, poff,
1672                                                          pg->count);
1673                 }
1674                 requested_nob += pg->count;
1675
1676                 if (i > 0 && can_merge_pages(pg_prev, pg)) {
1677                         niobuf--;
1678                         niobuf->rnb_len += pg->count;
1679                 } else {
1680                         niobuf->rnb_offset = pg->off;
1681                         niobuf->rnb_len    = pg->count;
1682                         niobuf->rnb_flags  = pg->flag;
1683                 }
1684                 pg_prev = pg;
1685         }
1686
1687         LASSERTF((void *)(niobuf - niocount) ==
1688                 req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE),
1689                 "want %p - real %p\n", req_capsule_client_get(&req->rq_pill,
1690                 &RMF_NIOBUF_REMOTE), (void *)(niobuf - niocount));
1691
1692         osc_announce_cached(cli, &body->oa, opc == OST_WRITE ? requested_nob:0);
1693         if (resend) {
1694                 if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0) {
1695                         body->oa.o_valid |= OBD_MD_FLFLAGS;
1696                         body->oa.o_flags = 0;
1697                 }
1698                 body->oa.o_flags |= OBD_FL_RECOV_RESEND;
1699         }
1700
1701         if (osc_should_shrink_grant(cli))
1702                 osc_shrink_grant_local(cli, &body->oa);
1703
1704         if (!cli->cl_checksum || sptlrpc_flavor_has_bulk(&req->rq_flvr))
1705                 enable_checksum = false;
1706
1707         /* size[REQ_REC_OFF] still sizeof (*body) */
1708         if (opc == OST_WRITE) {
1709                 if (enable_checksum) {
1710                         /* store cl_cksum_type in a local variable since
1711                          * it can be changed via lprocfs */
1712                         enum cksum_types cksum_type = cli->cl_cksum_type;
1713
1714                         if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0)
1715                                 body->oa.o_flags = 0;
1716
1717                         body->oa.o_flags |= obd_cksum_type_pack(obd_name,
1718                                                                 cksum_type);
1719                         body->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1720
1721                         rc = osc_checksum_bulk_rw(obd_name, cksum_type,
1722                                                   requested_nob, page_count,
1723                                                   pga, OST_WRITE,
1724                                                   &body->oa.o_cksum, resend);
1725                         if (rc < 0) {
1726                                 CDEBUG(D_PAGE, "failed to checksum: rc = %d\n",
1727                                        rc);
1728                                 GOTO(out, rc);
1729                         }
1730                         CDEBUG(D_PAGE | (resend ? D_HA : 0),
1731                                "checksum at write origin: %x (%x)\n",
1732                                body->oa.o_cksum, cksum_type);
1733
1734                         /* save this in 'oa', too, for later checking */
1735                         oa->o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1736                         oa->o_flags |= obd_cksum_type_pack(obd_name,
1737                                                            cksum_type);
1738                 } else {
1739                         /* clear out the checksum flag, in case this is a
1740                          * resend but cl_checksum is no longer set. b=11238 */
1741                         oa->o_valid &= ~OBD_MD_FLCKSUM;
1742                 }
1743                 oa->o_cksum = body->oa.o_cksum;
1744                 /* 1 RC per niobuf */
1745                 req_capsule_set_size(pill, &RMF_RCS, RCL_SERVER,
1746                                      sizeof(__u32) * niocount);
1747         } else {
1748                 if (enable_checksum) {
1749                         if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0)
1750                                 body->oa.o_flags = 0;
1751                         body->oa.o_flags |= obd_cksum_type_pack(obd_name,
1752                                 cli->cl_cksum_type);
1753                         body->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1754                 }
1755
1756                 /* Client cksum has been already copied to wire obdo in previous
1757                  * lustre_set_wire_obdo(), and in the case a bulk-read is being
1758                  * resent due to cksum error, this will allow Server to
1759                  * check+dump pages on its side */
1760         }
1761         ptlrpc_request_set_replen(req);
1762
1763         aa = ptlrpc_req_async_args(aa, req);
1764         aa->aa_oa = oa;
1765         aa->aa_requested_nob = requested_nob;
1766         aa->aa_nio_count = niocount;
1767         aa->aa_page_count = page_count;
1768         aa->aa_resends = 0;
1769         aa->aa_ppga = pga;
1770         aa->aa_cli = cli;
1771         INIT_LIST_HEAD(&aa->aa_oaps);
1772
1773         *reqp = req;
1774         niobuf = req_capsule_client_get(pill, &RMF_NIOBUF_REMOTE);
1775         CDEBUG(D_RPCTRACE, "brw rpc %p - object "DOSTID" offset %lld<>%lld\n",
1776                 req, POSTID(&oa->o_oi), niobuf[0].rnb_offset,
1777                 niobuf[niocount - 1].rnb_offset + niobuf[niocount - 1].rnb_len);
1778         RETURN(0);
1779
1780  out:
1781         ptlrpc_req_finished(req);
1782         RETURN(rc);
1783 }
1784
1785 char dbgcksum_file_name[PATH_MAX];
1786
1787 static void dump_all_bulk_pages(struct obdo *oa, __u32 page_count,
1788                                 struct brw_page **pga, __u32 server_cksum,
1789                                 __u32 client_cksum)
1790 {
1791         struct file *filp;
1792         int rc, i;
1793         unsigned int len;
1794         char *buf;
1795
1796         /* will only keep dump of pages on first error for the same range in
1797          * file/fid, not during the resends/retries. */
1798         snprintf(dbgcksum_file_name, sizeof(dbgcksum_file_name),
1799                  "%s-checksum_dump-osc-"DFID":[%llu-%llu]-%x-%x",
1800                  (strncmp(libcfs_debug_file_path, "NONE", 4) != 0 ?
1801                   libcfs_debug_file_path : LIBCFS_DEBUG_FILE_PATH_DEFAULT),
1802                  oa->o_valid & OBD_MD_FLFID ? oa->o_parent_seq : 0ULL,
1803                  oa->o_valid & OBD_MD_FLFID ? oa->o_parent_oid : 0,
1804                  oa->o_valid & OBD_MD_FLFID ? oa->o_parent_ver : 0,
1805                  pga[0]->off,
1806                  pga[page_count-1]->off + pga[page_count-1]->count - 1,
1807                  client_cksum, server_cksum);
1808         CWARN("dumping checksum data to %s\n", dbgcksum_file_name);
1809         filp = filp_open(dbgcksum_file_name,
1810                          O_CREAT | O_EXCL | O_WRONLY | O_LARGEFILE, 0600);
1811         if (IS_ERR(filp)) {
1812                 rc = PTR_ERR(filp);
1813                 if (rc == -EEXIST)
1814                         CDEBUG(D_INFO, "%s: can't open to dump pages with "
1815                                "checksum error: rc = %d\n", dbgcksum_file_name,
1816                                rc);
1817                 else
1818                         CERROR("%s: can't open to dump pages with checksum "
1819                                "error: rc = %d\n", dbgcksum_file_name, rc);
1820                 return;
1821         }
1822
1823         for (i = 0; i < page_count; i++) {
1824                 len = pga[i]->count;
1825                 buf = kmap(pga[i]->pg);
1826                 while (len != 0) {
1827                         rc = cfs_kernel_write(filp, buf, len, &filp->f_pos);
1828                         if (rc < 0) {
1829                                 CERROR("%s: wanted to write %u but got %d "
1830                                        "error\n", dbgcksum_file_name, len, rc);
1831                                 break;
1832                         }
1833                         len -= rc;
1834                         buf += rc;
1835                 }
1836                 kunmap(pga[i]->pg);
1837         }
1838
1839         rc = vfs_fsync_range(filp, 0, LLONG_MAX, 1);
1840         if (rc)
1841                 CERROR("%s: sync returns %d\n", dbgcksum_file_name, rc);
1842         filp_close(filp, NULL);
1843
1844         libcfs_debug_dumplog();
1845 }
1846
1847 static int
1848 check_write_checksum(struct obdo *oa, const struct lnet_process_id *peer,
1849                      __u32 client_cksum, __u32 server_cksum,
1850                      struct osc_brw_async_args *aa)
1851 {
1852         const char *obd_name = aa->aa_cli->cl_import->imp_obd->obd_name;
1853         enum cksum_types cksum_type;
1854         obd_dif_csum_fn *fn = NULL;
1855         int sector_size = 0;
1856         __u32 new_cksum;
1857         char *msg;
1858         int rc;
1859
1860         if (server_cksum == client_cksum) {
1861                 CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
1862                 return 0;
1863         }
1864
1865         if (aa->aa_cli->cl_checksum_dump)
1866                 dump_all_bulk_pages(oa, aa->aa_page_count, aa->aa_ppga,
1867                                     server_cksum, client_cksum);
1868
1869         cksum_type = obd_cksum_type_unpack(oa->o_valid & OBD_MD_FLFLAGS ?
1870                                            oa->o_flags : 0);
1871
1872         switch (cksum_type) {
1873         case OBD_CKSUM_T10IP512:
1874                 fn = obd_dif_ip_fn;
1875                 sector_size = 512;
1876                 break;
1877         case OBD_CKSUM_T10IP4K:
1878                 fn = obd_dif_ip_fn;
1879                 sector_size = 4096;
1880                 break;
1881         case OBD_CKSUM_T10CRC512:
1882                 fn = obd_dif_crc_fn;
1883                 sector_size = 512;
1884                 break;
1885         case OBD_CKSUM_T10CRC4K:
1886                 fn = obd_dif_crc_fn;
1887                 sector_size = 4096;
1888                 break;
1889         default:
1890                 break;
1891         }
1892
1893         if (fn)
1894                 rc = osc_checksum_bulk_t10pi(obd_name, aa->aa_requested_nob,
1895                                              aa->aa_page_count, aa->aa_ppga,
1896                                              OST_WRITE, fn, sector_size,
1897                                              &new_cksum, true);
1898         else
1899                 rc = osc_checksum_bulk(aa->aa_requested_nob, aa->aa_page_count,
1900                                        aa->aa_ppga, OST_WRITE, cksum_type,
1901                                        &new_cksum);
1902
1903         if (rc < 0)
1904                 msg = "failed to calculate the client write checksum";
1905         else if (cksum_type != obd_cksum_type_unpack(aa->aa_oa->o_flags))
1906                 msg = "the server did not use the checksum type specified in "
1907                       "the original request - likely a protocol problem";
1908         else if (new_cksum == server_cksum)
1909                 msg = "changed on the client after we checksummed it - "
1910                       "likely false positive due to mmap IO (bug 11742)";
1911         else if (new_cksum == client_cksum)
1912                 msg = "changed in transit before arrival at OST";
1913         else
1914                 msg = "changed in transit AND doesn't match the original - "
1915                       "likely false positive due to mmap IO (bug 11742)";
1916
1917         LCONSOLE_ERROR_MSG(0x132, "%s: BAD WRITE CHECKSUM: %s: from %s inode "
1918                            DFID " object "DOSTID" extent [%llu-%llu], original "
1919                            "client csum %x (type %x), server csum %x (type %x),"
1920                            " client csum now %x\n",
1921                            obd_name, msg, libcfs_nid2str(peer->nid),
1922                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_seq : (__u64)0,
1923                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_oid : 0,
1924                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_ver : 0,
1925                            POSTID(&oa->o_oi), aa->aa_ppga[0]->off,
1926                            aa->aa_ppga[aa->aa_page_count - 1]->off +
1927                                 aa->aa_ppga[aa->aa_page_count-1]->count - 1,
1928                            client_cksum,
1929                            obd_cksum_type_unpack(aa->aa_oa->o_flags),
1930                            server_cksum, cksum_type, new_cksum);
1931         return 1;
1932 }
1933
1934 /* Note rc enters this function as number of bytes transferred */
1935 static int osc_brw_fini_request(struct ptlrpc_request *req, int rc)
1936 {
1937         struct osc_brw_async_args *aa = (void *)&req->rq_async_args;
1938         struct client_obd *cli = aa->aa_cli;
1939         const char *obd_name = cli->cl_import->imp_obd->obd_name;
1940         const struct lnet_process_id *peer =
1941                 &req->rq_import->imp_connection->c_peer;
1942         struct ost_body *body;
1943         u32 client_cksum = 0;
1944         struct inode *inode;
1945         unsigned int blockbits = 0, blocksize = 0;
1946
1947         ENTRY;
1948
1949         if (rc < 0 && rc != -EDQUOT) {
1950                 DEBUG_REQ(D_INFO, req, "Failed request: rc = %d", rc);
1951                 RETURN(rc);
1952         }
1953
1954         LASSERTF(req->rq_repmsg != NULL, "rc = %d\n", rc);
1955         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
1956         if (body == NULL) {
1957                 DEBUG_REQ(D_INFO, req, "cannot unpack body");
1958                 RETURN(-EPROTO);
1959         }
1960
1961         /* set/clear over quota flag for a uid/gid/projid */
1962         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE &&
1963             body->oa.o_valid & (OBD_MD_FLALLQUOTA)) {
1964                 unsigned qid[LL_MAXQUOTAS] = {
1965                                          body->oa.o_uid, body->oa.o_gid,
1966                                          body->oa.o_projid };
1967                 CDEBUG(D_QUOTA,
1968                        "setdq for [%u %u %u] with valid %#llx, flags %x\n",
1969                        body->oa.o_uid, body->oa.o_gid, body->oa.o_projid,
1970                        body->oa.o_valid, body->oa.o_flags);
1971                        osc_quota_setdq(cli, req->rq_xid, qid, body->oa.o_valid,
1972                                        body->oa.o_flags);
1973         }
1974
1975         osc_update_grant(cli, body);
1976
1977         if (rc < 0)
1978                 RETURN(rc);
1979
1980         if (aa->aa_oa->o_valid & OBD_MD_FLCKSUM)
1981                 client_cksum = aa->aa_oa->o_cksum; /* save for later */
1982
1983         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE) {
1984                 if (rc > 0) {
1985                         CERROR("%s: unexpected positive size %d\n",
1986                                obd_name, rc);
1987                         RETURN(-EPROTO);
1988                 }
1989
1990                 if (req->rq_bulk != NULL &&
1991                     sptlrpc_cli_unwrap_bulk_write(req, req->rq_bulk))
1992                         RETURN(-EAGAIN);
1993
1994                 if ((aa->aa_oa->o_valid & OBD_MD_FLCKSUM) && client_cksum &&
1995                     check_write_checksum(&body->oa, peer, client_cksum,
1996                                          body->oa.o_cksum, aa))
1997                         RETURN(-EAGAIN);
1998
1999                 rc = check_write_rcs(req, aa->aa_requested_nob,
2000                                      aa->aa_nio_count, aa->aa_page_count,
2001                                      aa->aa_ppga);
2002                 GOTO(out, rc);
2003         }
2004
2005         /* The rest of this function executes only for OST_READs */
2006
2007         if (req->rq_bulk == NULL) {
2008                 rc = req_capsule_get_size(&req->rq_pill, &RMF_SHORT_IO,
2009                                           RCL_SERVER);
2010                 LASSERT(rc == req->rq_status);
2011         } else {
2012                 /* if unwrap_bulk failed, return -EAGAIN to retry */
2013                 rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk, rc);
2014         }
2015         if (rc < 0)
2016                 GOTO(out, rc = -EAGAIN);
2017
2018         if (rc > aa->aa_requested_nob) {
2019                 CERROR("%s: unexpected size %d, requested %d\n", obd_name,
2020                        rc, aa->aa_requested_nob);
2021                 RETURN(-EPROTO);
2022         }
2023
2024         if (req->rq_bulk != NULL && rc != req->rq_bulk->bd_nob_transferred) {
2025                 CERROR("%s: unexpected size %d, transferred %d\n", obd_name,
2026                        rc, req->rq_bulk->bd_nob_transferred);
2027                 RETURN(-EPROTO);
2028         }
2029
2030         if (req->rq_bulk == NULL) {
2031                 /* short io */
2032                 int nob, pg_count, i = 0;
2033                 unsigned char *buf;
2034
2035                 CDEBUG(D_CACHE, "Using short io read, size %d\n", rc);
2036                 pg_count = aa->aa_page_count;
2037                 buf = req_capsule_server_sized_get(&req->rq_pill, &RMF_SHORT_IO,
2038                                                    rc);
2039                 nob = rc;
2040                 while (nob > 0 && pg_count > 0) {
2041                         unsigned char *ptr;
2042                         int count = aa->aa_ppga[i]->count > nob ?
2043                                     nob : aa->aa_ppga[i]->count;
2044
2045                         CDEBUG(D_CACHE, "page %p count %d\n",
2046                                aa->aa_ppga[i]->pg, count);
2047                         ptr = kmap_atomic(aa->aa_ppga[i]->pg);
2048                         memcpy(ptr + (aa->aa_ppga[i]->off & ~PAGE_MASK), buf,
2049                                count);
2050                         kunmap_atomic((void *) ptr);
2051
2052                         buf += count;
2053                         nob -= count;
2054                         i++;
2055                         pg_count--;
2056                 }
2057         }
2058
2059         if (rc < aa->aa_requested_nob)
2060                 handle_short_read(rc, aa->aa_page_count, aa->aa_ppga);
2061
2062         if (body->oa.o_valid & OBD_MD_FLCKSUM) {
2063                 static int cksum_counter;
2064                 u32 server_cksum = body->oa.o_cksum;
2065                 int nob = rc;
2066                 char *via = "";
2067                 char *router = "";
2068                 enum cksum_types cksum_type;
2069                 u32 o_flags = body->oa.o_valid & OBD_MD_FLFLAGS ?
2070                         body->oa.o_flags : 0;
2071
2072                 cksum_type = obd_cksum_type_unpack(o_flags);
2073                 rc = osc_checksum_bulk_rw(obd_name, cksum_type, nob,
2074                                           aa->aa_page_count, aa->aa_ppga,
2075                                           OST_READ, &client_cksum, false);
2076                 if (rc < 0)
2077                         GOTO(out, rc);
2078
2079                 if (req->rq_bulk != NULL &&
2080                     peer->nid != req->rq_bulk->bd_sender) {
2081                         via = " via ";
2082                         router = libcfs_nid2str(req->rq_bulk->bd_sender);
2083                 }
2084
2085                 if (server_cksum != client_cksum) {
2086                         struct ost_body *clbody;
2087                         __u32 client_cksum2;
2088                         u32 page_count = aa->aa_page_count;
2089
2090                         osc_checksum_bulk_rw(obd_name, cksum_type, nob,
2091                                              page_count, aa->aa_ppga,
2092                                              OST_READ, &client_cksum2, true);
2093                         clbody = req_capsule_client_get(&req->rq_pill,
2094                                                         &RMF_OST_BODY);
2095                         if (cli->cl_checksum_dump)
2096                                 dump_all_bulk_pages(&clbody->oa, page_count,
2097                                                     aa->aa_ppga, server_cksum,
2098                                                     client_cksum);
2099
2100                         LCONSOLE_ERROR_MSG(0x133, "%s: BAD READ CHECKSUM: from "
2101                                            "%s%s%s inode "DFID" object "DOSTID
2102                                            " extent [%llu-%llu], client %x/%x, "
2103                                            "server %x, cksum_type %x\n",
2104                                            obd_name,
2105                                            libcfs_nid2str(peer->nid),
2106                                            via, router,
2107                                            clbody->oa.o_valid & OBD_MD_FLFID ?
2108                                                 clbody->oa.o_parent_seq : 0ULL,
2109                                            clbody->oa.o_valid & OBD_MD_FLFID ?
2110                                                 clbody->oa.o_parent_oid : 0,
2111                                            clbody->oa.o_valid & OBD_MD_FLFID ?
2112                                                 clbody->oa.o_parent_ver : 0,
2113                                            POSTID(&body->oa.o_oi),
2114                                            aa->aa_ppga[0]->off,
2115                                            aa->aa_ppga[page_count-1]->off +
2116                                            aa->aa_ppga[page_count-1]->count - 1,
2117                                            client_cksum, client_cksum2,
2118                                            server_cksum, cksum_type);
2119                         cksum_counter = 0;
2120                         aa->aa_oa->o_cksum = client_cksum;
2121                         rc = -EAGAIN;
2122                 } else {
2123                         cksum_counter++;
2124                         CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
2125                         rc = 0;
2126                 }
2127         } else if (unlikely(client_cksum)) {
2128                 static int cksum_missed;
2129
2130                 cksum_missed++;
2131                 if ((cksum_missed & (-cksum_missed)) == cksum_missed)
2132                         CERROR("%s: checksum %u requested from %s but not sent\n",
2133                                obd_name, cksum_missed,
2134                                libcfs_nid2str(peer->nid));
2135         } else {
2136                 rc = 0;
2137         }
2138
2139         inode = page2inode(aa->aa_ppga[0]->pg);
2140         if (inode == NULL) {
2141                 /* Try to get reference to inode from cl_page if we are
2142                  * dealing with direct IO, as handled pages are not
2143                  * actual page cache pages.
2144                  */
2145                 struct osc_async_page *oap = brw_page2oap(aa->aa_ppga[0]);
2146
2147                 inode = oap2cl_page(oap)->cp_inode;
2148                 if (inode) {
2149                         blockbits = inode->i_blkbits;
2150                         blocksize = 1 << blockbits;
2151                 }
2152         }
2153         if (inode && IS_ENCRYPTED(inode)) {
2154                 int idx;
2155
2156                 if (!llcrypt_has_encryption_key(inode)) {
2157                         CDEBUG(D_SEC, "no enc key for ino %lu\n", inode->i_ino);
2158                         GOTO(out, rc);
2159                 }
2160                 for (idx = 0; idx < aa->aa_page_count; idx++) {
2161                         struct brw_page *pg = aa->aa_ppga[idx];
2162                         unsigned int offs = 0;
2163
2164                         while (offs < PAGE_SIZE) {
2165                                 /* do not decrypt if page is all 0s */
2166                                 if (memchr_inv(page_address(pg->pg) + offs, 0,
2167                                          LUSTRE_ENCRYPTION_UNIT_SIZE) == NULL) {
2168                                         /* if page is empty forward info to
2169                                          * upper layers (ll_io_zero_page) by
2170                                          * clearing PagePrivate2
2171                                          */
2172                                         if (!offs)
2173                                                 ClearPagePrivate2(pg->pg);
2174                                         break;
2175                                 }
2176
2177                                 if (blockbits) {
2178                                         /* This is direct IO case. Directly call
2179                                          * decrypt function that takes inode as
2180                                          * input parameter. Page does not need
2181                                          * to be locked.
2182                                          */
2183                                         u64 lblk_num =
2184                                                 ((u64)(pg->off >> PAGE_SHIFT) <<
2185                                                      (PAGE_SHIFT - blockbits)) +
2186                                                        (offs >> blockbits);
2187                                         unsigned int i;
2188
2189                                         for (i = offs;
2190                                              i < offs +
2191                                                     LUSTRE_ENCRYPTION_UNIT_SIZE;
2192                                              i += blocksize, lblk_num++) {
2193                                                 rc =
2194                                                   llcrypt_decrypt_block_inplace(
2195                                                           inode, pg->pg,
2196                                                           blocksize, i,
2197                                                           lblk_num);
2198                                                 if (rc)
2199                                                         break;
2200                                         }
2201                                 } else {
2202                                         rc = llcrypt_decrypt_pagecache_blocks(
2203                                                 pg->pg,
2204                                                 LUSTRE_ENCRYPTION_UNIT_SIZE,
2205                                                 offs);
2206                                 }
2207                                 if (rc)
2208                                         GOTO(out, rc);
2209
2210                                 offs += LUSTRE_ENCRYPTION_UNIT_SIZE;
2211                         }
2212                 }
2213         }
2214
2215 out:
2216         if (rc >= 0)
2217                 lustre_get_wire_obdo(&req->rq_import->imp_connect_data,
2218                                      aa->aa_oa, &body->oa);
2219
2220         RETURN(rc);
2221 }
2222
2223 static int osc_brw_redo_request(struct ptlrpc_request *request,
2224                                 struct osc_brw_async_args *aa, int rc)
2225 {
2226         struct ptlrpc_request *new_req;
2227         struct osc_brw_async_args *new_aa;
2228         struct osc_async_page *oap;
2229         ENTRY;
2230
2231         /* The below message is checked in replay-ost-single.sh test_8ae*/
2232         DEBUG_REQ(rc == -EINPROGRESS ? D_RPCTRACE : D_ERROR, request,
2233                   "redo for recoverable error %d", rc);
2234
2235         rc = osc_brw_prep_request(lustre_msg_get_opc(request->rq_reqmsg) ==
2236                                 OST_WRITE ? OBD_BRW_WRITE : OBD_BRW_READ,
2237                                   aa->aa_cli, aa->aa_oa, aa->aa_page_count,
2238                                   aa->aa_ppga, &new_req, 1);
2239         if (rc)
2240                 RETURN(rc);
2241
2242         list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
2243                 if (oap->oap_request != NULL) {
2244                         LASSERTF(request == oap->oap_request,
2245                                  "request %p != oap_request %p\n",
2246                                  request, oap->oap_request);
2247                 }
2248         }
2249         /*
2250          * New request takes over pga and oaps from old request.
2251          * Note that copying a list_head doesn't work, need to move it...
2252          */
2253         aa->aa_resends++;
2254         new_req->rq_interpret_reply = request->rq_interpret_reply;
2255         new_req->rq_async_args = request->rq_async_args;
2256         new_req->rq_commit_cb = request->rq_commit_cb;
2257         /* cap resend delay to the current request timeout, this is similar to
2258          * what ptlrpc does (see after_reply()) */
2259         if (aa->aa_resends > new_req->rq_timeout)
2260                 new_req->rq_sent = ktime_get_real_seconds() + new_req->rq_timeout;
2261         else
2262                 new_req->rq_sent = ktime_get_real_seconds() + aa->aa_resends;
2263         new_req->rq_generation_set = 1;
2264         new_req->rq_import_generation = request->rq_import_generation;
2265
2266         new_aa = ptlrpc_req_async_args(new_aa, new_req);
2267
2268         INIT_LIST_HEAD(&new_aa->aa_oaps);
2269         list_splice_init(&aa->aa_oaps, &new_aa->aa_oaps);
2270         INIT_LIST_HEAD(&new_aa->aa_exts);
2271         list_splice_init(&aa->aa_exts, &new_aa->aa_exts);
2272         new_aa->aa_resends = aa->aa_resends;
2273
2274         list_for_each_entry(oap, &new_aa->aa_oaps, oap_rpc_item) {
2275                 if (oap->oap_request) {
2276                         ptlrpc_req_finished(oap->oap_request);
2277                         oap->oap_request = ptlrpc_request_addref(new_req);
2278                 }
2279         }
2280
2281         /* XXX: This code will run into problem if we're going to support
2282          * to add a series of BRW RPCs into a self-defined ptlrpc_request_set
2283          * and wait for all of them to be finished. We should inherit request
2284          * set from old request. */
2285         ptlrpcd_add_req(new_req);
2286
2287         DEBUG_REQ(D_INFO, new_req, "new request");
2288         RETURN(0);
2289 }
2290
2291 /*
2292  * ugh, we want disk allocation on the target to happen in offset order.  we'll
2293  * follow sedgewicks advice and stick to the dead simple shellsort -- it'll do
2294  * fine for our small page arrays and doesn't require allocation.  its an
2295  * insertion sort that swaps elements that are strides apart, shrinking the
2296  * stride down until its '1' and the array is sorted.
2297  */
2298 static void sort_brw_pages(struct brw_page **array, int num)
2299 {
2300         int stride, i, j;
2301         struct brw_page *tmp;
2302
2303         if (num == 1)
2304                 return;
2305         for (stride = 1; stride < num ; stride = (stride * 3) + 1)
2306                 ;
2307
2308         do {
2309                 stride /= 3;
2310                 for (i = stride ; i < num ; i++) {
2311                         tmp = array[i];
2312                         j = i;
2313                         while (j >= stride && array[j - stride]->off > tmp->off) {
2314                                 array[j] = array[j - stride];
2315                                 j -= stride;
2316                         }
2317                         array[j] = tmp;
2318                 }
2319         } while (stride > 1);
2320 }
2321
2322 static void osc_release_ppga(struct brw_page **ppga, size_t count)
2323 {
2324         LASSERT(ppga != NULL);
2325         OBD_FREE_PTR_ARRAY_LARGE(ppga, count);
2326 }
2327
2328 static int brw_interpret(const struct lu_env *env,
2329                          struct ptlrpc_request *req, void *args, int rc)
2330 {
2331         struct osc_brw_async_args *aa = args;
2332         struct osc_extent *ext;
2333         struct osc_extent *tmp;
2334         struct client_obd *cli = aa->aa_cli;
2335         unsigned long transferred = 0;
2336
2337         ENTRY;
2338
2339         rc = osc_brw_fini_request(req, rc);
2340         CDEBUG(D_INODE, "request %p aa %p rc %d\n", req, aa, rc);
2341
2342         /* restore clear text pages */
2343         osc_release_bounce_pages(aa->aa_ppga, aa->aa_page_count);
2344
2345         /*
2346          * When server returns -EINPROGRESS, client should always retry
2347          * regardless of the number of times the bulk was resent already.
2348          */
2349         if (osc_recoverable_error(rc) && !req->rq_no_delay) {
2350                 if (req->rq_import_generation !=
2351                     req->rq_import->imp_generation) {
2352                         CDEBUG(D_HA, "%s: resend cross eviction for object: "
2353                                ""DOSTID", rc = %d.\n",
2354                                req->rq_import->imp_obd->obd_name,
2355                                POSTID(&aa->aa_oa->o_oi), rc);
2356                 } else if (rc == -EINPROGRESS ||
2357                            client_should_resend(aa->aa_resends, aa->aa_cli)) {
2358                         rc = osc_brw_redo_request(req, aa, rc);
2359                 } else {
2360                         CERROR("%s: too many resent retries for object: "
2361                                "%llu:%llu, rc = %d.\n",
2362                                req->rq_import->imp_obd->obd_name,
2363                                POSTID(&aa->aa_oa->o_oi), rc);
2364                 }
2365
2366                 if (rc == 0)
2367                         RETURN(0);
2368                 else if (rc == -EAGAIN || rc == -EINPROGRESS)
2369                         rc = -EIO;
2370         }
2371
2372         if (rc == 0) {
2373                 struct obdo *oa = aa->aa_oa;
2374                 struct cl_attr *attr = &osc_env_info(env)->oti_attr;
2375                 unsigned long valid = 0;
2376                 struct cl_object *obj;
2377                 struct osc_async_page *last;
2378
2379                 last = brw_page2oap(aa->aa_ppga[aa->aa_page_count - 1]);
2380                 obj = osc2cl(last->oap_obj);
2381
2382                 cl_object_attr_lock(obj);
2383                 if (oa->o_valid & OBD_MD_FLBLOCKS) {
2384                         attr->cat_blocks = oa->o_blocks;
2385                         valid |= CAT_BLOCKS;
2386                 }
2387                 if (oa->o_valid & OBD_MD_FLMTIME) {
2388                         attr->cat_mtime = oa->o_mtime;
2389                         valid |= CAT_MTIME;
2390                 }
2391                 if (oa->o_valid & OBD_MD_FLATIME) {
2392                         attr->cat_atime = oa->o_atime;
2393                         valid |= CAT_ATIME;
2394                 }
2395                 if (oa->o_valid & OBD_MD_FLCTIME) {
2396                         attr->cat_ctime = oa->o_ctime;
2397                         valid |= CAT_CTIME;
2398                 }
2399
2400                 if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE) {
2401                         struct lov_oinfo *loi = cl2osc(obj)->oo_oinfo;
2402                         loff_t last_off = last->oap_count + last->oap_obj_off +
2403                                 last->oap_page_off;
2404
2405                         /* Change file size if this is an out of quota or
2406                          * direct IO write and it extends the file size */
2407                         if (loi->loi_lvb.lvb_size < last_off) {
2408                                 attr->cat_size = last_off;
2409                                 valid |= CAT_SIZE;
2410                         }
2411                         /* Extend KMS if it's not a lockless write */
2412                         if (loi->loi_kms < last_off &&
2413                             oap2osc_page(last)->ops_srvlock == 0) {
2414                                 attr->cat_kms = last_off;
2415                                 valid |= CAT_KMS;
2416                         }
2417                 }
2418
2419                 if (valid != 0)
2420                         cl_object_attr_update(env, obj, attr, valid);
2421                 cl_object_attr_unlock(obj);
2422         }
2423         OBD_SLAB_FREE_PTR(aa->aa_oa, osc_obdo_kmem);
2424         aa->aa_oa = NULL;
2425
2426         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE && rc == 0)
2427                 osc_inc_unstable_pages(req);
2428
2429         list_for_each_entry_safe(ext, tmp, &aa->aa_exts, oe_link) {
2430                 list_del_init(&ext->oe_link);
2431                 osc_extent_finish(env, ext, 1,
2432                                   rc && req->rq_no_delay ? -EAGAIN : rc);
2433         }
2434         LASSERT(list_empty(&aa->aa_exts));
2435         LASSERT(list_empty(&aa->aa_oaps));
2436
2437         transferred = (req->rq_bulk == NULL ? /* short io */
2438                        aa->aa_requested_nob :
2439                        req->rq_bulk->bd_nob_transferred);
2440
2441         osc_release_ppga(aa->aa_ppga, aa->aa_page_count);
2442         ptlrpc_lprocfs_brw(req, transferred);
2443
2444         spin_lock(&cli->cl_loi_list_lock);
2445         /* We need to decrement before osc_ap_completion->osc_wake_cache_waiters
2446          * is called so we know whether to go to sync BRWs or wait for more
2447          * RPCs to complete */
2448         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE)
2449                 cli->cl_w_in_flight--;
2450         else
2451                 cli->cl_r_in_flight--;
2452         osc_wake_cache_waiters(cli);
2453         spin_unlock(&cli->cl_loi_list_lock);
2454
2455         osc_io_unplug(env, cli, NULL);
2456         RETURN(rc);
2457 }
2458
2459 static void brw_commit(struct ptlrpc_request *req)
2460 {
2461         /* If osc_inc_unstable_pages (via osc_extent_finish) races with
2462          * this called via the rq_commit_cb, I need to ensure
2463          * osc_dec_unstable_pages is still called. Otherwise unstable
2464          * pages may be leaked. */
2465         spin_lock(&req->rq_lock);
2466         if (likely(req->rq_unstable)) {
2467                 req->rq_unstable = 0;
2468                 spin_unlock(&req->rq_lock);
2469
2470                 osc_dec_unstable_pages(req);
2471         } else {
2472                 req->rq_committed = 1;
2473                 spin_unlock(&req->rq_lock);
2474         }
2475 }
2476
2477 /**
2478  * Build an RPC by the list of extent @ext_list. The caller must ensure
2479  * that the total pages in this list are NOT over max pages per RPC.
2480  * Extents in the list must be in OES_RPC state.
2481  */
2482 int osc_build_rpc(const struct lu_env *env, struct client_obd *cli,
2483                   struct list_head *ext_list, int cmd)
2484 {
2485         struct ptlrpc_request           *req = NULL;
2486         struct osc_extent               *ext;
2487         struct brw_page                 **pga = NULL;
2488         struct osc_brw_async_args       *aa = NULL;
2489         struct obdo                     *oa = NULL;
2490         struct osc_async_page           *oap;
2491         struct osc_object               *obj = NULL;
2492         struct cl_req_attr              *crattr = NULL;
2493         loff_t                          starting_offset = OBD_OBJECT_EOF;
2494         loff_t                          ending_offset = 0;
2495         /* '1' for consistency with code that checks !mpflag to restore */
2496         int mpflag = 1;
2497         int                             mem_tight = 0;
2498         int                             page_count = 0;
2499         bool                            soft_sync = false;
2500         bool                            ndelay = false;
2501         int                             i;
2502         int                             grant = 0;
2503         int                             rc;
2504         __u32                           layout_version = 0;
2505         LIST_HEAD(rpc_list);
2506         struct ost_body                 *body;
2507         ENTRY;
2508         LASSERT(!list_empty(ext_list));
2509
2510         /* add pages into rpc_list to build BRW rpc */
2511         list_for_each_entry(ext, ext_list, oe_link) {
2512                 LASSERT(ext->oe_state == OES_RPC);
2513                 mem_tight |= ext->oe_memalloc;
2514                 grant += ext->oe_grants;
2515                 page_count += ext->oe_nr_pages;
2516                 layout_version = max(layout_version, ext->oe_layout_version);
2517                 if (obj == NULL)
2518                         obj = ext->oe_obj;
2519         }
2520
2521         soft_sync = osc_over_unstable_soft_limit(cli);
2522         if (mem_tight)
2523                 mpflag = memalloc_noreclaim_save();
2524
2525         OBD_ALLOC_PTR_ARRAY_LARGE(pga, page_count);
2526         if (pga == NULL)
2527                 GOTO(out, rc = -ENOMEM);
2528
2529         OBD_SLAB_ALLOC_PTR_GFP(oa, osc_obdo_kmem, GFP_NOFS);
2530         if (oa == NULL)
2531                 GOTO(out, rc = -ENOMEM);
2532
2533         i = 0;
2534         list_for_each_entry(ext, ext_list, oe_link) {
2535                 list_for_each_entry(oap, &ext->oe_pages, oap_pending_item) {
2536                         if (mem_tight)
2537                                 oap->oap_brw_flags |= OBD_BRW_MEMALLOC;
2538                         if (soft_sync)
2539                                 oap->oap_brw_flags |= OBD_BRW_SOFT_SYNC;
2540                         pga[i] = &oap->oap_brw_page;
2541                         pga[i]->off = oap->oap_obj_off + oap->oap_page_off;
2542                         i++;
2543
2544                         list_add_tail(&oap->oap_rpc_item, &rpc_list);
2545                         if (starting_offset == OBD_OBJECT_EOF ||
2546                             starting_offset > oap->oap_obj_off)
2547                                 starting_offset = oap->oap_obj_off;
2548                         else
2549                                 LASSERT(oap->oap_page_off == 0);
2550                         if (ending_offset < oap->oap_obj_off + oap->oap_count)
2551                                 ending_offset = oap->oap_obj_off +
2552                                                 oap->oap_count;
2553                         else
2554                                 LASSERT(oap->oap_page_off + oap->oap_count ==
2555                                         PAGE_SIZE);
2556                 }
2557                 if (ext->oe_ndelay)
2558                         ndelay = true;
2559         }
2560
2561         /* first page in the list */
2562         oap = list_first_entry(&rpc_list, typeof(*oap), oap_rpc_item);
2563
2564         crattr = &osc_env_info(env)->oti_req_attr;
2565         memset(crattr, 0, sizeof(*crattr));
2566         crattr->cra_type = (cmd & OBD_BRW_WRITE) ? CRT_WRITE : CRT_READ;
2567         crattr->cra_flags = ~0ULL;
2568         crattr->cra_page = oap2cl_page(oap);
2569         crattr->cra_oa = oa;
2570         cl_req_attr_set(env, osc2cl(obj), crattr);
2571
2572         if (cmd == OBD_BRW_WRITE) {
2573                 oa->o_grant_used = grant;
2574                 if (layout_version > 0) {
2575                         CDEBUG(D_LAYOUT, DFID": write with layout version %u\n",
2576                                PFID(&oa->o_oi.oi_fid), layout_version);
2577
2578                         oa->o_layout_version = layout_version;
2579                         oa->o_valid |= OBD_MD_LAYOUT_VERSION;
2580                 }
2581         }
2582
2583         sort_brw_pages(pga, page_count);
2584         rc = osc_brw_prep_request(cmd, cli, oa, page_count, pga, &req, 0);
2585         if (rc != 0) {
2586                 CERROR("prep_req failed: %d\n", rc);
2587                 GOTO(out, rc);
2588         }
2589
2590         req->rq_commit_cb = brw_commit;
2591         req->rq_interpret_reply = brw_interpret;
2592         req->rq_memalloc = mem_tight != 0;
2593         oap->oap_request = ptlrpc_request_addref(req);
2594         if (ndelay) {
2595                 req->rq_no_resend = req->rq_no_delay = 1;
2596                 /* probably set a shorter timeout value.
2597                  * to handle ETIMEDOUT in brw_interpret() correctly. */
2598                 /* lustre_msg_set_timeout(req, req->rq_timeout / 2); */
2599         }
2600
2601         /* Need to update the timestamps after the request is built in case
2602          * we race with setattr (locally or in queue at OST).  If OST gets
2603          * later setattr before earlier BRW (as determined by the request xid),
2604          * the OST will not use BRW timestamps.  Sadly, there is no obvious
2605          * way to do this in a single call.  bug 10150 */
2606         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
2607         crattr->cra_oa = &body->oa;
2608         crattr->cra_flags = OBD_MD_FLMTIME | OBD_MD_FLCTIME | OBD_MD_FLATIME;
2609         cl_req_attr_set(env, osc2cl(obj), crattr);
2610         lustre_msg_set_jobid(req->rq_reqmsg, crattr->cra_jobid);
2611
2612         aa = ptlrpc_req_async_args(aa, req);
2613         INIT_LIST_HEAD(&aa->aa_oaps);
2614         list_splice_init(&rpc_list, &aa->aa_oaps);
2615         INIT_LIST_HEAD(&aa->aa_exts);
2616         list_splice_init(ext_list, &aa->aa_exts);
2617
2618         spin_lock(&cli->cl_loi_list_lock);
2619         starting_offset >>= PAGE_SHIFT;
2620         if (cmd == OBD_BRW_READ) {
2621                 cli->cl_r_in_flight++;
2622                 lprocfs_oh_tally_log2(&cli->cl_read_page_hist, page_count);
2623                 lprocfs_oh_tally(&cli->cl_read_rpc_hist, cli->cl_r_in_flight);
2624                 lprocfs_oh_tally_log2(&cli->cl_read_offset_hist,
2625                                       starting_offset + 1);
2626         } else {
2627                 cli->cl_w_in_flight++;
2628                 lprocfs_oh_tally_log2(&cli->cl_write_page_hist, page_count);
2629                 lprocfs_oh_tally(&cli->cl_write_rpc_hist, cli->cl_w_in_flight);
2630                 lprocfs_oh_tally_log2(&cli->cl_write_offset_hist,
2631                                       starting_offset + 1);
2632         }
2633         spin_unlock(&cli->cl_loi_list_lock);
2634
2635         DEBUG_REQ(D_INODE, req, "%d pages, aa %p, now %ur/%uw in flight",
2636                   page_count, aa, cli->cl_r_in_flight,
2637                   cli->cl_w_in_flight);
2638         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_DELAY_IO, cfs_fail_val);
2639
2640         ptlrpcd_add_req(req);
2641         rc = 0;
2642         EXIT;
2643
2644 out:
2645         if (mem_tight)
2646                 memalloc_noreclaim_restore(mpflag);
2647
2648         if (rc != 0) {
2649                 LASSERT(req == NULL);
2650
2651                 if (oa)
2652                         OBD_SLAB_FREE_PTR(oa, osc_obdo_kmem);
2653                 if (pga) {
2654                         osc_release_bounce_pages(pga, page_count);
2655                         osc_release_ppga(pga, page_count);
2656                 }
2657                 /* this should happen rarely and is pretty bad, it makes the
2658                  * pending list not follow the dirty order
2659                  */
2660                 while ((ext = list_first_entry_or_null(ext_list,
2661                                                        struct osc_extent,
2662                                                        oe_link)) != NULL) {
2663                         list_del_init(&ext->oe_link);
2664                         osc_extent_finish(env, ext, 0, rc);
2665                 }
2666         }
2667         RETURN(rc);
2668 }
2669
2670 /* This is to refresh our lock in face of no RPCs. */
2671 void osc_send_empty_rpc(struct osc_object *osc, pgoff_t start)
2672 {
2673         struct ptlrpc_request *req;
2674         struct obdo oa;
2675         struct brw_page bpg = { .off = start, .count = 1};
2676         struct brw_page *pga = &bpg;
2677         int rc;
2678
2679         memset(&oa, 0, sizeof(oa));
2680         oa.o_oi = osc->oo_oinfo->loi_oi;
2681         oa.o_valid = OBD_MD_FLID | OBD_MD_FLGROUP | OBD_MD_FLFLAGS;
2682         /* For updated servers - don't do a read */
2683         oa.o_flags = OBD_FL_NORPC;
2684
2685         rc = osc_brw_prep_request(OBD_BRW_READ, osc_cli(osc), &oa, 1, &pga,
2686                                   &req, 0);
2687
2688         /* If we succeeded we ship it off, if not there's no point in doing
2689          * anything. Also no resends.
2690          * No interpret callback, no commit callback.
2691          */
2692         if (!rc) {
2693                 req->rq_no_resend = 1;
2694                 ptlrpcd_add_req(req);
2695         }
2696 }
2697
2698 static int osc_set_lock_data(struct ldlm_lock *lock, void *data)
2699 {
2700         int set = 0;
2701
2702         LASSERT(lock != NULL);
2703
2704         lock_res_and_lock(lock);
2705
2706         if (lock->l_ast_data == NULL)
2707                 lock->l_ast_data = data;
2708         if (lock->l_ast_data == data)
2709                 set = 1;
2710
2711         unlock_res_and_lock(lock);
2712
2713         return set;
2714 }
2715
2716 int osc_enqueue_fini(struct ptlrpc_request *req, osc_enqueue_upcall_f upcall,
2717                      void *cookie, struct lustre_handle *lockh,
2718                      enum ldlm_mode mode, __u64 *flags, bool speculative,
2719                      int errcode)
2720 {
2721         bool intent = *flags & LDLM_FL_HAS_INTENT;
2722         int rc;
2723         ENTRY;
2724
2725         /* The request was created before ldlm_cli_enqueue call. */
2726         if (intent && errcode == ELDLM_LOCK_ABORTED) {
2727                 struct ldlm_reply *rep;
2728
2729                 rep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
2730                 LASSERT(rep != NULL);
2731
2732                 rep->lock_policy_res1 =
2733                         ptlrpc_status_ntoh(rep->lock_policy_res1);
2734                 if (rep->lock_policy_res1)
2735                         errcode = rep->lock_policy_res1;
2736                 if (!speculative)
2737                         *flags |= LDLM_FL_LVB_READY;
2738         } else if (errcode == ELDLM_OK) {
2739                 *flags |= LDLM_FL_LVB_READY;
2740         }
2741
2742         /* Call the update callback. */
2743         rc = (*upcall)(cookie, lockh, errcode);
2744
2745         /* release the reference taken in ldlm_cli_enqueue() */
2746         if (errcode == ELDLM_LOCK_MATCHED)
2747                 errcode = ELDLM_OK;
2748         if (errcode == ELDLM_OK && lustre_handle_is_used(lockh))
2749                 ldlm_lock_decref(lockh, mode);
2750
2751         RETURN(rc);
2752 }
2753
2754 int osc_enqueue_interpret(const struct lu_env *env, struct ptlrpc_request *req,
2755                           void *args, int rc)
2756 {
2757         struct osc_enqueue_args *aa = args;
2758         struct ldlm_lock *lock;
2759         struct lustre_handle *lockh = &aa->oa_lockh;
2760         enum ldlm_mode mode = aa->oa_mode;
2761         struct ost_lvb *lvb = aa->oa_lvb;
2762         __u32 lvb_len = sizeof(*lvb);
2763         __u64 flags = 0;
2764         struct ldlm_enqueue_info einfo = {
2765                 .ei_type = aa->oa_type,
2766                 .ei_mode = mode,
2767         };
2768
2769         ENTRY;
2770
2771         /* ldlm_cli_enqueue is holding a reference on the lock, so it must
2772          * be valid. */
2773         lock = ldlm_handle2lock(lockh);
2774         LASSERTF(lock != NULL,
2775                  "lockh %#llx, req %p, aa %p - client evicted?\n",
2776                  lockh->cookie, req, aa);
2777
2778         /* Take an additional reference so that a blocking AST that
2779          * ldlm_cli_enqueue_fini() might post for a failed lock, is guaranteed
2780          * to arrive after an upcall has been executed by
2781          * osc_enqueue_fini(). */
2782         ldlm_lock_addref(lockh, mode);
2783
2784         /* Let cl_lock_state_wait fail with -ERESTARTSYS to unuse sublocks. */
2785         OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_ENQUEUE_HANG, 2);
2786
2787         /* Let CP AST to grant the lock first. */
2788         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_ENQ_RACE, 1);
2789
2790         if (aa->oa_speculative) {
2791                 LASSERT(aa->oa_lvb == NULL);
2792                 LASSERT(aa->oa_flags == NULL);
2793                 aa->oa_flags = &flags;
2794         }
2795
2796         /* Complete obtaining the lock procedure. */
2797         rc = ldlm_cli_enqueue_fini(aa->oa_exp, req, &einfo, 1, aa->oa_flags,
2798                                    lvb, lvb_len, lockh, rc);
2799         /* Complete osc stuff. */
2800         rc = osc_enqueue_fini(req, aa->oa_upcall, aa->oa_cookie, lockh, mode,
2801                               aa->oa_flags, aa->oa_speculative, rc);
2802
2803         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_CANCEL_RACE, 10);
2804
2805         ldlm_lock_decref(lockh, mode);
2806         LDLM_LOCK_PUT(lock);
2807         RETURN(rc);
2808 }
2809
2810 /* When enqueuing asynchronously, locks are not ordered, we can obtain a lock
2811  * from the 2nd OSC before a lock from the 1st one. This does not deadlock with
2812  * other synchronous requests, however keeping some locks and trying to obtain
2813  * others may take a considerable amount of time in a case of ost failure; and
2814  * when other sync requests do not get released lock from a client, the client
2815  * is evicted from the cluster -- such scenarious make the life difficult, so
2816  * release locks just after they are obtained. */
2817 int osc_enqueue_base(struct obd_export *exp, struct ldlm_res_id *res_id,
2818                      __u64 *flags, union ldlm_policy_data *policy,
2819                      struct ost_lvb *lvb, osc_enqueue_upcall_f upcall,
2820                      void *cookie, struct ldlm_enqueue_info *einfo,
2821                      struct ptlrpc_request_set *rqset, int async,
2822                      bool speculative)
2823 {
2824         struct obd_device *obd = exp->exp_obd;
2825         struct lustre_handle lockh = { 0 };
2826         struct ptlrpc_request *req = NULL;
2827         int intent = *flags & LDLM_FL_HAS_INTENT;
2828         __u64 match_flags = *flags;
2829         enum ldlm_mode mode;
2830         int rc;
2831         ENTRY;
2832
2833         /* Filesystem lock extents are extended to page boundaries so that
2834          * dealing with the page cache is a little smoother.  */
2835         policy->l_extent.start -= policy->l_extent.start & ~PAGE_MASK;
2836         policy->l_extent.end |= ~PAGE_MASK;
2837
2838         /* Next, search for already existing extent locks that will cover us */
2839         /* If we're trying to read, we also search for an existing PW lock.  The
2840          * VFS and page cache already protect us locally, so lots of readers/
2841          * writers can share a single PW lock.
2842          *
2843          * There are problems with conversion deadlocks, so instead of
2844          * converting a read lock to a write lock, we'll just enqueue a new
2845          * one.
2846          *
2847          * At some point we should cancel the read lock instead of making them
2848          * send us a blocking callback, but there are problems with canceling
2849          * locks out from other users right now, too. */
2850         mode = einfo->ei_mode;
2851         if (einfo->ei_mode == LCK_PR)
2852                 mode |= LCK_PW;
2853         /* Normal lock requests must wait for the LVB to be ready before
2854          * matching a lock; speculative lock requests do not need to,
2855          * because they will not actually use the lock. */
2856         if (!speculative)
2857                 match_flags |= LDLM_FL_LVB_READY;
2858         if (intent != 0)
2859                 match_flags |= LDLM_FL_BLOCK_GRANTED;
2860         mode = ldlm_lock_match(obd->obd_namespace, match_flags, res_id,
2861                                einfo->ei_type, policy, mode, &lockh);
2862         if (mode) {
2863                 struct ldlm_lock *matched;
2864
2865                 if (*flags & LDLM_FL_TEST_LOCK)
2866                         RETURN(ELDLM_OK);
2867
2868                 matched = ldlm_handle2lock(&lockh);
2869                 if (speculative) {
2870                         /* This DLM lock request is speculative, and does not
2871                          * have an associated IO request. Therefore if there
2872                          * is already a DLM lock, it wll just inform the
2873                          * caller to cancel the request for this stripe.*/
2874                         lock_res_and_lock(matched);
2875                         if (ldlm_extent_equal(&policy->l_extent,
2876                             &matched->l_policy_data.l_extent))
2877                                 rc = -EEXIST;
2878                         else
2879                                 rc = -ECANCELED;
2880                         unlock_res_and_lock(matched);
2881
2882                         ldlm_lock_decref(&lockh, mode);
2883                         LDLM_LOCK_PUT(matched);
2884                         RETURN(rc);
2885                 } else if (osc_set_lock_data(matched, einfo->ei_cbdata)) {
2886                         *flags |= LDLM_FL_LVB_READY;
2887
2888                         /* We already have a lock, and it's referenced. */
2889                         (*upcall)(cookie, &lockh, ELDLM_LOCK_MATCHED);
2890
2891                         ldlm_lock_decref(&lockh, mode);
2892                         LDLM_LOCK_PUT(matched);
2893                         RETURN(ELDLM_OK);
2894                 } else {
2895                         ldlm_lock_decref(&lockh, mode);
2896                         LDLM_LOCK_PUT(matched);
2897                 }
2898         }
2899
2900         if (*flags & (LDLM_FL_TEST_LOCK | LDLM_FL_MATCH_LOCK))
2901                 RETURN(-ENOLCK);
2902
2903         /* users of osc_enqueue() can pass this flag for ldlm_lock_match() */
2904         *flags &= ~LDLM_FL_BLOCK_GRANTED;
2905
2906         rc = ldlm_cli_enqueue(exp, &req, einfo, res_id, policy, flags, lvb,
2907                               sizeof(*lvb), LVB_T_OST, &lockh, async);
2908         if (async) {
2909                 if (!rc) {
2910                         struct osc_enqueue_args *aa;
2911                         aa = ptlrpc_req_async_args(aa, req);
2912                         aa->oa_exp         = exp;
2913                         aa->oa_mode        = einfo->ei_mode;
2914                         aa->oa_type        = einfo->ei_type;
2915                         lustre_handle_copy(&aa->oa_lockh, &lockh);
2916                         aa->oa_upcall      = upcall;
2917                         aa->oa_cookie      = cookie;
2918                         aa->oa_speculative = speculative;
2919                         if (!speculative) {
2920                                 aa->oa_flags  = flags;
2921                                 aa->oa_lvb    = lvb;
2922                         } else {
2923                                 /* speculative locks are essentially to enqueue
2924                                  * a DLM lock  in advance, so we don't care
2925                                  * about the result of the enqueue. */
2926                                 aa->oa_lvb    = NULL;
2927                                 aa->oa_flags  = NULL;
2928                         }
2929
2930                         req->rq_interpret_reply = osc_enqueue_interpret;
2931                         ptlrpc_set_add_req(rqset, req);
2932                 }
2933                 RETURN(rc);
2934         }
2935
2936         rc = osc_enqueue_fini(req, upcall, cookie, &lockh, einfo->ei_mode,
2937                               flags, speculative, rc);
2938
2939         RETURN(rc);
2940 }
2941
2942 int osc_match_base(const struct lu_env *env, struct obd_export *exp,
2943                    struct ldlm_res_id *res_id, enum ldlm_type type,
2944                    union ldlm_policy_data *policy, enum ldlm_mode mode,
2945                    __u64 *flags, struct osc_object *obj,
2946                    struct lustre_handle *lockh, enum ldlm_match_flags match_flags)
2947 {
2948         struct obd_device *obd = exp->exp_obd;
2949         __u64 lflags = *flags;
2950         enum ldlm_mode rc;
2951         ENTRY;
2952
2953         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_MATCH))
2954                 RETURN(-EIO);
2955
2956         /* Filesystem lock extents are extended to page boundaries so that
2957          * dealing with the page cache is a little smoother */
2958         policy->l_extent.start -= policy->l_extent.start & ~PAGE_MASK;
2959         policy->l_extent.end |= ~PAGE_MASK;
2960
2961         /* Next, search for already existing extent locks that will cover us */
2962         rc = ldlm_lock_match_with_skip(obd->obd_namespace, lflags, 0,
2963                                         res_id, type, policy, mode, lockh,
2964                                         match_flags);
2965         if (rc == 0 || lflags & LDLM_FL_TEST_LOCK)
2966                 RETURN(rc);
2967
2968         if (obj != NULL) {
2969                 struct ldlm_lock *lock = ldlm_handle2lock(lockh);
2970
2971                 LASSERT(lock != NULL);
2972                 if (osc_set_lock_data(lock, obj)) {
2973                         lock_res_and_lock(lock);
2974                         if (!ldlm_is_lvb_cached(lock)) {
2975                                 LASSERT(lock->l_ast_data == obj);
2976                                 osc_lock_lvb_update(env, obj, lock, NULL);
2977                                 ldlm_set_lvb_cached(lock);
2978                         }
2979                         unlock_res_and_lock(lock);
2980                 } else {
2981                         ldlm_lock_decref(lockh, rc);
2982                         rc = 0;
2983                 }
2984                 LDLM_LOCK_PUT(lock);
2985         }
2986         RETURN(rc);
2987 }
2988
2989 static int osc_statfs_interpret(const struct lu_env *env,
2990                                 struct ptlrpc_request *req, void *args, int rc)
2991 {
2992         struct osc_async_args *aa = args;
2993         struct obd_statfs *msfs;
2994
2995         ENTRY;
2996         if (rc == -EBADR)
2997                 /*
2998                  * The request has in fact never been sent due to issues at
2999                  * a higher level (LOV).  Exit immediately since the caller
3000                  * is aware of the problem and takes care of the clean up.
3001                  */
3002                 RETURN(rc);
3003
3004         if ((rc == -ENOTCONN || rc == -EAGAIN) &&
3005             (aa->aa_oi->oi_flags & OBD_STATFS_NODELAY))
3006                 GOTO(out, rc = 0);
3007
3008         if (rc != 0)
3009                 GOTO(out, rc);
3010
3011         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
3012         if (msfs == NULL)
3013                 GOTO(out, rc = -EPROTO);
3014
3015         *aa->aa_oi->oi_osfs = *msfs;
3016 out:
3017         rc = aa->aa_oi->oi_cb_up(aa->aa_oi, rc);
3018
3019         RETURN(rc);
3020 }
3021
3022 static int osc_statfs_async(struct obd_export *exp,
3023                             struct obd_info *oinfo, time64_t max_age,
3024                             struct ptlrpc_request_set *rqset)
3025 {
3026         struct obd_device     *obd = class_exp2obd(exp);
3027         struct ptlrpc_request *req;
3028         struct osc_async_args *aa;
3029         int rc;
3030         ENTRY;
3031
3032         if (obd->obd_osfs_age >= max_age) {
3033                 CDEBUG(D_SUPER,
3034                        "%s: use %p cache blocks %llu/%llu objects %llu/%llu\n",
3035                        obd->obd_name, &obd->obd_osfs,
3036                        obd->obd_osfs.os_bavail, obd->obd_osfs.os_blocks,
3037                        obd->obd_osfs.os_ffree, obd->obd_osfs.os_files);
3038                 spin_lock(&obd->obd_osfs_lock);
3039                 memcpy(oinfo->oi_osfs, &obd->obd_osfs, sizeof(*oinfo->oi_osfs));
3040                 spin_unlock(&obd->obd_osfs_lock);
3041                 oinfo->oi_flags |= OBD_STATFS_FROM_CACHE;
3042                 if (oinfo->oi_cb_up)
3043                         oinfo->oi_cb_up(oinfo, 0);
3044
3045                 RETURN(0);
3046         }
3047
3048         /* We could possibly pass max_age in the request (as an absolute
3049          * timestamp or a "seconds.usec ago") so the target can avoid doing
3050          * extra calls into the filesystem if that isn't necessary (e.g.
3051          * during mount that would help a bit).  Having relative timestamps
3052          * is not so great if request processing is slow, while absolute
3053          * timestamps are not ideal because they need time synchronization. */
3054         req = ptlrpc_request_alloc(obd->u.cli.cl_import, &RQF_OST_STATFS);
3055         if (req == NULL)
3056                 RETURN(-ENOMEM);
3057
3058         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
3059         if (rc) {
3060                 ptlrpc_request_free(req);
3061                 RETURN(rc);
3062         }
3063         ptlrpc_request_set_replen(req);
3064         req->rq_request_portal = OST_CREATE_PORTAL;
3065         ptlrpc_at_set_req_timeout(req);
3066
3067         if (oinfo->oi_flags & OBD_STATFS_NODELAY) {
3068                 /* procfs requests not want stat in wait for avoid deadlock */
3069                 req->rq_no_resend = 1;
3070                 req->rq_no_delay = 1;
3071         }
3072
3073         req->rq_interpret_reply = osc_statfs_interpret;
3074         aa = ptlrpc_req_async_args(aa, req);
3075         aa->aa_oi = oinfo;
3076
3077         ptlrpc_set_add_req(rqset, req);
3078         RETURN(0);
3079 }
3080
3081 static int osc_statfs(const struct lu_env *env, struct obd_export *exp,
3082                       struct obd_statfs *osfs, time64_t max_age, __u32 flags)
3083 {
3084         struct obd_device     *obd = class_exp2obd(exp);
3085         struct obd_statfs     *msfs;
3086         struct ptlrpc_request *req;
3087         struct obd_import     *imp, *imp0;
3088         int rc;
3089         ENTRY;
3090
3091         /*Since the request might also come from lprocfs, so we need
3092          *sync this with client_disconnect_export Bug15684
3093          */
3094         with_imp_locked(obd, imp0, rc)
3095                 imp = class_import_get(imp0);
3096         if (rc)
3097                 RETURN(rc);
3098
3099         /* We could possibly pass max_age in the request (as an absolute
3100          * timestamp or a "seconds.usec ago") so the target can avoid doing
3101          * extra calls into the filesystem if that isn't necessary (e.g.
3102          * during mount that would help a bit).  Having relative timestamps
3103          * is not so great if request processing is slow, while absolute
3104          * timestamps are not ideal because they need time synchronization. */
3105         req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
3106
3107         class_import_put(imp);
3108
3109         if (req == NULL)
3110                 RETURN(-ENOMEM);
3111
3112         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
3113         if (rc) {
3114                 ptlrpc_request_free(req);
3115                 RETURN(rc);
3116         }
3117         ptlrpc_request_set_replen(req);
3118         req->rq_request_portal = OST_CREATE_PORTAL;
3119         ptlrpc_at_set_req_timeout(req);
3120
3121         if (flags & OBD_STATFS_NODELAY) {
3122                 /* procfs requests not want stat in wait for avoid deadlock */
3123                 req->rq_no_resend = 1;
3124                 req->rq_no_delay = 1;
3125         }
3126
3127         rc = ptlrpc_queue_wait(req);
3128         if (rc)
3129                 GOTO(out, rc);
3130
3131         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
3132         if (msfs == NULL)
3133                 GOTO(out, rc = -EPROTO);
3134
3135         *osfs = *msfs;
3136
3137         EXIT;
3138 out:
3139         ptlrpc_req_finished(req);
3140         return rc;
3141 }
3142
3143 static int osc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
3144                          void *karg, void __user *uarg)
3145 {
3146         struct obd_device *obd = exp->exp_obd;
3147         struct obd_ioctl_data *data = karg;
3148         int rc = 0;
3149
3150         ENTRY;
3151         if (!try_module_get(THIS_MODULE)) {
3152                 CERROR("%s: cannot get module '%s'\n", obd->obd_name,
3153                        module_name(THIS_MODULE));
3154                 return -EINVAL;
3155         }
3156         switch (cmd) {
3157         case OBD_IOC_CLIENT_RECOVER:
3158                 rc = ptlrpc_recover_import(obd->u.cli.cl_import,
3159                                            data->ioc_inlbuf1, 0);
3160                 if (rc > 0)
3161                         rc = 0;
3162                 break;
3163         case IOC_OSC_SET_ACTIVE:
3164                 rc = ptlrpc_set_import_active(obd->u.cli.cl_import,
3165                                               data->ioc_offset);
3166                 break;
3167         default:
3168                 rc = -ENOTTY;
3169                 CDEBUG(D_INODE, "%s: unrecognised ioctl %#x by %s: rc = %d\n",
3170                        obd->obd_name, cmd, current->comm, rc);
3171                 break;
3172         }
3173
3174         module_put(THIS_MODULE);
3175         return rc;
3176 }
3177
3178 int osc_set_info_async(const struct lu_env *env, struct obd_export *exp,
3179                        u32 keylen, void *key, u32 vallen, void *val,
3180                        struct ptlrpc_request_set *set)
3181 {
3182         struct ptlrpc_request *req;
3183         struct obd_device     *obd = exp->exp_obd;
3184         struct obd_import     *imp = class_exp2cliimp(exp);
3185         char                  *tmp;
3186         int                    rc;
3187         ENTRY;
3188
3189         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_SHUTDOWN, 10);
3190
3191         if (KEY_IS(KEY_CHECKSUM)) {
3192                 if (vallen != sizeof(int))
3193                         RETURN(-EINVAL);
3194                 exp->exp_obd->u.cli.cl_checksum = (*(int *)val) ? 1 : 0;
3195                 RETURN(0);
3196         }
3197
3198         if (KEY_IS(KEY_SPTLRPC_CONF)) {
3199                 sptlrpc_conf_client_adapt(obd);
3200                 RETURN(0);
3201         }
3202
3203         if (KEY_IS(KEY_FLUSH_CTX)) {
3204                 sptlrpc_import_flush_my_ctx(imp);
3205                 RETURN(0);
3206         }
3207
3208         if (KEY_IS(KEY_CACHE_LRU_SHRINK)) {
3209                 struct client_obd *cli = &obd->u.cli;
3210                 long nr = atomic_long_read(&cli->cl_lru_in_list) >> 1;
3211                 long target = *(long *)val;
3212
3213                 nr = osc_lru_shrink(env, cli, min(nr, target), true);
3214                 *(long *)val -= nr;
3215                 RETURN(0);
3216         }
3217
3218         if (!set && !KEY_IS(KEY_GRANT_SHRINK))
3219                 RETURN(-EINVAL);
3220
3221         /* We pass all other commands directly to OST. Since nobody calls osc
3222            methods directly and everybody is supposed to go through LOV, we
3223            assume lov checked invalid values for us.
3224            The only recognised values so far are evict_by_nid and mds_conn.
3225            Even if something bad goes through, we'd get a -EINVAL from OST
3226            anyway. */
3227
3228         req = ptlrpc_request_alloc(imp, KEY_IS(KEY_GRANT_SHRINK) ?
3229                                                 &RQF_OST_SET_GRANT_INFO :
3230                                                 &RQF_OBD_SET_INFO);
3231         if (req == NULL)
3232                 RETURN(-ENOMEM);
3233
3234         req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
3235                              RCL_CLIENT, keylen);
3236         if (!KEY_IS(KEY_GRANT_SHRINK))
3237                 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_VAL,
3238                                      RCL_CLIENT, vallen);
3239         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SET_INFO);
3240         if (rc) {
3241                 ptlrpc_request_free(req);
3242                 RETURN(rc);
3243         }
3244
3245         tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
3246         memcpy(tmp, key, keylen);
3247         tmp = req_capsule_client_get(&req->rq_pill, KEY_IS(KEY_GRANT_SHRINK) ?
3248                                                         &RMF_OST_BODY :
3249                                                         &RMF_SETINFO_VAL);
3250         memcpy(tmp, val, vallen);
3251
3252         if (KEY_IS(KEY_GRANT_SHRINK)) {
3253                 struct osc_grant_args *aa;
3254                 struct obdo *oa;
3255
3256                 aa = ptlrpc_req_async_args(aa, req);
3257                 OBD_SLAB_ALLOC_PTR_GFP(oa, osc_obdo_kmem, GFP_NOFS);
3258                 if (!oa) {
3259                         ptlrpc_req_finished(req);
3260                         RETURN(-ENOMEM);
3261                 }
3262                 *oa = ((struct ost_body *)val)->oa;
3263                 aa->aa_oa = oa;
3264                 req->rq_interpret_reply = osc_shrink_grant_interpret;
3265         }
3266
3267         ptlrpc_request_set_replen(req);
3268         if (!KEY_IS(KEY_GRANT_SHRINK)) {
3269                 LASSERT(set != NULL);
3270                 ptlrpc_set_add_req(set, req);
3271                 ptlrpc_check_set(NULL, set);
3272         } else {
3273                 ptlrpcd_add_req(req);
3274         }
3275
3276         RETURN(0);
3277 }
3278 EXPORT_SYMBOL(osc_set_info_async);
3279
3280 int osc_reconnect(const struct lu_env *env, struct obd_export *exp,
3281                   struct obd_device *obd, struct obd_uuid *cluuid,
3282                   struct obd_connect_data *data, void *localdata)
3283 {
3284         struct client_obd *cli = &obd->u.cli;
3285
3286         if (data != NULL && (data->ocd_connect_flags & OBD_CONNECT_GRANT)) {
3287                 long lost_grant;
3288                 long grant;
3289
3290                 spin_lock(&cli->cl_loi_list_lock);
3291                 grant = cli->cl_avail_grant + cli->cl_reserved_grant;
3292                 if (data->ocd_connect_flags & OBD_CONNECT_GRANT_PARAM) {
3293                         /* restore ocd_grant_blkbits as client page bits */
3294                         data->ocd_grant_blkbits = PAGE_SHIFT;
3295                         grant += cli->cl_dirty_grant;
3296                 } else {
3297                         grant += cli->cl_dirty_pages << PAGE_SHIFT;
3298                 }
3299                 data->ocd_grant = grant ? : 2 * cli_brw_size(obd);
3300                 lost_grant = cli->cl_lost_grant;
3301                 cli->cl_lost_grant = 0;
3302                 spin_unlock(&cli->cl_loi_list_lock);
3303
3304                 CDEBUG(D_RPCTRACE, "ocd_connect_flags: %#llx ocd_version: %d"
3305                        " ocd_grant: %d, lost: %ld.\n", data->ocd_connect_flags,
3306                        data->ocd_version, data->ocd_grant, lost_grant);
3307         }
3308
3309         RETURN(0);
3310 }
3311 EXPORT_SYMBOL(osc_reconnect);
3312
3313 int osc_disconnect(struct obd_export *exp)
3314 {
3315         struct obd_device *obd = class_exp2obd(exp);
3316         int rc;
3317
3318         rc = client_disconnect_export(exp);
3319         /**
3320          * Initially we put del_shrink_grant before disconnect_export, but it
3321          * causes the following problem if setup (connect) and cleanup
3322          * (disconnect) are tangled together.
3323          *      connect p1                     disconnect p2
3324          *   ptlrpc_connect_import
3325          *     ...............               class_manual_cleanup
3326          *                                     osc_disconnect
3327          *                                     del_shrink_grant
3328          *   ptlrpc_connect_interrupt
3329          *     osc_init_grant
3330          *   add this client to shrink list
3331          *                                      cleanup_osc
3332          * Bang! grant shrink thread trigger the shrink. BUG18662
3333          */
3334         osc_del_grant_list(&obd->u.cli);
3335         return rc;
3336 }
3337 EXPORT_SYMBOL(osc_disconnect);
3338
3339 int osc_ldlm_resource_invalidate(struct cfs_hash *hs, struct cfs_hash_bd *bd,
3340                                  struct hlist_node *hnode, void *arg)
3341 {
3342         struct lu_env *env = arg;
3343         struct ldlm_resource *res = cfs_hash_object(hs, hnode);
3344         struct ldlm_lock *lock;
3345         struct osc_object *osc = NULL;
3346         ENTRY;
3347
3348         lock_res(res);
3349         list_for_each_entry(lock, &res->lr_granted, l_res_link) {
3350                 if (lock->l_ast_data != NULL && osc == NULL) {
3351                         osc = lock->l_ast_data;
3352                         cl_object_get(osc2cl(osc));
3353                 }
3354
3355                 /* clear LDLM_FL_CLEANED flag to make sure it will be canceled
3356                  * by the 2nd round of ldlm_namespace_clean() call in
3357                  * osc_import_event(). */
3358                 ldlm_clear_cleaned(lock);
3359         }
3360         unlock_res(res);
3361
3362         if (osc != NULL) {
3363                 osc_object_invalidate(env, osc);
3364                 cl_object_put(env, osc2cl(osc));
3365         }
3366
3367         RETURN(0);
3368 }
3369 EXPORT_SYMBOL(osc_ldlm_resource_invalidate);
3370
3371 static int osc_import_event(struct obd_device *obd,
3372                             struct obd_import *imp,
3373                             enum obd_import_event event)
3374 {
3375         struct client_obd *cli;
3376         int rc = 0;
3377
3378         ENTRY;
3379         LASSERT(imp->imp_obd == obd);
3380
3381         switch (event) {
3382         case IMP_EVENT_DISCON: {
3383                 cli = &obd->u.cli;
3384                 spin_lock(&cli->cl_loi_list_lock);
3385                 cli->cl_avail_grant = 0;
3386                 cli->cl_lost_grant = 0;
3387                 spin_unlock(&cli->cl_loi_list_lock);
3388                 break;
3389         }
3390         case IMP_EVENT_INACTIVE: {
3391                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_INACTIVE);
3392                 break;
3393         }
3394         case IMP_EVENT_INVALIDATE: {
3395                 struct ldlm_namespace *ns = obd->obd_namespace;
3396                 struct lu_env         *env;
3397                 __u16                  refcheck;
3398
3399                 ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
3400
3401                 env = cl_env_get(&refcheck);
3402                 if (!IS_ERR(env)) {
3403                         osc_io_unplug(env, &obd->u.cli, NULL);
3404
3405                         cfs_hash_for_each_nolock(ns->ns_rs_hash,
3406                                                  osc_ldlm_resource_invalidate,
3407                                                  env, 0);
3408                         cl_env_put(env, &refcheck);
3409
3410                         ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
3411                 } else
3412                         rc = PTR_ERR(env);
3413                 break;
3414         }
3415         case IMP_EVENT_ACTIVE: {
3416                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE);
3417                 break;
3418         }
3419         case IMP_EVENT_OCD: {
3420                 struct obd_connect_data *ocd = &imp->imp_connect_data;
3421
3422                 if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT)
3423                         osc_init_grant(&obd->u.cli, ocd);
3424
3425                 /* See bug 7198 */
3426                 if (ocd->ocd_connect_flags & OBD_CONNECT_REQPORTAL)
3427                         imp->imp_client->cli_request_portal =OST_REQUEST_PORTAL;
3428
3429                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_OCD);
3430                 break;
3431         }
3432         case IMP_EVENT_DEACTIVATE: {
3433                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_DEACTIVATE);
3434                 break;
3435         }
3436         case IMP_EVENT_ACTIVATE: {
3437                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVATE);
3438                 break;
3439         }
3440         default:
3441                 CERROR("Unknown import event %d\n", event);
3442                 LBUG();
3443         }
3444         RETURN(rc);
3445 }
3446
3447 /**
3448  * Determine whether the lock can be canceled before replaying the lock
3449  * during recovery, see bug16774 for detailed information.
3450  *
3451  * \retval zero the lock can't be canceled
3452  * \retval other ok to cancel
3453  */
3454 static int osc_cancel_weight(struct ldlm_lock *lock)
3455 {
3456         /*
3457          * Cancel all unused and granted extent lock.
3458          */
3459         if (lock->l_resource->lr_type == LDLM_EXTENT &&
3460             ldlm_is_granted(lock) &&
3461             osc_ldlm_weigh_ast(lock) == 0)
3462                 RETURN(1);
3463
3464         RETURN(0);
3465 }
3466
3467 static int brw_queue_work(const struct lu_env *env, void *data)
3468 {
3469         struct client_obd *cli = data;
3470
3471         CDEBUG(D_CACHE, "Run writeback work for client obd %p.\n", cli);
3472
3473         osc_io_unplug(env, cli, NULL);
3474         RETURN(0);
3475 }
3476
3477 int osc_setup_common(struct obd_device *obd, struct lustre_cfg *lcfg)
3478 {
3479         struct client_obd *cli = &obd->u.cli;
3480         void *handler;
3481         int rc;
3482
3483         ENTRY;
3484
3485         rc = ptlrpcd_addref();
3486         if (rc)
3487                 RETURN(rc);
3488
3489         rc = client_obd_setup(obd, lcfg);
3490         if (rc)
3491                 GOTO(out_ptlrpcd, rc);
3492
3493
3494         handler = ptlrpcd_alloc_work(cli->cl_import, brw_queue_work, cli);
3495         if (IS_ERR(handler))
3496                 GOTO(out_ptlrpcd_work, rc = PTR_ERR(handler));
3497         cli->cl_writeback_work = handler;
3498
3499         handler = ptlrpcd_alloc_work(cli->cl_import, lru_queue_work, cli);
3500         if (IS_ERR(handler))
3501                 GOTO(out_ptlrpcd_work, rc = PTR_ERR(handler));
3502         cli->cl_lru_work = handler;
3503
3504         rc = osc_quota_setup(obd);
3505         if (rc)
3506                 GOTO(out_ptlrpcd_work, rc);
3507
3508         cli->cl_grant_shrink_interval = GRANT_SHRINK_INTERVAL;
3509         osc_update_next_shrink(cli);
3510
3511         RETURN(rc);
3512
3513 out_ptlrpcd_work:
3514         if (cli->cl_writeback_work != NULL) {
3515                 ptlrpcd_destroy_work(cli->cl_writeback_work);
3516                 cli->cl_writeback_work = NULL;
3517         }
3518         if (cli->cl_lru_work != NULL) {
3519                 ptlrpcd_destroy_work(cli->cl_lru_work);
3520                 cli->cl_lru_work = NULL;
3521         }
3522         client_obd_cleanup(obd);
3523 out_ptlrpcd:
3524         ptlrpcd_decref();
3525         RETURN(rc);
3526 }
3527 EXPORT_SYMBOL(osc_setup_common);
3528
3529 int osc_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
3530 {
3531         struct client_obd *cli = &obd->u.cli;
3532         int                adding;
3533         int                added;
3534         int                req_count;
3535         int                rc;
3536
3537         ENTRY;
3538
3539         rc = osc_setup_common(obd, lcfg);
3540         if (rc < 0)
3541                 RETURN(rc);
3542
3543         rc = osc_tunables_init(obd);
3544         if (rc)
3545                 RETURN(rc);
3546
3547         /*
3548          * We try to control the total number of requests with a upper limit
3549          * osc_reqpool_maxreqcount. There might be some race which will cause
3550          * over-limit allocation, but it is fine.
3551          */
3552         req_count = atomic_read(&osc_pool_req_count);
3553         if (req_count < osc_reqpool_maxreqcount) {
3554                 adding = cli->cl_max_rpcs_in_flight + 2;
3555                 if (req_count + adding > osc_reqpool_maxreqcount)
3556                         adding = osc_reqpool_maxreqcount - req_count;
3557
3558                 added = ptlrpc_add_rqs_to_pool(osc_rq_pool, adding);
3559                 atomic_add(added, &osc_pool_req_count);
3560         }
3561
3562         ns_register_cancel(obd->obd_namespace, osc_cancel_weight);
3563
3564         spin_lock(&osc_shrink_lock);
3565         list_add_tail(&cli->cl_shrink_list, &osc_shrink_list);
3566         spin_unlock(&osc_shrink_lock);
3567         cli->cl_import->imp_idle_timeout = osc_idle_timeout;
3568         cli->cl_import->imp_idle_debug = D_HA;
3569
3570         RETURN(0);
3571 }
3572
3573 int osc_precleanup_common(struct obd_device *obd)
3574 {
3575         struct client_obd *cli = &obd->u.cli;
3576         ENTRY;
3577
3578         /* LU-464
3579          * for echo client, export may be on zombie list, wait for
3580          * zombie thread to cull it, because cli.cl_import will be
3581          * cleared in client_disconnect_export():
3582          *   class_export_destroy() -> obd_cleanup() ->
3583          *   echo_device_free() -> echo_client_cleanup() ->
3584          *   obd_disconnect() -> osc_disconnect() ->
3585          *   client_disconnect_export()
3586          */
3587         obd_zombie_barrier();
3588         if (cli->cl_writeback_work) {
3589                 ptlrpcd_destroy_work(cli->cl_writeback_work);
3590                 cli->cl_writeback_work = NULL;
3591         }
3592
3593         if (cli->cl_lru_work) {
3594                 ptlrpcd_destroy_work(cli->cl_lru_work);
3595                 cli->cl_lru_work = NULL;
3596         }
3597
3598         obd_cleanup_client_import(obd);
3599         RETURN(0);
3600 }
3601 EXPORT_SYMBOL(osc_precleanup_common);
3602
3603 static int osc_precleanup(struct obd_device *obd)
3604 {
3605         ENTRY;
3606
3607         osc_precleanup_common(obd);
3608
3609         ptlrpc_lprocfs_unregister_obd(obd);
3610         RETURN(0);
3611 }
3612
3613 int osc_cleanup_common(struct obd_device *obd)
3614 {
3615         struct client_obd *cli = &obd->u.cli;
3616         int rc;
3617
3618         ENTRY;
3619
3620         spin_lock(&osc_shrink_lock);
3621         list_del(&cli->cl_shrink_list);
3622         spin_unlock(&osc_shrink_lock);
3623
3624         /* lru cleanup */
3625         if (cli->cl_cache != NULL) {
3626                 LASSERT(atomic_read(&cli->cl_cache->ccc_users) > 0);
3627                 spin_lock(&cli->cl_cache->ccc_lru_lock);
3628                 list_del_init(&cli->cl_lru_osc);
3629                 spin_unlock(&cli->cl_cache->ccc_lru_lock);
3630                 cli->cl_lru_left = NULL;
3631                 cl_cache_decref(cli->cl_cache);
3632                 cli->cl_cache = NULL;
3633         }
3634
3635         /* free memory of osc quota cache */
3636         osc_quota_cleanup(obd);
3637
3638         rc = client_obd_cleanup(obd);
3639
3640         ptlrpcd_decref();
3641         RETURN(rc);
3642 }
3643 EXPORT_SYMBOL(osc_cleanup_common);
3644
3645 static const struct obd_ops osc_obd_ops = {
3646         .o_owner                = THIS_MODULE,
3647         .o_setup                = osc_setup,
3648         .o_precleanup           = osc_precleanup,
3649         .o_cleanup              = osc_cleanup_common,
3650         .o_add_conn             = client_import_add_conn,
3651         .o_del_conn             = client_import_del_conn,
3652         .o_connect              = client_connect_import,
3653         .o_reconnect            = osc_reconnect,
3654         .o_disconnect           = osc_disconnect,
3655         .o_statfs               = osc_statfs,
3656         .o_statfs_async         = osc_statfs_async,
3657         .o_create               = osc_create,
3658         .o_destroy              = osc_destroy,
3659         .o_getattr              = osc_getattr,
3660         .o_setattr              = osc_setattr,
3661         .o_iocontrol            = osc_iocontrol,
3662         .o_set_info_async       = osc_set_info_async,
3663         .o_import_event         = osc_import_event,
3664         .o_quotactl             = osc_quotactl,
3665 };
3666
3667 LIST_HEAD(osc_shrink_list);
3668 DEFINE_SPINLOCK(osc_shrink_lock);
3669
3670 #ifdef HAVE_SHRINKER_COUNT
3671 static struct shrinker osc_cache_shrinker = {
3672         .count_objects  = osc_cache_shrink_count,
3673         .scan_objects   = osc_cache_shrink_scan,
3674         .seeks          = DEFAULT_SEEKS,
3675 };
3676 #else
3677 static int osc_cache_shrink(struct shrinker *shrinker,
3678                             struct shrink_control *sc)
3679 {
3680         (void)osc_cache_shrink_scan(shrinker, sc);
3681
3682         return osc_cache_shrink_count(shrinker, sc);
3683 }
3684
3685 static struct shrinker osc_cache_shrinker = {
3686         .shrink   = osc_cache_shrink,
3687         .seeks    = DEFAULT_SEEKS,
3688 };
3689 #endif
3690
3691 static int __init osc_init(void)
3692 {
3693         unsigned int reqpool_size;
3694         unsigned int reqsize;
3695         int rc;
3696         ENTRY;
3697
3698         /* print an address of _any_ initialized kernel symbol from this
3699          * module, to allow debugging with gdb that doesn't support data
3700          * symbols from modules.*/
3701         CDEBUG(D_INFO, "Lustre OSC module (%p).\n", &osc_caches);
3702
3703         rc = lu_kmem_init(osc_caches);
3704         if (rc)
3705                 RETURN(rc);
3706
3707         rc = class_register_type(&osc_obd_ops, NULL, true,
3708                                  LUSTRE_OSC_NAME, &osc_device_type);
3709         if (rc)
3710                 GOTO(out_kmem, rc);
3711
3712         rc = register_shrinker(&osc_cache_shrinker);
3713         if (rc)
3714                 GOTO(out_type, rc);
3715
3716         /* This is obviously too much memory, only prevent overflow here */
3717         if (osc_reqpool_mem_max >= 1 << 12 || osc_reqpool_mem_max == 0)
3718                 GOTO(out_shrinker, rc = -EINVAL);
3719
3720         reqpool_size = osc_reqpool_mem_max << 20;
3721
3722         reqsize = 1;
3723         while (reqsize < OST_IO_MAXREQSIZE)
3724                 reqsize = reqsize << 1;
3725
3726         /*
3727          * We don't enlarge the request count in OSC pool according to
3728          * cl_max_rpcs_in_flight. The allocation from the pool will only be
3729          * tried after normal allocation failed. So a small OSC pool won't
3730          * cause much performance degression in most of cases.
3731          */
3732         osc_reqpool_maxreqcount = reqpool_size / reqsize;
3733
3734         atomic_set(&osc_pool_req_count, 0);
3735         osc_rq_pool = ptlrpc_init_rq_pool(0, OST_IO_MAXREQSIZE,
3736                                           ptlrpc_add_rqs_to_pool);
3737
3738         if (osc_rq_pool == NULL)
3739                 GOTO(out_shrinker, rc = -ENOMEM);
3740
3741         rc = osc_start_grant_work();
3742         if (rc != 0)
3743                 GOTO(out_req_pool, rc);
3744
3745         RETURN(rc);
3746
3747 out_req_pool:
3748         ptlrpc_free_rq_pool(osc_rq_pool);
3749 out_shrinker:
3750         unregister_shrinker(&osc_cache_shrinker);
3751 out_type:
3752         class_unregister_type(LUSTRE_OSC_NAME);
3753 out_kmem:
3754         lu_kmem_fini(osc_caches);
3755
3756         RETURN(rc);
3757 }
3758
3759 static void __exit osc_exit(void)
3760 {
3761         osc_stop_grant_work();
3762         unregister_shrinker(&osc_cache_shrinker);
3763         class_unregister_type(LUSTRE_OSC_NAME);
3764         lu_kmem_fini(osc_caches);
3765         ptlrpc_free_rq_pool(osc_rq_pool);
3766 }
3767
3768 MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
3769 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC)");
3770 MODULE_VERSION(LUSTRE_VERSION_STRING);
3771 MODULE_LICENSE("GPL");
3772
3773 module_init(osc_init);
3774 module_exit(osc_exit);