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.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2013, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #define DEBUG_SUBSYSTEM S_OSC
  38
  39 #include <libcfs/libcfs.h>
  40
  41 #ifndef __KERNEL__
  42 # include <liblustre.h>
  43 #endif
  44
  45 #include <lustre_dlm.h>
  46 #include <lustre_net.h>
  47 #include <lustre/lustre_user.h>
  48 #include <obd_cksum.h>
  49
  50 #ifdef  __CYGWIN__
  51 # include <ctype.h>
  52 #endif
  53
  54 #include <lustre_ha.h>
  55 #include <lprocfs_status.h>
  56 #include <lustre_ioctl.h>
  57 #include <lustre_log.h>
  58 #include <lustre_debug.h>
  59 #include <lustre_param.h>
  60 #include <lustre_fid.h>
  61 #include "osc_internal.h"
  62 #include "osc_cl_internal.h"
  63
  64 struct osc_brw_async_args {
  65         struct obdo              *aa_oa;
  66         int                       aa_requested_nob;
  67         int                       aa_nio_count;
  68         obd_count                 aa_page_count;
  69         int                       aa_resends;
  70         struct brw_page **aa_ppga;
  71         struct client_obd        *aa_cli;
  72         struct list_head          aa_oaps;
  73         struct list_head          aa_exts;
  74         struct obd_capa  *aa_ocapa;
  75         struct cl_req            *aa_clerq;
  76 };
  77
  78 #define osc_grant_args osc_brw_async_args
  79
  80 struct osc_async_args {
  81         struct obd_info *aa_oi;
  82 };
  83
  84 struct osc_setattr_args {
  85         struct obdo             *sa_oa;
  86         obd_enqueue_update_f     sa_upcall;
  87         void                    *sa_cookie;
  88 };
  89
  90 struct osc_fsync_args {
  91         struct obd_info *fa_oi;
  92         obd_enqueue_update_f     fa_upcall;
  93         void                    *fa_cookie;
  94 };
  95
  96 struct osc_enqueue_args {
  97         struct obd_export               *oa_exp;
  98         __u64                           *oa_flags;
  99         obd_enqueue_update_f             oa_upcall;
 100         void                            *oa_cookie;
 101         struct ost_lvb                  *oa_lvb;
 102         struct lustre_handle            *oa_lockh;
 103         struct ldlm_enqueue_info        *oa_ei;
 104         unsigned int                     oa_agl:1;
 105 };
 106
 107 static void osc_release_ppga(struct brw_page **ppga, obd_count count);
 108 static int brw_interpret(const struct lu_env *env, struct ptlrpc_request *req,
 109                          void *data, int rc);
 110
 111 /* Unpack OSC object metadata from disk storage (LE byte order). */
 112 static int osc_unpackmd(struct obd_export *exp, struct lov_stripe_md **lsmp,
 113                         struct lov_mds_md *lmm, int lmm_bytes)
 114 {
 115         int lsm_size;
 116         struct obd_import *imp = class_exp2cliimp(exp);
 117         ENTRY;
 118
 119         if (lmm != NULL) {
 120                 if (lmm_bytes < sizeof(*lmm)) {
 121                         CERROR("%s: lov_mds_md too small: %d, need %d\n",
 122                                exp->exp_obd->obd_name, lmm_bytes,
 123                                (int)sizeof(*lmm));
 124                         RETURN(-EINVAL);
 125                 }
 126                 /* XXX LOV_MAGIC etc check? */
 127
 128                 if (unlikely(ostid_id(&lmm->lmm_oi) == 0)) {
 129                         CERROR("%s: zero lmm_object_id: rc = %d\n",
 130                                exp->exp_obd->obd_name, -EINVAL);
 131                         RETURN(-EINVAL);
 132                 }
 133         }
 134
 135         lsm_size = lov_stripe_md_size(1);
 136         if (lsmp == NULL)
 137                 RETURN(lsm_size);
 138
 139         if (*lsmp != NULL && lmm == NULL) {
 140                 OBD_FREE((*lsmp)->lsm_oinfo[0], sizeof(struct lov_oinfo));
 141                 OBD_FREE(*lsmp, lsm_size);
 142                 *lsmp = NULL;
 143                 RETURN(0);
 144         }
 145
 146         if (*lsmp == NULL) {
 147                 OBD_ALLOC(*lsmp, lsm_size);
 148                 if (unlikely(*lsmp == NULL))
 149                         RETURN(-ENOMEM);
 150                 OBD_ALLOC((*lsmp)->lsm_oinfo[0], sizeof(struct lov_oinfo));
 151                 if (unlikely((*lsmp)->lsm_oinfo[0] == NULL)) {
 152                         OBD_FREE(*lsmp, lsm_size);
 153                         RETURN(-ENOMEM);
 154                 }
 155                 loi_init((*lsmp)->lsm_oinfo[0]);
 156         } else if (unlikely(ostid_id(&(*lsmp)->lsm_oi) == 0)) {
 157                 RETURN(-EBADF);
 158         }
 159
 160         if (lmm != NULL)
 161                 /* XXX zero *lsmp? */
 162                 ostid_le_to_cpu(&lmm->lmm_oi, &(*lsmp)->lsm_oi);
 163
 164         if (imp != NULL &&
 165             (imp->imp_connect_data.ocd_connect_flags & OBD_CONNECT_MAXBYTES))
 166                 (*lsmp)->lsm_maxbytes = imp->imp_connect_data.ocd_maxbytes;
 167         else
 168                 (*lsmp)->lsm_maxbytes = LUSTRE_STRIPE_MAXBYTES;
 169
 170         RETURN(lsm_size);
 171 }
 172
 173 static inline void osc_pack_capa(struct ptlrpc_request *req,
 174                                  struct ost_body *body, void *capa)
 175 {
 176         struct obd_capa *oc = (struct obd_capa *)capa;
 177         struct lustre_capa *c;
 178
 179         if (!capa)
 180                 return;
 181
 182         c = req_capsule_client_get(&req->rq_pill, &RMF_CAPA1);
 183         LASSERT(c);
 184         capa_cpy(c, oc);
 185         body->oa.o_valid |= OBD_MD_FLOSSCAPA;
 186         DEBUG_CAPA(D_SEC, c, "pack");
 187 }
 188
 189 static inline void osc_pack_req_body(struct ptlrpc_request *req,
 190                                      struct obd_info *oinfo)
 191 {
 192         struct ost_body *body;
 193
 194         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 195         LASSERT(body);
 196
 197         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa,
 198                              oinfo->oi_oa);
 199         osc_pack_capa(req, body, oinfo->oi_capa);
 200 }
 201
 202 static inline void osc_set_capa_size(struct ptlrpc_request *req,
 203                                      const struct req_msg_field *field,
 204                                      struct obd_capa *oc)
 205 {
 206         if (oc == NULL)
 207                 req_capsule_set_size(&req->rq_pill, field, RCL_CLIENT, 0);
 208         else
 209                 /* it is already calculated as sizeof struct obd_capa */
 210                 ;
 211 }
 212
 213 static int osc_getattr_interpret(const struct lu_env *env,
 214                                  struct ptlrpc_request *req,
 215                                  struct osc_async_args *aa, int rc)
 216 {
 217         struct ost_body *body;
 218         ENTRY;
 219
 220         if (rc != 0)
 221                 GOTO(out, rc);
 222
 223         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 224         if (body) {
 225                 CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
 226                 lustre_get_wire_obdo(&req->rq_import->imp_connect_data,
 227                                      aa->aa_oi->oi_oa, &body->oa);
 228
 229                 /* This should really be sent by the OST */
 230                 aa->aa_oi->oi_oa->o_blksize = DT_MAX_BRW_SIZE;
 231                 aa->aa_oi->oi_oa->o_valid |= OBD_MD_FLBLKSZ;
 232         } else {
 233                 CDEBUG(D_INFO, "can't unpack ost_body\n");
 234                 rc = -EPROTO;
 235                 aa->aa_oi->oi_oa->o_valid = 0;
 236         }
 237 out:
 238         rc = aa->aa_oi->oi_cb_up(aa->aa_oi, rc);
 239         RETURN(rc);
 240 }
 241
 242 static int osc_getattr_async(struct obd_export *exp, struct obd_info *oinfo,
 243                              struct ptlrpc_request_set *set)
 244 {
 245         struct ptlrpc_request *req;
 246         struct osc_async_args *aa;
 247         int                    rc;
 248         ENTRY;
 249
 250         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_GETATTR);
 251         if (req == NULL)
 252                 RETURN(-ENOMEM);
 253
 254         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
 255         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GETATTR);
 256         if (rc) {
 257                 ptlrpc_request_free(req);
 258                 RETURN(rc);
 259         }
 260
 261         osc_pack_req_body(req, oinfo);
 262
 263         ptlrpc_request_set_replen(req);
 264         req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_getattr_interpret;
 265
 266         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
 267         aa = ptlrpc_req_async_args(req);
 268         aa->aa_oi = oinfo;
 269
 270         ptlrpc_set_add_req(set, req);
 271         RETURN(0);
 272 }
 273
 274 static int osc_getattr(const struct lu_env *env, struct obd_export *exp,
 275                        struct obd_info *oinfo)
 276 {
 277         struct ptlrpc_request *req;
 278         struct ost_body       *body;
 279         int                    rc;
 280         ENTRY;
 281
 282         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_GETATTR);
 283         if (req == NULL)
 284                 RETURN(-ENOMEM);
 285
 286         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
 287         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GETATTR);
 288         if (rc) {
 289                 ptlrpc_request_free(req);
 290                 RETURN(rc);
 291         }
 292
 293         osc_pack_req_body(req, oinfo);
 294
 295         ptlrpc_request_set_replen(req);
 296
 297         rc = ptlrpc_queue_wait(req);
 298         if (rc)
 299                 GOTO(out, rc);
 300
 301         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 302         if (body == NULL)
 303                 GOTO(out, rc = -EPROTO);
 304
 305         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
 306         lustre_get_wire_obdo(&req->rq_import->imp_connect_data, oinfo->oi_oa,
 307                              &body->oa);
 308
 309         oinfo->oi_oa->o_blksize = cli_brw_size(exp->exp_obd);
 310         oinfo->oi_oa->o_valid |= OBD_MD_FLBLKSZ;
 311
 312         EXIT;
 313  out:
 314         ptlrpc_req_finished(req);
 315         return rc;
 316 }
 317
 318 static int osc_setattr(const struct lu_env *env, struct obd_export *exp,
 319                        struct obd_info *oinfo, struct obd_trans_info *oti)
 320 {
 321         struct ptlrpc_request *req;
 322         struct ost_body       *body;
 323         int                    rc;
 324         ENTRY;
 325
 326         LASSERT(oinfo->oi_oa->o_valid & OBD_MD_FLGROUP);
 327
 328         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SETATTR);
 329         if (req == NULL)
 330                 RETURN(-ENOMEM);
 331
 332         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
 333         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SETATTR);
 334         if (rc) {
 335                 ptlrpc_request_free(req);
 336                 RETURN(rc);
 337         }
 338
 339         osc_pack_req_body(req, oinfo);
 340
 341         ptlrpc_request_set_replen(req);
 342
 343         rc = ptlrpc_queue_wait(req);
 344         if (rc)
 345                 GOTO(out, rc);
 346
 347         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 348         if (body == NULL)
 349                 GOTO(out, rc = -EPROTO);
 350
 351         lustre_get_wire_obdo(&req->rq_import->imp_connect_data, oinfo->oi_oa,
 352                              &body->oa);
 353
 354         EXIT;
 355 out:
 356         ptlrpc_req_finished(req);
 357         RETURN(rc);
 358 }
 359
 360 static int osc_setattr_interpret(const struct lu_env *env,
 361                                  struct ptlrpc_request *req,
 362                                  struct osc_setattr_args *sa, int rc)
 363 {
 364         struct ost_body *body;
 365         ENTRY;
 366
 367         if (rc != 0)
 368                 GOTO(out, rc);
 369
 370         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 371         if (body == NULL)
 372                 GOTO(out, rc = -EPROTO);
 373
 374         lustre_get_wire_obdo(&req->rq_import->imp_connect_data, sa->sa_oa,
 375                              &body->oa);
 376 out:
 377         rc = sa->sa_upcall(sa->sa_cookie, rc);
 378         RETURN(rc);
 379 }
 380
 381 int osc_setattr_async_base(struct obd_export *exp, struct obd_info *oinfo,
 382                            struct obd_trans_info *oti,
 383                            obd_enqueue_update_f upcall, void *cookie,
 384                            struct ptlrpc_request_set *rqset)
 385 {
 386         struct ptlrpc_request   *req;
 387         struct osc_setattr_args *sa;
 388         int                      rc;
 389         ENTRY;
 390
 391         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SETATTR);
 392         if (req == NULL)
 393                 RETURN(-ENOMEM);
 394
 395         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
 396         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SETATTR);
 397         if (rc) {
 398                 ptlrpc_request_free(req);
 399                 RETURN(rc);
 400         }
 401
 402         if (oti && oinfo->oi_oa->o_valid & OBD_MD_FLCOOKIE)
 403                 oinfo->oi_oa->o_lcookie = *oti->oti_logcookies;
 404
 405         osc_pack_req_body(req, oinfo);
 406
 407         ptlrpc_request_set_replen(req);
 408
 409         /* do mds to ost setattr asynchronously */
 410         if (!rqset) {
 411                 /* Do not wait for response. */
 412                 ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
 413         } else {
 414                 req->rq_interpret_reply =
 415                         (ptlrpc_interpterer_t)osc_setattr_interpret;
 416
 417                 CLASSERT (sizeof(*sa) <= sizeof(req->rq_async_args));
 418                 sa = ptlrpc_req_async_args(req);
 419                 sa->sa_oa = oinfo->oi_oa;
 420                 sa->sa_upcall = upcall;
 421                 sa->sa_cookie = cookie;
 422
 423                 if (rqset == PTLRPCD_SET)
 424                         ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
 425                 else
 426                         ptlrpc_set_add_req(rqset, req);
 427         }
 428
 429         RETURN(0);
 430 }
 431
 432 static int osc_setattr_async(struct obd_export *exp, struct obd_info *oinfo,
 433                              struct obd_trans_info *oti,
 434                              struct ptlrpc_request_set *rqset)
 435 {
 436         return osc_setattr_async_base(exp, oinfo, oti,
 437                                       oinfo->oi_cb_up, oinfo, rqset);
 438 }
 439
 440 int osc_real_create(struct obd_export *exp, struct obdo *oa,
 441                     struct lov_stripe_md **ea, struct obd_trans_info *oti)
 442 {
 443         struct ptlrpc_request *req;
 444         struct ost_body       *body;
 445         struct lov_stripe_md  *lsm;
 446         int                    rc;
 447         ENTRY;
 448
 449         LASSERT(oa);
 450         LASSERT(ea);
 451
 452         lsm = *ea;
 453         if (!lsm) {
 454                 rc = obd_alloc_memmd(exp, &lsm);
 455                 if (rc < 0)
 456                         RETURN(rc);
 457         }
 458
 459         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_CREATE);
 460         if (req == NULL)
 461                 GOTO(out, rc = -ENOMEM);
 462
 463         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_CREATE);
 464         if (rc) {
 465                 ptlrpc_request_free(req);
 466                 GOTO(out, rc);
 467         }
 468
 469         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 470         LASSERT(body);
 471
 472         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
 473
 474         ptlrpc_request_set_replen(req);
 475
 476         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
 477             oa->o_flags == OBD_FL_DELORPHAN) {
 478                 DEBUG_REQ(D_HA, req,
 479                           "delorphan from OST integration");
 480                 /* Don't resend the delorphan req */
 481                 req->rq_no_resend = req->rq_no_delay = 1;
 482         }
 483
 484         rc = ptlrpc_queue_wait(req);
 485         if (rc)
 486                 GOTO(out_req, rc);
 487
 488         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 489         if (body == NULL)
 490                 GOTO(out_req, rc = -EPROTO);
 491
 492         CDEBUG(D_INFO, "oa flags %x\n", oa->o_flags);
 493         lustre_get_wire_obdo(&req->rq_import->imp_connect_data, oa, &body->oa);
 494
 495         oa->o_blksize = cli_brw_size(exp->exp_obd);
 496         oa->o_valid |= OBD_MD_FLBLKSZ;
 497
 498         /* XXX LOV STACKING: the lsm that is passed to us from LOV does not
 499          * have valid lsm_oinfo data structs, so don't go touching that.
 500          * This needs to be fixed in a big way.
 501          */
 502         lsm->lsm_oi = oa->o_oi;
 503         *ea = lsm;
 504
 505         if (oti != NULL) {
 506                 if (oa->o_valid & OBD_MD_FLCOOKIE) {
 507                         if (oti->oti_logcookies == NULL)
 508                                 oti->oti_logcookies = &oti->oti_onecookie;
 509
 510                         *oti->oti_logcookies = oa->o_lcookie;
 511                 }
 512         }
 513
 514         CDEBUG(D_HA, "transno: "LPD64"\n",
 515                lustre_msg_get_transno(req->rq_repmsg));
 516 out_req:
 517         ptlrpc_req_finished(req);
 518 out:
 519         if (rc && !*ea)
 520                 obd_free_memmd(exp, &lsm);
 521         RETURN(rc);
 522 }
 523
 524 int osc_punch_base(struct obd_export *exp, struct obd_info *oinfo,
 525                    obd_enqueue_update_f upcall, void *cookie,
 526                    struct ptlrpc_request_set *rqset)
 527 {
 528         struct ptlrpc_request   *req;
 529         struct osc_setattr_args *sa;
 530         struct ost_body         *body;
 531         int                      rc;
 532         ENTRY;
 533
 534         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_PUNCH);
 535         if (req == NULL)
 536                 RETURN(-ENOMEM);
 537
 538         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
 539         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_PUNCH);
 540         if (rc) {
 541                 ptlrpc_request_free(req);
 542                 RETURN(rc);
 543         }
 544         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
 545         ptlrpc_at_set_req_timeout(req);
 546
 547         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 548         LASSERT(body);
 549         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa,
 550                              oinfo->oi_oa);
 551         osc_pack_capa(req, body, oinfo->oi_capa);
 552
 553         ptlrpc_request_set_replen(req);
 554
 555         req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_setattr_interpret;
 556         CLASSERT (sizeof(*sa) <= sizeof(req->rq_async_args));
 557         sa = ptlrpc_req_async_args(req);
 558         sa->sa_oa     = oinfo->oi_oa;
 559         sa->sa_upcall = upcall;
 560         sa->sa_cookie = cookie;
 561         if (rqset == PTLRPCD_SET)
 562                 ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
 563         else
 564                 ptlrpc_set_add_req(rqset, req);
 565
 566         RETURN(0);
 567 }
 568
 569 static int osc_sync_interpret(const struct lu_env *env,
 570                               struct ptlrpc_request *req,
 571                               void *arg, int rc)
 572 {
 573         struct osc_fsync_args *fa = arg;
 574         struct ost_body *body;
 575         ENTRY;
 576
 577         if (rc)
 578                 GOTO(out, rc);
 579
 580         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 581         if (body == NULL) {
 582                 CERROR ("can't unpack ost_body\n");
 583                 GOTO(out, rc = -EPROTO);
 584         }
 585
 586         *fa->fa_oi->oi_oa = body->oa;
 587 out:
 588         rc = fa->fa_upcall(fa->fa_cookie, rc);
 589         RETURN(rc);
 590 }
 591
 592 int osc_sync_base(struct obd_export *exp, struct obd_info *oinfo,
 593                   obd_enqueue_update_f upcall, void *cookie,
 594                   struct ptlrpc_request_set *rqset)
 595 {
 596         struct ptlrpc_request *req;
 597         struct ost_body       *body;
 598         struct osc_fsync_args *fa;
 599         int                    rc;
 600         ENTRY;
 601
 602         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SYNC);
 603         if (req == NULL)
 604                 RETURN(-ENOMEM);
 605
 606         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
 607         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SYNC);
 608         if (rc) {
 609                 ptlrpc_request_free(req);
 610                 RETURN(rc);
 611         }
 612
 613         /* overload the size and blocks fields in the oa with start/end */
 614         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 615         LASSERT(body);
 616         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa,
 617                              oinfo->oi_oa);
 618         osc_pack_capa(req, body, oinfo->oi_capa);
 619
 620         ptlrpc_request_set_replen(req);
 621         req->rq_interpret_reply = osc_sync_interpret;
 622
 623         CLASSERT(sizeof(*fa) <= sizeof(req->rq_async_args));
 624         fa = ptlrpc_req_async_args(req);
 625         fa->fa_oi = oinfo;
 626         fa->fa_upcall = upcall;
 627         fa->fa_cookie = cookie;
 628
 629         if (rqset == PTLRPCD_SET)
 630                 ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
 631         else
 632                 ptlrpc_set_add_req(rqset, req);
 633
 634         RETURN (0);
 635 }
 636
 637 /* Find and cancel locally locks matched by @mode in the resource found by
 638  * @objid. Found locks are added into @cancel list. Returns the amount of
 639  * locks added to @cancels list. */
 640 static int osc_resource_get_unused(struct obd_export *exp, struct obdo *oa,
 641                                    struct list_head *cancels,
 642                                    ldlm_mode_t mode, __u64 lock_flags)
 643 {
 644         struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
 645         struct ldlm_res_id res_id;
 646         struct ldlm_resource *res;
 647         int count;
 648         ENTRY;
 649
 650         /* Return, i.e. cancel nothing, only if ELC is supported (flag in
 651          * export) but disabled through procfs (flag in NS).
 652          *
 653          * This distinguishes from a case when ELC is not supported originally,
 654          * when we still want to cancel locks in advance and just cancel them
 655          * locally, without sending any RPC. */
 656         if (exp_connect_cancelset(exp) && !ns_connect_cancelset(ns))
 657                 RETURN(0);
 658
 659         ostid_build_res_name(&oa->o_oi, &res_id);
 660         res = ldlm_resource_get(ns, NULL, &res_id, 0, 0);
 661         if (IS_ERR(res))
 662                 RETURN(0);
 663
 664         LDLM_RESOURCE_ADDREF(res);
 665         count = ldlm_cancel_resource_local(res, cancels, NULL, mode,
 666                                            lock_flags, 0, NULL);
 667         LDLM_RESOURCE_DELREF(res);
 668         ldlm_resource_putref(res);
 669         RETURN(count);
 670 }
 671
 672 static int osc_destroy_interpret(const struct lu_env *env,
 673                                  struct ptlrpc_request *req, void *data,
 674                                  int rc)
 675 {
 676         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
 677
 678         atomic_dec(&cli->cl_destroy_in_flight);
 679         wake_up(&cli->cl_destroy_waitq);
 680         return 0;
 681 }
 682
 683 static int osc_can_send_destroy(struct client_obd *cli)
 684 {
 685         if (atomic_inc_return(&cli->cl_destroy_in_flight) <=
 686             cli->cl_max_rpcs_in_flight) {
 687                 /* The destroy request can be sent */
 688                 return 1;
 689         }
 690         if (atomic_dec_return(&cli->cl_destroy_in_flight) <
 691             cli->cl_max_rpcs_in_flight) {
 692                 /*
 693                  * The counter has been modified between the two atomic
 694                  * operations.
 695                  */
 696                 wake_up(&cli->cl_destroy_waitq);
 697         }
 698         return 0;
 699 }
 700
 701 int osc_create(const struct lu_env *env, struct obd_export *exp,
 702                struct obdo *oa, struct lov_stripe_md **ea,
 703                struct obd_trans_info *oti)
 704 {
 705         int rc = 0;
 706         ENTRY;
 707
 708         LASSERT(oa);
 709         LASSERT(ea);
 710         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
 711
 712         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
 713             oa->o_flags == OBD_FL_RECREATE_OBJS) {
 714                 RETURN(osc_real_create(exp, oa, ea, oti));
 715         }
 716
 717         if (!fid_seq_is_mdt(ostid_seq(&oa->o_oi)))
 718                 RETURN(osc_real_create(exp, oa, ea, oti));
 719
 720         /* we should not get here anymore */
 721         LBUG();
 722
 723         RETURN(rc);
 724 }
 725
 726 /* Destroy requests can be async always on the client, and we don't even really
 727  * care about the return code since the client cannot do anything at all about
 728  * a destroy failure.
 729  * When the MDS is unlinking a filename, it saves the file objects into a
 730  * recovery llog, and these object records are cancelled when the OST reports
 731  * they were destroyed and sync'd to disk (i.e. transaction committed).
 732  * If the client dies, or the OST is down when the object should be destroyed,
 733  * the records are not cancelled, and when the OST reconnects to the MDS next,
 734  * it will retrieve the llog unlink logs and then sends the log cancellation
 735  * cookies to the MDS after committing destroy transactions. */
 736 static int osc_destroy(const struct lu_env *env, struct obd_export *exp,
 737                        struct obdo *oa, struct lov_stripe_md *ea,
 738                        struct obd_trans_info *oti, struct obd_export *md_export,
 739                        void *capa)
 740 {
 741         struct client_obd     *cli = &exp->exp_obd->u.cli;
 742         struct ptlrpc_request *req;
 743         struct ost_body       *body;
 744         struct list_head       cancels = LIST_HEAD_INIT(cancels);
 745         int rc, count;
 746         ENTRY;
 747
 748         if (!oa) {
 749                 CDEBUG(D_INFO, "oa NULL\n");
 750                 RETURN(-EINVAL);
 751         }
 752
 753         count = osc_resource_get_unused(exp, oa, &cancels, LCK_PW,
 754                                         LDLM_FL_DISCARD_DATA);
 755
 756         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_DESTROY);
 757         if (req == NULL) {
 758                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 759                 RETURN(-ENOMEM);
 760         }
 761
 762         osc_set_capa_size(req, &RMF_CAPA1, (struct obd_capa *)capa);
 763         rc = ldlm_prep_elc_req(exp, req, LUSTRE_OST_VERSION, OST_DESTROY,
 764                                0, &cancels, count);
 765         if (rc) {
 766                 ptlrpc_request_free(req);
 767                 RETURN(rc);
 768         }
 769
 770         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
 771         ptlrpc_at_set_req_timeout(req);
 772
 773         if (oti != NULL && oa->o_valid & OBD_MD_FLCOOKIE)
 774                 oa->o_lcookie = *oti->oti_logcookies;
 775         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 776         LASSERT(body);
 777         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
 778
 779         osc_pack_capa(req, body, (struct obd_capa *)capa);
 780         ptlrpc_request_set_replen(req);
 781
 782         /* If osc_destory is for destroying the unlink orphan,
 783          * sent from MDT to OST, which should not be blocked here,
 784          * because the process might be triggered by ptlrpcd, and
 785          * it is not good to block ptlrpcd thread (b=16006)*/
 786         if (!(oa->o_flags & OBD_FL_DELORPHAN)) {
 787                 req->rq_interpret_reply = osc_destroy_interpret;
 788                 if (!osc_can_send_destroy(cli)) {
 789                         struct l_wait_info lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP,
 790                                                           NULL);
 791
 792                         /*
 793                          * Wait until the number of on-going destroy RPCs drops
 794                          * under max_rpc_in_flight
 795                          */
 796                         l_wait_event_exclusive(cli->cl_destroy_waitq,
 797                                                osc_can_send_destroy(cli), &lwi);
 798                 }
 799         }
 800
 801         /* Do not wait for response */
 802         ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
 803         RETURN(0);
 804 }
 805
 806 static void osc_announce_cached(struct client_obd *cli, struct obdo *oa,
 807                                 long writing_bytes)
 808 {
 809         obd_flag bits = OBD_MD_FLBLOCKS|OBD_MD_FLGRANT;
 810
 811         LASSERT(!(oa->o_valid & bits));
 812
 813         oa->o_valid |= bits;
 814         client_obd_list_lock(&cli->cl_loi_list_lock);
 815         oa->o_dirty = cli->cl_dirty_pages << PAGE_CACHE_SHIFT;
 816         if (unlikely(cli->cl_dirty_pages - cli->cl_dirty_transit >
 817                      cli->cl_dirty_max_pages)) {
 818                 CERROR("dirty %lu - %lu > dirty_max %lu\n",
 819                        cli->cl_dirty_pages, cli->cl_dirty_transit,
 820                        cli->cl_dirty_max_pages);
 821                 oa->o_undirty = 0;
 822         } else if (unlikely(atomic_read(&obd_dirty_pages) -
 823                             atomic_read(&obd_dirty_transit_pages) >
 824                             (long)(obd_max_dirty_pages + 1))) {
 825                 /* The atomic_read() allowing the atomic_inc() are
 826                  * not covered by a lock thus they may safely race and trip
 827                  * this CERROR() unless we add in a small fudge factor (+1). */
 828                 CERROR("%s: dirty %d - %d > system dirty_max %d\n",
 829                        cli->cl_import->imp_obd->obd_name,
 830                        atomic_read(&obd_dirty_pages),
 831                        atomic_read(&obd_dirty_transit_pages),
 832                        obd_max_dirty_pages);
 833                 oa->o_undirty = 0;
 834         } else if (unlikely(cli->cl_dirty_max_pages - cli->cl_dirty_pages >
 835                             0x7fffffff)) {
 836                 CERROR("dirty %lu - dirty_max %lu too big???\n",
 837                        cli->cl_dirty_pages, cli->cl_dirty_max_pages);
 838                 oa->o_undirty = 0;
 839         } else {
 840                 long max_in_flight = (cli->cl_max_pages_per_rpc <<
 841                                       PAGE_CACHE_SHIFT) *
 842                                      (cli->cl_max_rpcs_in_flight + 1);
 843                 oa->o_undirty = max(cli->cl_dirty_max_pages << PAGE_CACHE_SHIFT,
 844                                     max_in_flight);
 845         }
 846         oa->o_grant = cli->cl_avail_grant + cli->cl_reserved_grant;
 847         oa->o_dropped = cli->cl_lost_grant;
 848         cli->cl_lost_grant = 0;
 849         client_obd_list_unlock(&cli->cl_loi_list_lock);
 850         CDEBUG(D_CACHE,"dirty: "LPU64" undirty: %u dropped %u grant: "LPU64"\n",
 851                oa->o_dirty, oa->o_undirty, oa->o_dropped, oa->o_grant);
 852
 853 }
 854
 855 void osc_update_next_shrink(struct client_obd *cli)
 856 {
 857         cli->cl_next_shrink_grant =
 858                 cfs_time_shift(cli->cl_grant_shrink_interval);
 859         CDEBUG(D_CACHE, "next time %ld to shrink grant \n",
 860                cli->cl_next_shrink_grant);
 861 }
 862
 863 static void __osc_update_grant(struct client_obd *cli, obd_size grant)
 864 {
 865         client_obd_list_lock(&cli->cl_loi_list_lock);
 866         cli->cl_avail_grant += grant;
 867         client_obd_list_unlock(&cli->cl_loi_list_lock);
 868 }
 869
 870 static void osc_update_grant(struct client_obd *cli, struct ost_body *body)
 871 {
 872         if (body->oa.o_valid & OBD_MD_FLGRANT) {
 873                 CDEBUG(D_CACHE, "got "LPU64" extra grant\n", body->oa.o_grant);
 874                 __osc_update_grant(cli, body->oa.o_grant);
 875         }
 876 }
 877
 878 static int osc_set_info_async(const struct lu_env *env, struct obd_export *exp,
 879                               obd_count keylen, void *key, obd_count vallen,
 880                               void *val, struct ptlrpc_request_set *set);
 881
 882 static int osc_shrink_grant_interpret(const struct lu_env *env,
 883                                       struct ptlrpc_request *req,
 884                                       void *aa, int rc)
 885 {
 886         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
 887         struct obdo *oa = ((struct osc_grant_args *)aa)->aa_oa;
 888         struct ost_body *body;
 889
 890         if (rc != 0) {
 891                 __osc_update_grant(cli, oa->o_grant);
 892                 GOTO(out, rc);
 893         }
 894
 895         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 896         LASSERT(body);
 897         osc_update_grant(cli, body);
 898 out:
 899         OBDO_FREE(oa);
 900         return rc;
 901 }
 902
 903 static void osc_shrink_grant_local(struct client_obd *cli, struct obdo *oa)
 904 {
 905         client_obd_list_lock(&cli->cl_loi_list_lock);
 906         oa->o_grant = cli->cl_avail_grant / 4;
 907         cli->cl_avail_grant -= oa->o_grant;
 908         client_obd_list_unlock(&cli->cl_loi_list_lock);
 909         if (!(oa->o_valid & OBD_MD_FLFLAGS)) {
 910                 oa->o_valid |= OBD_MD_FLFLAGS;
 911                 oa->o_flags = 0;
 912         }
 913         oa->o_flags |= OBD_FL_SHRINK_GRANT;
 914         osc_update_next_shrink(cli);
 915 }
 916
 917 /* Shrink the current grant, either from some large amount to enough for a
 918  * full set of in-flight RPCs, or if we have already shrunk to that limit
 919  * then to enough for a single RPC.  This avoids keeping more grant than
 920  * needed, and avoids shrinking the grant piecemeal. */
 921 static int osc_shrink_grant(struct client_obd *cli)
 922 {
 923         __u64 target_bytes = (cli->cl_max_rpcs_in_flight + 1) *
 924                              (cli->cl_max_pages_per_rpc << PAGE_CACHE_SHIFT);
 925
 926         client_obd_list_lock(&cli->cl_loi_list_lock);
 927         if (cli->cl_avail_grant <= target_bytes)
 928                 target_bytes = cli->cl_max_pages_per_rpc << PAGE_CACHE_SHIFT;
 929         client_obd_list_unlock(&cli->cl_loi_list_lock);
 930
 931         return osc_shrink_grant_to_target(cli, target_bytes);
 932 }
 933
 934 int osc_shrink_grant_to_target(struct client_obd *cli, __u64 target_bytes)
 935 {
 936         int                     rc = 0;
 937         struct ost_body        *body;
 938         ENTRY;
 939
 940         client_obd_list_lock(&cli->cl_loi_list_lock);
 941         /* Don't shrink if we are already above or below the desired limit
 942          * We don't want to shrink below a single RPC, as that will negatively
 943          * impact block allocation and long-term performance. */
 944         if (target_bytes < cli->cl_max_pages_per_rpc << PAGE_CACHE_SHIFT)
 945                 target_bytes = cli->cl_max_pages_per_rpc << PAGE_CACHE_SHIFT;
 946
 947         if (target_bytes >= cli->cl_avail_grant) {
 948                 client_obd_list_unlock(&cli->cl_loi_list_lock);
 949                 RETURN(0);
 950         }
 951         client_obd_list_unlock(&cli->cl_loi_list_lock);
 952
 953         OBD_ALLOC_PTR(body);
 954         if (!body)
 955                 RETURN(-ENOMEM);
 956
 957         osc_announce_cached(cli, &body->oa, 0);
 958
 959         client_obd_list_lock(&cli->cl_loi_list_lock);
 960         body->oa.o_grant = cli->cl_avail_grant - target_bytes;
 961         cli->cl_avail_grant = target_bytes;
 962         client_obd_list_unlock(&cli->cl_loi_list_lock);
 963         if (!(body->oa.o_valid & OBD_MD_FLFLAGS)) {
 964                 body->oa.o_valid |= OBD_MD_FLFLAGS;
 965                 body->oa.o_flags = 0;
 966         }
 967         body->oa.o_flags |= OBD_FL_SHRINK_GRANT;
 968         osc_update_next_shrink(cli);
 969
 970         rc = osc_set_info_async(NULL, cli->cl_import->imp_obd->obd_self_export,
 971                                 sizeof(KEY_GRANT_SHRINK), KEY_GRANT_SHRINK,
 972                                 sizeof(*body), body, NULL);
 973         if (rc != 0)
 974                 __osc_update_grant(cli, body->oa.o_grant);
 975         OBD_FREE_PTR(body);
 976         RETURN(rc);
 977 }
 978
 979 static int osc_should_shrink_grant(struct client_obd *client)
 980 {
 981         cfs_time_t time = cfs_time_current();
 982         cfs_time_t next_shrink = client->cl_next_shrink_grant;
 983
 984         if ((client->cl_import->imp_connect_data.ocd_connect_flags &
 985              OBD_CONNECT_GRANT_SHRINK) == 0)
 986                 return 0;
 987
 988         if (cfs_time_aftereq(time, next_shrink - 5 * CFS_TICK)) {
 989                 /* Get the current RPC size directly, instead of going via:
 990                  * cli_brw_size(obd->u.cli.cl_import->imp_obd->obd_self_export)
 991                  * Keep comment here so that it can be found by searching. */
 992                 int brw_size = client->cl_max_pages_per_rpc << PAGE_CACHE_SHIFT;
 993
 994                 if (client->cl_import->imp_state == LUSTRE_IMP_FULL &&
 995                     client->cl_avail_grant > brw_size)
 996                         return 1;
 997                 else
 998                         osc_update_next_shrink(client);
 999         }
1000         return 0;
1001 }
1002
1003 static int osc_grant_shrink_grant_cb(struct timeout_item *item, void *data)
1004 {
1005         struct client_obd *client;
1006
1007         list_for_each_entry(client, &item->ti_obd_list, cl_grant_shrink_list) {
1008                 if (osc_should_shrink_grant(client))
1009                         osc_shrink_grant(client);
1010         }
1011         return 0;
1012 }
1013
1014 static int osc_add_shrink_grant(struct client_obd *client)
1015 {
1016         int rc;
1017
1018         rc = ptlrpc_add_timeout_client(client->cl_grant_shrink_interval,
1019                                        TIMEOUT_GRANT,
1020                                        osc_grant_shrink_grant_cb, NULL,
1021                                        &client->cl_grant_shrink_list);
1022         if (rc) {
1023                 CERROR("add grant client %s error %d\n",
1024                         client->cl_import->imp_obd->obd_name, rc);
1025                 return rc;
1026         }
1027         CDEBUG(D_CACHE, "add grant client %s \n",
1028                client->cl_import->imp_obd->obd_name);
1029         osc_update_next_shrink(client);
1030         return 0;
1031 }
1032
1033 static int osc_del_shrink_grant(struct client_obd *client)
1034 {
1035         return ptlrpc_del_timeout_client(&client->cl_grant_shrink_list,
1036                                          TIMEOUT_GRANT);
1037 }
1038
1039 static void osc_init_grant(struct client_obd *cli, struct obd_connect_data *ocd)
1040 {
1041         /*
1042          * ocd_grant is the total grant amount we're expect to hold: if we've
1043          * been evicted, it's the new avail_grant amount, cl_dirty_pages will
1044          * drop to 0 as inflight RPCs fail out; otherwise, it's avail_grant +
1045          * dirty.
1046          *
1047          * race is tolerable here: if we're evicted, but imp_state already
1048          * left EVICTED state, then cl_dirty_pages must be 0 already.
1049          */
1050         client_obd_list_lock(&cli->cl_loi_list_lock);
1051         if (cli->cl_import->imp_state == LUSTRE_IMP_EVICTED)
1052                 cli->cl_avail_grant = ocd->ocd_grant;
1053         else
1054                 cli->cl_avail_grant = ocd->ocd_grant -
1055                                       (cli->cl_dirty_pages << PAGE_CACHE_SHIFT);
1056
1057         if (cli->cl_avail_grant < 0) {
1058                 CWARN("%s: available grant < 0: avail/ocd/dirty %ld/%u/%ld\n",
1059                       cli->cl_import->imp_obd->obd_name, cli->cl_avail_grant,
1060                       ocd->ocd_grant, cli->cl_dirty_pages << PAGE_CACHE_SHIFT);
1061                 /* workaround for servers which do not have the patch from
1062                  * LU-2679 */
1063                 cli->cl_avail_grant = ocd->ocd_grant;
1064         }
1065
1066         /* determine the appropriate chunk size used by osc_extent. */
1067         cli->cl_chunkbits = max_t(int, PAGE_CACHE_SHIFT, ocd->ocd_blocksize);
1068         client_obd_list_unlock(&cli->cl_loi_list_lock);
1069
1070         CDEBUG(D_CACHE, "%s, setting cl_avail_grant: %ld cl_lost_grant: %ld."
1071                 "chunk bits: %d.\n", cli->cl_import->imp_obd->obd_name,
1072                 cli->cl_avail_grant, cli->cl_lost_grant, cli->cl_chunkbits);
1073
1074         if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT_SHRINK &&
1075             list_empty(&cli->cl_grant_shrink_list))
1076                 osc_add_shrink_grant(cli);
1077 }
1078
1079 /* We assume that the reason this OSC got a short read is because it read
1080  * beyond the end of a stripe file; i.e. lustre is reading a sparse file
1081  * via the LOV, and it _knows_ it's reading inside the file, it's just that
1082  * this stripe never got written at or beyond this stripe offset yet. */
1083 static void handle_short_read(int nob_read, obd_count page_count,
1084                               struct brw_page **pga)
1085 {
1086         char *ptr;
1087         int i = 0;
1088
1089         /* skip bytes read OK */
1090         while (nob_read > 0) {
1091                 LASSERT (page_count > 0);
1092
1093                 if (pga[i]->count > nob_read) {
1094                         /* EOF inside this page */
1095                         ptr = kmap(pga[i]->pg) +
1096                                 (pga[i]->off & ~CFS_PAGE_MASK);
1097                         memset(ptr + nob_read, 0, pga[i]->count - nob_read);
1098                         kunmap(pga[i]->pg);
1099                         page_count--;
1100                         i++;
1101                         break;
1102                 }
1103
1104                 nob_read -= pga[i]->count;
1105                 page_count--;
1106                 i++;
1107         }
1108
1109         /* zero remaining pages */
1110         while (page_count-- > 0) {
1111                 ptr = kmap(pga[i]->pg) + (pga[i]->off & ~CFS_PAGE_MASK);
1112                 memset(ptr, 0, pga[i]->count);
1113                 kunmap(pga[i]->pg);
1114                 i++;
1115         }
1116 }
1117
1118 static int check_write_rcs(struct ptlrpc_request *req,
1119                            int requested_nob, int niocount,
1120                            obd_count page_count, struct brw_page **pga)
1121 {
1122         int     i;
1123         __u32   *remote_rcs;
1124
1125         remote_rcs = req_capsule_server_sized_get(&req->rq_pill, &RMF_RCS,
1126                                                   sizeof(*remote_rcs) *
1127                                                   niocount);
1128         if (remote_rcs == NULL) {
1129                 CDEBUG(D_INFO, "Missing/short RC vector on BRW_WRITE reply\n");
1130                 return(-EPROTO);
1131         }
1132
1133         /* return error if any niobuf was in error */
1134         for (i = 0; i < niocount; i++) {
1135                 if ((int)remote_rcs[i] < 0)
1136                         return(remote_rcs[i]);
1137
1138                 if (remote_rcs[i] != 0) {
1139                         CDEBUG(D_INFO, "rc[%d] invalid (%d) req %p\n",
1140                                 i, remote_rcs[i], req);
1141                         return(-EPROTO);
1142                 }
1143         }
1144
1145         if (req->rq_bulk->bd_nob_transferred != requested_nob) {
1146                 CERROR("Unexpected # bytes transferred: %d (requested %d)\n",
1147                        req->rq_bulk->bd_nob_transferred, requested_nob);
1148                 return(-EPROTO);
1149         }
1150
1151         return (0);
1152 }
1153
1154 static inline int can_merge_pages(struct brw_page *p1, struct brw_page *p2)
1155 {
1156         if (p1->flag != p2->flag) {
1157                 unsigned mask = ~(OBD_BRW_FROM_GRANT | OBD_BRW_NOCACHE |
1158                                   OBD_BRW_SYNC       | OBD_BRW_ASYNC   |
1159                                   OBD_BRW_NOQUOTA    | OBD_BRW_SOFT_SYNC);
1160
1161                 /* warn if we try to combine flags that we don't know to be
1162                  * safe to combine */
1163                 if (unlikely((p1->flag & mask) != (p2->flag & mask))) {
1164                         CWARN("Saw flags 0x%x and 0x%x in the same brw, please "
1165                               "report this at https://jira.hpdd.intel.com/\n",
1166                               p1->flag, p2->flag);
1167                 }
1168                 return 0;
1169         }
1170
1171         return (p1->off + p1->count == p2->off);
1172 }
1173
1174 static obd_count osc_checksum_bulk(int nob, obd_count pg_count,
1175                                    struct brw_page **pga, int opc,
1176                                    cksum_type_t cksum_type)
1177 {
1178         __u32                           cksum;
1179         int                             i = 0;
1180         struct cfs_crypto_hash_desc     *hdesc;
1181         unsigned int                    bufsize;
1182         int                             err;
1183         unsigned char                   cfs_alg = cksum_obd2cfs(cksum_type);
1184
1185         LASSERT(pg_count > 0);
1186
1187         hdesc = cfs_crypto_hash_init(cfs_alg, NULL, 0);
1188         if (IS_ERR(hdesc)) {
1189                 CERROR("Unable to initialize checksum hash %s\n",
1190                        cfs_crypto_hash_name(cfs_alg));
1191                 return PTR_ERR(hdesc);
1192         }
1193
1194         while (nob > 0 && pg_count > 0) {
1195                 int count = pga[i]->count > nob ? nob : pga[i]->count;
1196
1197                 /* corrupt the data before we compute the checksum, to
1198                  * simulate an OST->client data error */
1199                 if (i == 0 && opc == OST_READ &&
1200                     OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_RECEIVE)) {
1201                         unsigned char *ptr = kmap(pga[i]->pg);
1202                         int off = pga[i]->off & ~CFS_PAGE_MASK;
1203
1204                         memcpy(ptr + off, "bad1", min(4, nob));
1205                         kunmap(pga[i]->pg);
1206                 }
1207                 cfs_crypto_hash_update_page(hdesc, pga[i]->pg,
1208                                             pga[i]->off & ~CFS_PAGE_MASK,
1209                                             count);
1210                 LL_CDEBUG_PAGE(D_PAGE, pga[i]->pg, "off %d\n",
1211                                (int)(pga[i]->off & ~CFS_PAGE_MASK));
1212
1213                 nob -= pga[i]->count;
1214                 pg_count--;
1215                 i++;
1216         }
1217
1218         bufsize = sizeof(cksum);
1219         err = cfs_crypto_hash_final(hdesc, (unsigned char *)&cksum, &bufsize);
1220
1221         /* For sending we only compute the wrong checksum instead
1222          * of corrupting the data so it is still correct on a redo */
1223         if (opc == OST_WRITE && OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_SEND))
1224                 cksum++;
1225
1226         return cksum;
1227 }
1228
1229 static int osc_brw_prep_request(int cmd, struct client_obd *cli,struct obdo *oa,
1230                                 struct lov_stripe_md *lsm, obd_count page_count,
1231                                 struct brw_page **pga,
1232                                 struct ptlrpc_request **reqp,
1233                                 struct obd_capa *ocapa, int reserve,
1234                                 int resend)
1235 {
1236         struct ptlrpc_request   *req;
1237         struct ptlrpc_bulk_desc *desc;
1238         struct ost_body         *body;
1239         struct obd_ioobj        *ioobj;
1240         struct niobuf_remote    *niobuf;
1241         int niocount, i, requested_nob, opc, rc;
1242         struct osc_brw_async_args *aa;
1243         struct req_capsule      *pill;
1244         struct brw_page *pg_prev;
1245
1246         ENTRY;
1247         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_PREP_REQ))
1248                 RETURN(-ENOMEM); /* Recoverable */
1249         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_PREP_REQ2))
1250                 RETURN(-EINVAL); /* Fatal */
1251
1252         if ((cmd & OBD_BRW_WRITE) != 0) {
1253                 opc = OST_WRITE;
1254                 req = ptlrpc_request_alloc_pool(cli->cl_import,
1255                                                 cli->cl_import->imp_rq_pool,
1256                                                 &RQF_OST_BRW_WRITE);
1257         } else {
1258                 opc = OST_READ;
1259                 req = ptlrpc_request_alloc(cli->cl_import, &RQF_OST_BRW_READ);
1260         }
1261         if (req == NULL)
1262                 RETURN(-ENOMEM);
1263
1264         for (niocount = i = 1; i < page_count; i++) {
1265                 if (!can_merge_pages(pga[i - 1], pga[i]))
1266                         niocount++;
1267         }
1268
1269         pill = &req->rq_pill;
1270         req_capsule_set_size(pill, &RMF_OBD_IOOBJ, RCL_CLIENT,
1271                              sizeof(*ioobj));
1272         req_capsule_set_size(pill, &RMF_NIOBUF_REMOTE, RCL_CLIENT,
1273                              niocount * sizeof(*niobuf));
1274         osc_set_capa_size(req, &RMF_CAPA1, ocapa);
1275
1276         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, opc);
1277         if (rc) {
1278                 ptlrpc_request_free(req);
1279                 RETURN(rc);
1280         }
1281         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
1282         ptlrpc_at_set_req_timeout(req);
1283         /* ask ptlrpc not to resend on EINPROGRESS since BRWs have their own
1284          * retry logic */
1285         req->rq_no_retry_einprogress = 1;
1286
1287         desc = ptlrpc_prep_bulk_imp(req, page_count,
1288                 cli->cl_import->imp_connect_data.ocd_brw_size >> LNET_MTU_BITS,
1289                 opc == OST_WRITE ? BULK_GET_SOURCE : BULK_PUT_SINK,
1290                 OST_BULK_PORTAL);
1291
1292         if (desc == NULL)
1293                 GOTO(out, rc = -ENOMEM);
1294         /* NB request now owns desc and will free it when it gets freed */
1295
1296         body = req_capsule_client_get(pill, &RMF_OST_BODY);
1297         ioobj = req_capsule_client_get(pill, &RMF_OBD_IOOBJ);
1298         niobuf = req_capsule_client_get(pill, &RMF_NIOBUF_REMOTE);
1299         LASSERT(body != NULL && ioobj != NULL && niobuf != NULL);
1300
1301         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
1302
1303         obdo_to_ioobj(oa, ioobj);
1304         ioobj->ioo_bufcnt = niocount;
1305         /* The high bits of ioo_max_brw tells server _maximum_ number of bulks
1306          * that might be send for this request.  The actual number is decided
1307          * when the RPC is finally sent in ptlrpc_register_bulk(). It sends
1308          * "max - 1" for old client compatibility sending "0", and also so the
1309          * the actual maximum is a power-of-two number, not one less. LU-1431 */
1310         ioobj_max_brw_set(ioobj, desc->bd_md_max_brw);
1311         osc_pack_capa(req, body, ocapa);
1312         LASSERT(page_count > 0);
1313         pg_prev = pga[0];
1314         for (requested_nob = i = 0; i < page_count; i++, niobuf++) {
1315                 struct brw_page *pg = pga[i];
1316                 int poff = pg->off & ~CFS_PAGE_MASK;
1317
1318                 LASSERT(pg->count > 0);
1319                 /* make sure there is no gap in the middle of page array */
1320                 LASSERTF(page_count == 1 ||
1321                          (ergo(i == 0, poff + pg->count == PAGE_CACHE_SIZE) &&
1322                           ergo(i > 0 && i < page_count - 1,
1323                                poff == 0 && pg->count == PAGE_CACHE_SIZE)   &&
1324                           ergo(i == page_count - 1, poff == 0)),
1325                          "i: %d/%d pg: %p off: "LPU64", count: %u\n",
1326                          i, page_count, pg, pg->off, pg->count);
1327 #ifdef __linux__
1328                 LASSERTF(i == 0 || pg->off > pg_prev->off,
1329                          "i %d p_c %u pg %p [pri %lu ind %lu] off "LPU64
1330                          " prev_pg %p [pri %lu ind %lu] off "LPU64"\n",
1331                          i, page_count,
1332                          pg->pg, page_private(pg->pg), pg->pg->index, pg->off,
1333                          pg_prev->pg, page_private(pg_prev->pg),
1334                          pg_prev->pg->index, pg_prev->off);
1335 #else
1336                 LASSERTF(i == 0 || pg->off > pg_prev->off,
1337                          "i %d p_c %u\n", i, page_count);
1338 #endif
1339                 LASSERT((pga[0]->flag & OBD_BRW_SRVLOCK) ==
1340                         (pg->flag & OBD_BRW_SRVLOCK));
1341
1342                 ptlrpc_prep_bulk_page_pin(desc, pg->pg, poff, pg->count);
1343                 requested_nob += pg->count;
1344
1345                 if (i > 0 && can_merge_pages(pg_prev, pg)) {
1346                         niobuf--;
1347                         niobuf->rnb_len += pg->count;
1348                 } else {
1349                         niobuf->rnb_offset = pg->off;
1350                         niobuf->rnb_len    = pg->count;
1351                         niobuf->rnb_flags  = pg->flag;
1352                 }
1353                 pg_prev = pg;
1354         }
1355
1356         LASSERTF((void *)(niobuf - niocount) ==
1357                 req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE),
1358                 "want %p - real %p\n", req_capsule_client_get(&req->rq_pill,
1359                 &RMF_NIOBUF_REMOTE), (void *)(niobuf - niocount));
1360
1361         osc_announce_cached(cli, &body->oa, opc == OST_WRITE ? requested_nob:0);
1362         if (resend) {
1363                 if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0) {
1364                         body->oa.o_valid |= OBD_MD_FLFLAGS;
1365                         body->oa.o_flags = 0;
1366                 }
1367                 body->oa.o_flags |= OBD_FL_RECOV_RESEND;
1368         }
1369
1370         if (osc_should_shrink_grant(cli))
1371                 osc_shrink_grant_local(cli, &body->oa);
1372
1373         /* size[REQ_REC_OFF] still sizeof (*body) */
1374         if (opc == OST_WRITE) {
1375                 if (cli->cl_checksum &&
1376                     !sptlrpc_flavor_has_bulk(&req->rq_flvr)) {
1377                         /* store cl_cksum_type in a local variable since
1378                          * it can be changed via lprocfs */
1379                         cksum_type_t cksum_type = cli->cl_cksum_type;
1380
1381                         if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0) {
1382                                 oa->o_flags &= OBD_FL_LOCAL_MASK;
1383                                 body->oa.o_flags = 0;
1384                         }
1385                         body->oa.o_flags |= cksum_type_pack(cksum_type);
1386                         body->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1387                         body->oa.o_cksum = osc_checksum_bulk(requested_nob,
1388                                                              page_count, pga,
1389                                                              OST_WRITE,
1390                                                              cksum_type);
1391                         CDEBUG(D_PAGE, "checksum at write origin: %x\n",
1392                                body->oa.o_cksum);
1393                         /* save this in 'oa', too, for later checking */
1394                         oa->o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1395                         oa->o_flags |= cksum_type_pack(cksum_type);
1396                 } else {
1397                         /* clear out the checksum flag, in case this is a
1398                          * resend but cl_checksum is no longer set. b=11238 */
1399                         oa->o_valid &= ~OBD_MD_FLCKSUM;
1400                 }
1401                 oa->o_cksum = body->oa.o_cksum;
1402                 /* 1 RC per niobuf */
1403                 req_capsule_set_size(pill, &RMF_RCS, RCL_SERVER,
1404                                      sizeof(__u32) * niocount);
1405         } else {
1406                 if (cli->cl_checksum &&
1407                     !sptlrpc_flavor_has_bulk(&req->rq_flvr)) {
1408                         if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0)
1409                                 body->oa.o_flags = 0;
1410                         body->oa.o_flags |= cksum_type_pack(cli->cl_cksum_type);
1411                         body->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1412                 }
1413         }
1414         ptlrpc_request_set_replen(req);
1415
1416         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
1417         aa = ptlrpc_req_async_args(req);
1418         aa->aa_oa = oa;
1419         aa->aa_requested_nob = requested_nob;
1420         aa->aa_nio_count = niocount;
1421         aa->aa_page_count = page_count;
1422         aa->aa_resends = 0;
1423         aa->aa_ppga = pga;
1424         aa->aa_cli = cli;
1425         INIT_LIST_HEAD(&aa->aa_oaps);
1426         if (ocapa && reserve)
1427                 aa->aa_ocapa = capa_get(ocapa);
1428
1429         *reqp = req;
1430         RETURN(0);
1431
1432  out:
1433         ptlrpc_req_finished(req);
1434         RETURN(rc);
1435 }
1436
1437 static int check_write_checksum(struct obdo *oa, const lnet_process_id_t *peer,
1438                                 __u32 client_cksum, __u32 server_cksum, int nob,
1439                                 obd_count page_count, struct brw_page **pga,
1440                                 cksum_type_t client_cksum_type)
1441 {
1442         __u32 new_cksum;
1443         char *msg;
1444         cksum_type_t cksum_type;
1445
1446         if (server_cksum == client_cksum) {
1447                 CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
1448                 return 0;
1449         }
1450
1451         cksum_type = cksum_type_unpack(oa->o_valid & OBD_MD_FLFLAGS ?
1452                                        oa->o_flags : 0);
1453         new_cksum = osc_checksum_bulk(nob, page_count, pga, OST_WRITE,
1454                                       cksum_type);
1455
1456         if (cksum_type != client_cksum_type)
1457                 msg = "the server did not use the checksum type specified in "
1458                       "the original request - likely a protocol problem";
1459         else if (new_cksum == server_cksum)
1460                 msg = "changed on the client after we checksummed it - "
1461                       "likely false positive due to mmap IO (bug 11742)";
1462         else if (new_cksum == client_cksum)
1463                 msg = "changed in transit before arrival at OST";
1464         else
1465                 msg = "changed in transit AND doesn't match the original - "
1466                       "likely false positive due to mmap IO (bug 11742)";
1467
1468         LCONSOLE_ERROR_MSG(0x132, "BAD WRITE CHECKSUM: %s: from %s inode "DFID
1469                            " object "DOSTID" extent ["LPU64"-"LPU64"]\n",
1470                            msg, libcfs_nid2str(peer->nid),
1471                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_seq : (__u64)0,
1472                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_oid : 0,
1473                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_ver : 0,
1474                            POSTID(&oa->o_oi), pga[0]->off,
1475                            pga[page_count-1]->off + pga[page_count-1]->count - 1);
1476         CERROR("original client csum %x (type %x), server csum %x (type %x), "
1477                "client csum now %x\n", client_cksum, client_cksum_type,
1478                server_cksum, cksum_type, new_cksum);
1479         return 1;
1480 }
1481
1482 /* Note rc enters this function as number of bytes transferred */
1483 static int osc_brw_fini_request(struct ptlrpc_request *req, int rc)
1484 {
1485         struct osc_brw_async_args *aa = (void *)&req->rq_async_args;
1486         const lnet_process_id_t *peer =
1487                         &req->rq_import->imp_connection->c_peer;
1488         struct client_obd *cli = aa->aa_cli;
1489         struct ost_body *body;
1490         __u32 client_cksum = 0;
1491         ENTRY;
1492
1493         if (rc < 0 && rc != -EDQUOT) {
1494                 DEBUG_REQ(D_INFO, req, "Failed request with rc = %d\n", rc);
1495                 RETURN(rc);
1496         }
1497
1498         LASSERTF(req->rq_repmsg != NULL, "rc = %d\n", rc);
1499         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
1500         if (body == NULL) {
1501                 DEBUG_REQ(D_INFO, req, "Can't unpack body\n");
1502                 RETURN(-EPROTO);
1503         }
1504
1505         /* set/clear over quota flag for a uid/gid */
1506         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE &&
1507             body->oa.o_valid & (OBD_MD_FLUSRQUOTA | OBD_MD_FLGRPQUOTA)) {
1508                 unsigned int qid[MAXQUOTAS] = { body->oa.o_uid, body->oa.o_gid };
1509
1510                 CDEBUG(D_QUOTA, "setdq for [%u %u] with valid "LPX64", flags %x\n",
1511                        body->oa.o_uid, body->oa.o_gid, body->oa.o_valid,
1512                        body->oa.o_flags);
1513                 osc_quota_setdq(cli, qid, body->oa.o_valid, body->oa.o_flags);
1514         }
1515
1516         osc_update_grant(cli, body);
1517
1518         if (rc < 0)
1519                 RETURN(rc);
1520
1521         if (aa->aa_oa->o_valid & OBD_MD_FLCKSUM)
1522                 client_cksum = aa->aa_oa->o_cksum; /* save for later */
1523
1524         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE) {
1525                 if (rc > 0) {
1526                         CERROR("Unexpected +ve rc %d\n", rc);
1527                         RETURN(-EPROTO);
1528                 }
1529                 LASSERT(req->rq_bulk->bd_nob == aa->aa_requested_nob);
1530
1531                 if (sptlrpc_cli_unwrap_bulk_write(req, req->rq_bulk))
1532                         RETURN(-EAGAIN);
1533
1534                 if ((aa->aa_oa->o_valid & OBD_MD_FLCKSUM) && client_cksum &&
1535                     check_write_checksum(&body->oa, peer, client_cksum,
1536                                          body->oa.o_cksum, aa->aa_requested_nob,
1537                                          aa->aa_page_count, aa->aa_ppga,
1538                                          cksum_type_unpack(aa->aa_oa->o_flags)))
1539                         RETURN(-EAGAIN);
1540
1541                 rc = check_write_rcs(req, aa->aa_requested_nob,aa->aa_nio_count,
1542                                      aa->aa_page_count, aa->aa_ppga);
1543                 GOTO(out, rc);
1544         }
1545
1546         /* The rest of this function executes only for OST_READs */
1547
1548         /* if unwrap_bulk failed, return -EAGAIN to retry */
1549         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk, rc);
1550         if (rc < 0)
1551                 GOTO(out, rc = -EAGAIN);
1552
1553         if (rc > aa->aa_requested_nob) {
1554                 CERROR("Unexpected rc %d (%d requested)\n", rc,
1555                        aa->aa_requested_nob);
1556                 RETURN(-EPROTO);
1557         }
1558
1559         if (rc != req->rq_bulk->bd_nob_transferred) {
1560                 CERROR ("Unexpected rc %d (%d transferred)\n",
1561                         rc, req->rq_bulk->bd_nob_transferred);
1562                 return (-EPROTO);
1563         }
1564
1565         if (rc < aa->aa_requested_nob)
1566                 handle_short_read(rc, aa->aa_page_count, aa->aa_ppga);
1567
1568         if (body->oa.o_valid & OBD_MD_FLCKSUM) {
1569                 static int cksum_counter;
1570                 __u32      server_cksum = body->oa.o_cksum;
1571                 char      *via;
1572                 char      *router;
1573                 cksum_type_t cksum_type;
1574
1575                 cksum_type = cksum_type_unpack(body->oa.o_valid &OBD_MD_FLFLAGS?
1576                                                body->oa.o_flags : 0);
1577                 client_cksum = osc_checksum_bulk(rc, aa->aa_page_count,
1578                                                  aa->aa_ppga, OST_READ,
1579                                                  cksum_type);
1580
1581                 if (peer->nid == req->rq_bulk->bd_sender) {
1582                         via = router = "";
1583                 } else {
1584                         via = " via ";
1585                         router = libcfs_nid2str(req->rq_bulk->bd_sender);
1586                 }
1587
1588                 if (server_cksum != client_cksum) {
1589                         LCONSOLE_ERROR_MSG(0x133, "%s: BAD READ CHECKSUM: from "
1590                                            "%s%s%s inode "DFID" object "DOSTID
1591                                            " extent ["LPU64"-"LPU64"]\n",
1592                                            req->rq_import->imp_obd->obd_name,
1593                                            libcfs_nid2str(peer->nid),
1594                                            via, router,
1595                                            body->oa.o_valid & OBD_MD_FLFID ?
1596                                                 body->oa.o_parent_seq : (__u64)0,
1597                                            body->oa.o_valid & OBD_MD_FLFID ?
1598                                                 body->oa.o_parent_oid : 0,
1599                                            body->oa.o_valid & OBD_MD_FLFID ?
1600                                                 body->oa.o_parent_ver : 0,
1601                                            POSTID(&body->oa.o_oi),
1602                                            aa->aa_ppga[0]->off,
1603                                            aa->aa_ppga[aa->aa_page_count-1]->off +
1604                                            aa->aa_ppga[aa->aa_page_count-1]->count -
1605                                                                         1);
1606                         CERROR("client %x, server %x, cksum_type %x\n",
1607                                client_cksum, server_cksum, cksum_type);
1608                         cksum_counter = 0;
1609                         aa->aa_oa->o_cksum = client_cksum;
1610                         rc = -EAGAIN;
1611                 } else {
1612                         cksum_counter++;
1613                         CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
1614                         rc = 0;
1615                 }
1616         } else if (unlikely(client_cksum)) {
1617                 static int cksum_missed;
1618
1619                 cksum_missed++;
1620                 if ((cksum_missed & (-cksum_missed)) == cksum_missed)
1621                         CERROR("Checksum %u requested from %s but not sent\n",
1622                                cksum_missed, libcfs_nid2str(peer->nid));
1623         } else {
1624                 rc = 0;
1625         }
1626 out:
1627         if (rc >= 0)
1628                 lustre_get_wire_obdo(&req->rq_import->imp_connect_data,
1629                                      aa->aa_oa, &body->oa);
1630
1631         RETURN(rc);
1632 }
1633
1634 static int osc_brw_redo_request(struct ptlrpc_request *request,
1635                                 struct osc_brw_async_args *aa, int rc)
1636 {
1637         struct ptlrpc_request *new_req;
1638         struct osc_brw_async_args *new_aa;
1639         struct osc_async_page *oap;
1640         ENTRY;
1641
1642         DEBUG_REQ(rc == -EINPROGRESS ? D_RPCTRACE : D_ERROR, request,
1643                   "redo for recoverable error %d", rc);
1644
1645         rc = osc_brw_prep_request(lustre_msg_get_opc(request->rq_reqmsg) ==
1646                                         OST_WRITE ? OBD_BRW_WRITE :OBD_BRW_READ,
1647                                   aa->aa_cli, aa->aa_oa,
1648                                   NULL /* lsm unused by osc currently */,
1649                                   aa->aa_page_count, aa->aa_ppga,
1650                                   &new_req, aa->aa_ocapa, 0, 1);
1651         if (rc)
1652                 RETURN(rc);
1653
1654         list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
1655                 if (oap->oap_request != NULL) {
1656                         LASSERTF(request == oap->oap_request,
1657                                  "request %p != oap_request %p\n",
1658                                  request, oap->oap_request);
1659                         if (oap->oap_interrupted) {
1660                                 ptlrpc_req_finished(new_req);
1661                                 RETURN(-EINTR);
1662                         }
1663                 }
1664         }
1665         /* New request takes over pga and oaps from old request.
1666          * Note that copying a list_head doesn't work, need to move it... */
1667         aa->aa_resends++;
1668         new_req->rq_interpret_reply = request->rq_interpret_reply;
1669         new_req->rq_async_args = request->rq_async_args;
1670         new_req->rq_commit_cb = request->rq_commit_cb;
1671         /* cap resend delay to the current request timeout, this is similar to
1672          * what ptlrpc does (see after_reply()) */
1673         if (aa->aa_resends > new_req->rq_timeout)
1674                 new_req->rq_sent = cfs_time_current_sec() + new_req->rq_timeout;
1675         else
1676                 new_req->rq_sent = cfs_time_current_sec() + aa->aa_resends;
1677         new_req->rq_generation_set = 1;
1678         new_req->rq_import_generation = request->rq_import_generation;
1679
1680         new_aa = ptlrpc_req_async_args(new_req);
1681
1682         INIT_LIST_HEAD(&new_aa->aa_oaps);
1683         list_splice_init(&aa->aa_oaps, &new_aa->aa_oaps);
1684         INIT_LIST_HEAD(&new_aa->aa_exts);
1685         list_splice_init(&aa->aa_exts, &new_aa->aa_exts);
1686         new_aa->aa_resends = aa->aa_resends;
1687
1688         list_for_each_entry(oap, &new_aa->aa_oaps, oap_rpc_item) {
1689                 if (oap->oap_request) {
1690                         ptlrpc_req_finished(oap->oap_request);
1691                         oap->oap_request = ptlrpc_request_addref(new_req);
1692                 }
1693         }
1694
1695         new_aa->aa_ocapa = aa->aa_ocapa;
1696         aa->aa_ocapa = NULL;
1697
1698         /* XXX: This code will run into problem if we're going to support
1699          * to add a series of BRW RPCs into a self-defined ptlrpc_request_set
1700          * and wait for all of them to be finished. We should inherit request
1701          * set from old request. */
1702         ptlrpcd_add_req(new_req, PDL_POLICY_SAME, -1);
1703
1704         DEBUG_REQ(D_INFO, new_req, "new request");
1705         RETURN(0);
1706 }
1707
1708 /*
1709  * ugh, we want disk allocation on the target to happen in offset order.  we'll
1710  * follow sedgewicks advice and stick to the dead simple shellsort -- it'll do
1711  * fine for our small page arrays and doesn't require allocation.  its an
1712  * insertion sort that swaps elements that are strides apart, shrinking the
1713  * stride down until its '1' and the array is sorted.
1714  */
1715 static void sort_brw_pages(struct brw_page **array, int num)
1716 {
1717         int stride, i, j;
1718         struct brw_page *tmp;
1719
1720         if (num == 1)
1721                 return;
1722         for (stride = 1; stride < num ; stride = (stride * 3) + 1)
1723                 ;
1724
1725         do {
1726                 stride /= 3;
1727                 for (i = stride ; i < num ; i++) {
1728                         tmp = array[i];
1729                         j = i;
1730                         while (j >= stride && array[j - stride]->off > tmp->off) {
1731                                 array[j] = array[j - stride];
1732                                 j -= stride;
1733                         }
1734                         array[j] = tmp;
1735                 }
1736         } while (stride > 1);
1737 }
1738
1739 static void osc_release_ppga(struct brw_page **ppga, obd_count count)
1740 {
1741         LASSERT(ppga != NULL);
1742         OBD_FREE(ppga, sizeof(*ppga) * count);
1743 }
1744
1745 static int brw_interpret(const struct lu_env *env,
1746                          struct ptlrpc_request *req, void *data, int rc)
1747 {
1748         struct osc_brw_async_args *aa = data;
1749         struct osc_extent *ext;
1750         struct osc_extent *tmp;
1751         struct client_obd *cli = aa->aa_cli;
1752         ENTRY;
1753
1754         rc = osc_brw_fini_request(req, rc);
1755         CDEBUG(D_INODE, "request %p aa %p rc %d\n", req, aa, rc);
1756         /* When server return -EINPROGRESS, client should always retry
1757          * regardless of the number of times the bulk was resent already. */
1758         if (osc_recoverable_error(rc)) {
1759                 if (req->rq_import_generation !=
1760                     req->rq_import->imp_generation) {
1761                         CDEBUG(D_HA, "%s: resend cross eviction for object: "
1762                                ""DOSTID", rc = %d.\n",
1763                                req->rq_import->imp_obd->obd_name,
1764                                POSTID(&aa->aa_oa->o_oi), rc);
1765                 } else if (rc == -EINPROGRESS ||
1766                     client_should_resend(aa->aa_resends, aa->aa_cli)) {
1767                         rc = osc_brw_redo_request(req, aa, rc);
1768                 } else {
1769                         CERROR("%s: too many resent retries for object: "
1770                                ""LPU64":"LPU64", rc = %d.\n",
1771                                req->rq_import->imp_obd->obd_name,
1772                                POSTID(&aa->aa_oa->o_oi), rc);
1773                 }
1774
1775                 if (rc == 0)
1776                         RETURN(0);
1777                 else if (rc == -EAGAIN || rc == -EINPROGRESS)
1778                         rc = -EIO;
1779         }
1780
1781         if (aa->aa_ocapa) {
1782                 capa_put(aa->aa_ocapa);
1783                 aa->aa_ocapa = NULL;
1784         }
1785
1786         if (rc == 0) {
1787                 struct obdo *oa = aa->aa_oa;
1788                 struct cl_attr *attr = &osc_env_info(env)->oti_attr;
1789                 unsigned long valid = 0;
1790                 struct cl_object *obj;
1791                 struct osc_async_page *last;
1792
1793                 last = brw_page2oap(aa->aa_ppga[aa->aa_page_count - 1]);
1794                 obj = osc2cl(last->oap_obj);
1795
1796                 cl_object_attr_lock(obj);
1797                 if (oa->o_valid & OBD_MD_FLBLOCKS) {
1798                         attr->cat_blocks = oa->o_blocks;
1799                         valid |= CAT_BLOCKS;
1800                 }
1801                 if (oa->o_valid & OBD_MD_FLMTIME) {
1802                         attr->cat_mtime = oa->o_mtime;
1803                         valid |= CAT_MTIME;
1804                 }
1805                 if (oa->o_valid & OBD_MD_FLATIME) {
1806                         attr->cat_atime = oa->o_atime;
1807                         valid |= CAT_ATIME;
1808                 }
1809                 if (oa->o_valid & OBD_MD_FLCTIME) {
1810                         attr->cat_ctime = oa->o_ctime;
1811                         valid |= CAT_CTIME;
1812                 }
1813
1814                 if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE) {
1815                         struct lov_oinfo *loi = cl2osc(obj)->oo_oinfo;
1816                         loff_t last_off = last->oap_count + last->oap_obj_off +
1817                                 last->oap_page_off;
1818
1819                         /* Change file size if this is an out of quota or
1820                          * direct IO write and it extends the file size */
1821                         if (loi->loi_lvb.lvb_size < last_off) {
1822                                 attr->cat_size = last_off;
1823                                 valid |= CAT_SIZE;
1824                         }
1825                         /* Extend KMS if it's not a lockless write */
1826                         if (loi->loi_kms < last_off &&
1827                             oap2osc_page(last)->ops_srvlock == 0) {
1828                                 attr->cat_kms = last_off;
1829                                 valid |= CAT_KMS;
1830                         }
1831                 }
1832
1833                 if (valid != 0)
1834                         cl_object_attr_set(env, obj, attr, valid);
1835                 cl_object_attr_unlock(obj);
1836         }
1837         OBDO_FREE(aa->aa_oa);
1838
1839         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE && rc == 0)
1840                 osc_inc_unstable_pages(req);
1841
1842         list_for_each_entry_safe(ext, tmp, &aa->aa_exts, oe_link) {
1843                 list_del_init(&ext->oe_link);
1844                 osc_extent_finish(env, ext, 1, rc);
1845         }
1846         LASSERT(list_empty(&aa->aa_exts));
1847         LASSERT(list_empty(&aa->aa_oaps));
1848
1849         cl_req_completion(env, aa->aa_clerq, rc < 0 ? rc :
1850                           req->rq_bulk->bd_nob_transferred);
1851         osc_release_ppga(aa->aa_ppga, aa->aa_page_count);
1852         ptlrpc_lprocfs_brw(req, req->rq_bulk->bd_nob_transferred);
1853
1854         client_obd_list_lock(&cli->cl_loi_list_lock);
1855         /* We need to decrement before osc_ap_completion->osc_wake_cache_waiters
1856          * is called so we know whether to go to sync BRWs or wait for more
1857          * RPCs to complete */
1858         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE)
1859                 cli->cl_w_in_flight--;
1860         else
1861                 cli->cl_r_in_flight--;
1862         osc_wake_cache_waiters(cli);
1863         client_obd_list_unlock(&cli->cl_loi_list_lock);
1864
1865         osc_io_unplug(env, cli, NULL, PDL_POLICY_SAME);
1866         RETURN(rc);
1867 }
1868
1869 static void brw_commit(struct ptlrpc_request *req)
1870 {
1871         /* If osc_inc_unstable_pages (via osc_extent_finish) races with
1872          * this called via the rq_commit_cb, I need to ensure
1873          * osc_dec_unstable_pages is still called. Otherwise unstable
1874          * pages may be leaked. */
1875         spin_lock(&req->rq_lock);
1876         if (likely(req->rq_unstable)) {
1877                 req->rq_unstable = 0;
1878                 spin_unlock(&req->rq_lock);
1879
1880                 osc_dec_unstable_pages(req);
1881         } else {
1882                 req->rq_committed = 1;
1883                 spin_unlock(&req->rq_lock);
1884         }
1885 }
1886
1887 /**
1888  * Build an RPC by the list of extent @ext_list. The caller must ensure
1889  * that the total pages in this list are NOT over max pages per RPC.
1890  * Extents in the list must be in OES_RPC state.
1891  */
1892 int osc_build_rpc(const struct lu_env *env, struct client_obd *cli,
1893                   struct list_head *ext_list, int cmd, pdl_policy_t pol)
1894 {
1895         struct ptlrpc_request           *req = NULL;
1896         struct osc_extent               *ext;
1897         struct brw_page                 **pga = NULL;
1898         struct osc_brw_async_args       *aa = NULL;
1899         struct obdo                     *oa = NULL;
1900         struct osc_async_page           *oap;
1901         struct osc_async_page           *tmp;
1902         struct cl_req                   *clerq = NULL;
1903         enum cl_req_type                crt = (cmd & OBD_BRW_WRITE) ? CRT_WRITE :
1904                                                                       CRT_READ;
1905         struct cl_req_attr              *crattr = NULL;
1906         obd_off                         starting_offset = OBD_OBJECT_EOF;
1907         obd_off                         ending_offset = 0;
1908         int                             mpflag = 0;
1909         int                             mem_tight = 0;
1910         int                             page_count = 0;
1911         bool                            soft_sync = false;
1912         int                             i;
1913         int                             rc;
1914         struct list_head                rpc_list = LIST_HEAD_INIT(rpc_list);
1915
1916         ENTRY;
1917         LASSERT(!list_empty(ext_list));
1918
1919         /* add pages into rpc_list to build BRW rpc */
1920         list_for_each_entry(ext, ext_list, oe_link) {
1921                 LASSERT(ext->oe_state == OES_RPC);
1922                 mem_tight |= ext->oe_memalloc;
1923                 list_for_each_entry(oap, &ext->oe_pages, oap_pending_item) {
1924                         ++page_count;
1925                         list_add_tail(&oap->oap_rpc_item, &rpc_list);
1926                         if (starting_offset > oap->oap_obj_off)
1927                                 starting_offset = oap->oap_obj_off;
1928                         else
1929                                 LASSERT(oap->oap_page_off == 0);
1930                         if (ending_offset < oap->oap_obj_off + oap->oap_count)
1931                                 ending_offset = oap->oap_obj_off +
1932                                                 oap->oap_count;
1933                         else
1934                                 LASSERT(oap->oap_page_off + oap->oap_count ==
1935                                         PAGE_CACHE_SIZE);
1936                 }
1937         }
1938
1939         soft_sync = osc_over_unstable_soft_limit(cli);
1940         if (mem_tight)
1941                 mpflag = cfs_memory_pressure_get_and_set();
1942
1943         OBD_ALLOC(crattr, sizeof(*crattr));
1944         if (crattr == NULL)
1945                 GOTO(out, rc = -ENOMEM);
1946
1947         OBD_ALLOC(pga, sizeof(*pga) * page_count);
1948         if (pga == NULL)
1949                 GOTO(out, rc = -ENOMEM);
1950
1951         OBDO_ALLOC(oa);
1952         if (oa == NULL)
1953                 GOTO(out, rc = -ENOMEM);
1954
1955         i = 0;
1956         list_for_each_entry(oap, &rpc_list, oap_rpc_item) {
1957                 struct cl_page *page = oap2cl_page(oap);
1958                 if (clerq == NULL) {
1959                         clerq = cl_req_alloc(env, page, crt,
1960                                              1 /* only 1-object rpcs for now */);
1961                         if (IS_ERR(clerq))
1962                                 GOTO(out, rc = PTR_ERR(clerq));
1963                 }
1964                 if (mem_tight)
1965                         oap->oap_brw_flags |= OBD_BRW_MEMALLOC;
1966                 if (soft_sync)
1967                         oap->oap_brw_flags |= OBD_BRW_SOFT_SYNC;
1968                 pga[i] = &oap->oap_brw_page;
1969                 pga[i]->off = oap->oap_obj_off + oap->oap_page_off;
1970                 CDEBUG(0, "put page %p index %lu oap %p flg %x to pga\n",
1971                        pga[i]->pg, page_index(oap->oap_page), oap,
1972                        pga[i]->flag);
1973                 i++;
1974                 cl_req_page_add(env, clerq, page);
1975         }
1976
1977         /* always get the data for the obdo for the rpc */
1978         LASSERT(clerq != NULL);
1979         crattr->cra_oa = oa;
1980         cl_req_attr_set(env, clerq, crattr, ~0ULL);
1981
1982         rc = cl_req_prep(env, clerq);
1983         if (rc != 0) {
1984                 CERROR("cl_req_prep failed: %d\n", rc);
1985                 GOTO(out, rc);
1986         }
1987
1988         sort_brw_pages(pga, page_count);
1989         rc = osc_brw_prep_request(cmd, cli, oa, NULL, page_count,
1990                         pga, &req, crattr->cra_capa, 1, 0);
1991         if (rc != 0) {
1992                 CERROR("prep_req failed: %d\n", rc);
1993                 GOTO(out, rc);
1994         }
1995
1996         req->rq_commit_cb = brw_commit;
1997         req->rq_interpret_reply = brw_interpret;
1998
1999         if (mem_tight != 0)
2000                 req->rq_memalloc = 1;
2001
2002         /* Need to update the timestamps after the request is built in case
2003          * we race with setattr (locally or in queue at OST).  If OST gets
2004          * later setattr before earlier BRW (as determined by the request xid),
2005          * the OST will not use BRW timestamps.  Sadly, there is no obvious
2006          * way to do this in a single call.  bug 10150 */
2007         cl_req_attr_set(env, clerq, crattr,
2008                         OBD_MD_FLMTIME|OBD_MD_FLCTIME|OBD_MD_FLATIME);
2009
2010         lustre_msg_set_jobid(req->rq_reqmsg, crattr->cra_jobid);
2011
2012         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
2013         aa = ptlrpc_req_async_args(req);
2014         INIT_LIST_HEAD(&aa->aa_oaps);
2015         list_splice_init(&rpc_list, &aa->aa_oaps);
2016         INIT_LIST_HEAD(&aa->aa_exts);
2017         list_splice_init(ext_list, &aa->aa_exts);
2018         aa->aa_clerq = clerq;
2019
2020         /* queued sync pages can be torn down while the pages
2021          * were between the pending list and the rpc */
2022         tmp = NULL;
2023         list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
2024                 /* only one oap gets a request reference */
2025                 if (tmp == NULL)
2026                         tmp = oap;
2027                 if (oap->oap_interrupted && !req->rq_intr) {
2028                         CDEBUG(D_INODE, "oap %p in req %p interrupted\n",
2029                                         oap, req);
2030                         ptlrpc_mark_interrupted(req);
2031                 }
2032         }
2033         if (tmp != NULL)
2034                 tmp->oap_request = ptlrpc_request_addref(req);
2035
2036         client_obd_list_lock(&cli->cl_loi_list_lock);
2037         starting_offset >>= PAGE_CACHE_SHIFT;
2038         if (cmd == OBD_BRW_READ) {
2039                 cli->cl_r_in_flight++;
2040                 lprocfs_oh_tally_log2(&cli->cl_read_page_hist, page_count);
2041                 lprocfs_oh_tally(&cli->cl_read_rpc_hist, cli->cl_r_in_flight);
2042                 lprocfs_oh_tally_log2(&cli->cl_read_offset_hist,
2043                                       starting_offset + 1);
2044         } else {
2045                 cli->cl_w_in_flight++;
2046                 lprocfs_oh_tally_log2(&cli->cl_write_page_hist, page_count);
2047                 lprocfs_oh_tally(&cli->cl_write_rpc_hist, cli->cl_w_in_flight);
2048                 lprocfs_oh_tally_log2(&cli->cl_write_offset_hist,
2049                                       starting_offset + 1);
2050         }
2051         client_obd_list_unlock(&cli->cl_loi_list_lock);
2052
2053         DEBUG_REQ(D_INODE, req, "%d pages, aa %p. now %dr/%dw in flight",
2054                   page_count, aa, cli->cl_r_in_flight,
2055                   cli->cl_w_in_flight);
2056
2057         /* XXX: Maybe the caller can check the RPC bulk descriptor to
2058          * see which CPU/NUMA node the majority of pages were allocated
2059          * on, and try to assign the async RPC to the CPU core
2060          * (PDL_POLICY_PREFERRED) to reduce cross-CPU memory traffic.
2061          *
2062          * But on the other hand, we expect that multiple ptlrpcd
2063          * threads and the initial write sponsor can run in parallel,
2064          * especially when data checksum is enabled, which is CPU-bound
2065          * operation and single ptlrpcd thread cannot process in time.
2066          * So more ptlrpcd threads sharing BRW load
2067          * (with PDL_POLICY_ROUND) seems better.
2068          */
2069         ptlrpcd_add_req(req, pol, -1);
2070         rc = 0;
2071         EXIT;
2072
2073 out:
2074         if (mem_tight != 0)
2075                 cfs_memory_pressure_restore(mpflag);
2076
2077         if (crattr != NULL) {
2078                 capa_put(crattr->cra_capa);
2079                 OBD_FREE(crattr, sizeof(*crattr));
2080         }
2081
2082         if (rc != 0) {
2083                 LASSERT(req == NULL);
2084
2085                 if (oa)
2086                         OBDO_FREE(oa);
2087                 if (pga)
2088                         OBD_FREE(pga, sizeof(*pga) * page_count);
2089                 /* this should happen rarely and is pretty bad, it makes the
2090                  * pending list not follow the dirty order */
2091                 while (!list_empty(ext_list)) {
2092                         ext = list_entry(ext_list->next, struct osc_extent,
2093                                          oe_link);
2094                         list_del_init(&ext->oe_link);
2095                         osc_extent_finish(env, ext, 0, rc);
2096                 }
2097                 if (clerq && !IS_ERR(clerq))
2098                         cl_req_completion(env, clerq, rc);
2099         }
2100         RETURN(rc);
2101 }
2102
2103 static int osc_set_lock_data_with_check(struct ldlm_lock *lock,
2104                                         struct ldlm_enqueue_info *einfo)
2105 {
2106         void *data = einfo->ei_cbdata;
2107         int set = 0;
2108
2109         LASSERT(lock != NULL);
2110         LASSERT(lock->l_blocking_ast == einfo->ei_cb_bl);
2111         LASSERT(lock->l_resource->lr_type == einfo->ei_type);
2112         LASSERT(lock->l_completion_ast == einfo->ei_cb_cp);
2113         LASSERT(lock->l_glimpse_ast == einfo->ei_cb_gl);
2114
2115         lock_res_and_lock(lock);
2116         spin_lock(&osc_ast_guard);
2117
2118         if (lock->l_ast_data == NULL)
2119                 lock->l_ast_data = data;
2120         if (lock->l_ast_data == data)
2121                 set = 1;
2122
2123         spin_unlock(&osc_ast_guard);
2124         unlock_res_and_lock(lock);
2125
2126         return set;
2127 }
2128
2129 static int osc_set_data_with_check(struct lustre_handle *lockh,
2130                                    struct ldlm_enqueue_info *einfo)
2131 {
2132         struct ldlm_lock *lock = ldlm_handle2lock(lockh);
2133         int set = 0;
2134
2135         if (lock != NULL) {
2136                 set = osc_set_lock_data_with_check(lock, einfo);
2137                 LDLM_LOCK_PUT(lock);
2138         } else
2139                 CERROR("lockh %p, data %p - client evicted?\n",
2140                        lockh, einfo->ei_cbdata);
2141         return set;
2142 }
2143
2144 static int osc_change_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
2145                              ldlm_iterator_t replace, void *data)
2146 {
2147         struct ldlm_res_id res_id;
2148         struct obd_device *obd = class_exp2obd(exp);
2149
2150         ostid_build_res_name(&lsm->lsm_oi, &res_id);
2151         ldlm_resource_iterate(obd->obd_namespace, &res_id, replace, data);
2152         return 0;
2153 }
2154
2155 /* find any ldlm lock of the inode in osc
2156  * return 0    not find
2157  *        1    find one
2158  *      < 0    error */
2159 static int osc_find_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
2160                            ldlm_iterator_t replace, void *data)
2161 {
2162         struct ldlm_res_id res_id;
2163         struct obd_device *obd = class_exp2obd(exp);
2164         int rc = 0;
2165
2166         ostid_build_res_name(&lsm->lsm_oi, &res_id);
2167         rc = ldlm_resource_iterate(obd->obd_namespace, &res_id, replace, data);
2168         if (rc == LDLM_ITER_STOP)
2169                 return(1);
2170         if (rc == LDLM_ITER_CONTINUE)
2171                 return(0);
2172         return(rc);
2173 }
2174
2175 static int osc_enqueue_fini(struct ptlrpc_request *req, struct ost_lvb *lvb,
2176                             obd_enqueue_update_f upcall, void *cookie,
2177                             __u64 *flags, int agl, int rc)
2178 {
2179         int intent = *flags & LDLM_FL_HAS_INTENT;
2180         ENTRY;
2181
2182         if (intent) {
2183                 /* The request was created before ldlm_cli_enqueue call. */
2184                 if (rc == ELDLM_LOCK_ABORTED) {
2185                         struct ldlm_reply *rep;
2186                         rep = req_capsule_server_get(&req->rq_pill,
2187                                                      &RMF_DLM_REP);
2188
2189                         LASSERT(rep != NULL);
2190                         rep->lock_policy_res1 =
2191                                 ptlrpc_status_ntoh(rep->lock_policy_res1);
2192                         if (rep->lock_policy_res1)
2193                                 rc = rep->lock_policy_res1;
2194                 }
2195         }
2196
2197         if ((intent != 0 && rc == ELDLM_LOCK_ABORTED && agl == 0) ||
2198             (rc == 0)) {
2199                 *flags |= LDLM_FL_LVB_READY;
2200                 CDEBUG(D_INODE,"got kms "LPU64" blocks "LPU64" mtime "LPU64"\n",
2201                        lvb->lvb_size, lvb->lvb_blocks, lvb->lvb_mtime);
2202         }
2203
2204         /* Call the update callback. */
2205         rc = (*upcall)(cookie, rc);
2206         RETURN(rc);
2207 }
2208
2209 static int osc_enqueue_interpret(const struct lu_env *env,
2210                                  struct ptlrpc_request *req,
2211                                  struct osc_enqueue_args *aa, int rc)
2212 {
2213         struct ldlm_lock *lock;
2214         struct lustre_handle handle;
2215         __u32 mode;
2216         struct ost_lvb *lvb;
2217         __u32 lvb_len;
2218         __u64 *flags = aa->oa_flags;
2219
2220         /* Make a local copy of a lock handle and a mode, because aa->oa_*
2221          * might be freed anytime after lock upcall has been called. */
2222         lustre_handle_copy(&handle, aa->oa_lockh);
2223         mode = aa->oa_ei->ei_mode;
2224
2225         /* ldlm_cli_enqueue is holding a reference on the lock, so it must
2226          * be valid. */
2227         lock = ldlm_handle2lock(&handle);
2228
2229         /* Take an additional reference so that a blocking AST that
2230          * ldlm_cli_enqueue_fini() might post for a failed lock, is guaranteed
2231          * to arrive after an upcall has been executed by
2232          * osc_enqueue_fini(). */
2233         ldlm_lock_addref(&handle, mode);
2234
2235         /* Let cl_lock_state_wait fail with -ERESTARTSYS to unuse sublocks. */
2236         OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_ENQUEUE_HANG, 2);
2237
2238         /* Let CP AST to grant the lock first. */
2239         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_ENQ_RACE, 1);
2240
2241         if (aa->oa_agl && rc == ELDLM_LOCK_ABORTED) {
2242                 lvb = NULL;
2243                 lvb_len = 0;
2244         } else {
2245                 lvb = aa->oa_lvb;
2246                 lvb_len = sizeof(*aa->oa_lvb);
2247         }
2248
2249         /* Complete obtaining the lock procedure. */
2250         rc = ldlm_cli_enqueue_fini(aa->oa_exp, req, aa->oa_ei->ei_type, 1,
2251                                    mode, flags, lvb, lvb_len, &handle, rc);
2252         /* Complete osc stuff. */
2253         rc = osc_enqueue_fini(req, aa->oa_lvb, aa->oa_upcall, aa->oa_cookie,
2254                               flags, aa->oa_agl, rc);
2255
2256         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_CANCEL_RACE, 10);
2257
2258         /* Release the lock for async request. */
2259         if (lustre_handle_is_used(&handle) && rc == ELDLM_OK)
2260                 /*
2261                  * Releases a reference taken by ldlm_cli_enqueue(), if it is
2262                  * not already released by
2263                  * ldlm_cli_enqueue_fini()->failed_lock_cleanup()
2264                  */
2265                 ldlm_lock_decref(&handle, mode);
2266
2267         LASSERTF(lock != NULL, "lockh %p, req %p, aa %p - client evicted?\n",
2268                  aa->oa_lockh, req, aa);
2269         ldlm_lock_decref(&handle, mode);
2270         LDLM_LOCK_PUT(lock);
2271         return rc;
2272 }
2273
2274 struct ptlrpc_request_set *PTLRPCD_SET = (void *)1;
2275
2276 /* When enqueuing asynchronously, locks are not ordered, we can obtain a lock
2277  * from the 2nd OSC before a lock from the 1st one. This does not deadlock with
2278  * other synchronous requests, however keeping some locks and trying to obtain
2279  * others may take a considerable amount of time in a case of ost failure; and
2280  * when other sync requests do not get released lock from a client, the client
2281  * is excluded from the cluster -- such scenarious make the life difficult, so
2282  * release locks just after they are obtained. */
2283 int osc_enqueue_base(struct obd_export *exp, struct ldlm_res_id *res_id,
2284                      __u64 *flags, ldlm_policy_data_t *policy,
2285                      struct ost_lvb *lvb, int kms_valid,
2286                      obd_enqueue_update_f upcall, void *cookie,
2287                      struct ldlm_enqueue_info *einfo,
2288                      struct lustre_handle *lockh,
2289                      struct ptlrpc_request_set *rqset, int async, int agl)
2290 {
2291         struct obd_device *obd = exp->exp_obd;
2292         struct ptlrpc_request *req = NULL;
2293         int intent = *flags & LDLM_FL_HAS_INTENT;
2294         __u64 match_lvb = (agl != 0 ? 0 : LDLM_FL_LVB_READY);
2295         ldlm_mode_t mode;
2296         int rc;
2297         ENTRY;
2298
2299         /* Filesystem lock extents are extended to page boundaries so that
2300          * dealing with the page cache is a little smoother.  */
2301         policy->l_extent.start -= policy->l_extent.start & ~CFS_PAGE_MASK;
2302         policy->l_extent.end |= ~CFS_PAGE_MASK;
2303
2304         /*
2305          * kms is not valid when either object is completely fresh (so that no
2306          * locks are cached), or object was evicted. In the latter case cached
2307          * lock cannot be used, because it would prime inode state with
2308          * potentially stale LVB.
2309          */
2310         if (!kms_valid)
2311                 goto no_match;
2312
2313         /* Next, search for already existing extent locks that will cover us */
2314         /* If we're trying to read, we also search for an existing PW lock.  The
2315          * VFS and page cache already protect us locally, so lots of readers/
2316          * writers can share a single PW lock.
2317          *
2318          * There are problems with conversion deadlocks, so instead of
2319          * converting a read lock to a write lock, we'll just enqueue a new
2320          * one.
2321          *
2322          * At some point we should cancel the read lock instead of making them
2323          * send us a blocking callback, but there are problems with canceling
2324          * locks out from other users right now, too. */
2325         mode = einfo->ei_mode;
2326         if (einfo->ei_mode == LCK_PR)
2327                 mode |= LCK_PW;
2328         mode = ldlm_lock_match(obd->obd_namespace, *flags | match_lvb, res_id,
2329                                einfo->ei_type, policy, mode, lockh, 0);
2330         if (mode) {
2331                 struct ldlm_lock *matched = ldlm_handle2lock(lockh);
2332
2333                 if ((agl != 0) && !ldlm_is_lvb_ready(matched)) {
2334                         /* For AGL, if enqueue RPC is sent but the lock is not
2335                          * granted, then skip to process this strpe.
2336                          * Return -ECANCELED to tell the caller. */
2337                         ldlm_lock_decref(lockh, mode);
2338                         LDLM_LOCK_PUT(matched);
2339                         RETURN(-ECANCELED);
2340                 } else if (osc_set_lock_data_with_check(matched, einfo)) {
2341                         *flags |= LDLM_FL_LVB_READY;
2342                         /* addref the lock only if not async requests and PW
2343                          * lock is matched whereas we asked for PR. */
2344                         if (!rqset && einfo->ei_mode != mode)
2345                                 ldlm_lock_addref(lockh, LCK_PR);
2346                         if (intent) {
2347                                 /* I would like to be able to ASSERT here that
2348                                  * rss <= kms, but I can't, for reasons which
2349                                  * are explained in lov_enqueue() */
2350                         }
2351
2352                         /* We already have a lock, and it's referenced.
2353                          *
2354                          * At this point, the cl_lock::cll_state is CLS_QUEUING,
2355                          * AGL upcall may change it to CLS_HELD directly. */
2356                         (*upcall)(cookie, ELDLM_OK);
2357
2358                         if (einfo->ei_mode != mode)
2359                                 ldlm_lock_decref(lockh, LCK_PW);
2360                         else if (rqset)
2361                                 /* For async requests, decref the lock. */
2362                                 ldlm_lock_decref(lockh, einfo->ei_mode);
2363                         LDLM_LOCK_PUT(matched);
2364                         RETURN(ELDLM_OK);
2365                 } else {
2366                         ldlm_lock_decref(lockh, mode);
2367                         LDLM_LOCK_PUT(matched);
2368                 }
2369         }
2370
2371  no_match:
2372         if (intent) {
2373                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
2374                                            &RQF_LDLM_ENQUEUE_LVB);
2375                 if (req == NULL)
2376                         RETURN(-ENOMEM);
2377
2378                 rc = ptlrpc_request_pack(req, LUSTRE_DLM_VERSION, LDLM_ENQUEUE);
2379                 if (rc < 0) {
2380                         ptlrpc_request_free(req);
2381                         RETURN(rc);
2382                 }
2383
2384                 req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
2385                                      sizeof *lvb);
2386                 ptlrpc_request_set_replen(req);
2387         }
2388
2389         /* users of osc_enqueue() can pass this flag for ldlm_lock_match() */
2390         *flags &= ~LDLM_FL_BLOCK_GRANTED;
2391
2392         rc = ldlm_cli_enqueue(exp, &req, einfo, res_id, policy, flags, lvb,
2393                               sizeof(*lvb), LVB_T_OST, lockh, async);
2394         if (rqset) {
2395                 if (!rc) {
2396                         struct osc_enqueue_args *aa;
2397                         CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
2398                         aa = ptlrpc_req_async_args(req);
2399                         aa->oa_ei = einfo;
2400                         aa->oa_exp = exp;
2401                         aa->oa_flags  = flags;
2402                         aa->oa_upcall = upcall;
2403                         aa->oa_cookie = cookie;
2404                         aa->oa_lvb    = lvb;
2405                         aa->oa_lockh  = lockh;
2406                         aa->oa_agl    = !!agl;
2407
2408                         req->rq_interpret_reply =
2409                                 (ptlrpc_interpterer_t)osc_enqueue_interpret;
2410                         if (rqset == PTLRPCD_SET)
2411                                 ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
2412                         else
2413                                 ptlrpc_set_add_req(rqset, req);
2414                 } else if (intent) {
2415                         ptlrpc_req_finished(req);
2416                 }
2417                 RETURN(rc);
2418         }
2419
2420         rc = osc_enqueue_fini(req, lvb, upcall, cookie, flags, agl, rc);
2421         if (intent)
2422                 ptlrpc_req_finished(req);
2423
2424         RETURN(rc);
2425 }
2426
2427 int osc_match_base(struct obd_export *exp, struct ldlm_res_id *res_id,
2428                    __u32 type, ldlm_policy_data_t *policy, __u32 mode,
2429                    __u64 *flags, void *data, struct lustre_handle *lockh,
2430                    int unref)
2431 {
2432         struct obd_device *obd = exp->exp_obd;
2433         __u64 lflags = *flags;
2434         ldlm_mode_t rc;
2435         ENTRY;
2436
2437         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_MATCH))
2438                 RETURN(-EIO);
2439
2440         /* Filesystem lock extents are extended to page boundaries so that
2441          * dealing with the page cache is a little smoother */
2442         policy->l_extent.start -= policy->l_extent.start & ~CFS_PAGE_MASK;
2443         policy->l_extent.end |= ~CFS_PAGE_MASK;
2444
2445         /* Next, search for already existing extent locks that will cover us */
2446         /* If we're trying to read, we also search for an existing PW lock.  The
2447          * VFS and page cache already protect us locally, so lots of readers/
2448          * writers can share a single PW lock. */
2449         rc = mode;
2450         if (mode == LCK_PR)
2451                 rc |= LCK_PW;
2452         rc = ldlm_lock_match(obd->obd_namespace, lflags,
2453                              res_id, type, policy, rc, lockh, unref);
2454         if (rc) {
2455                 if (data != NULL) {
2456                         if (!osc_set_data_with_check(lockh, data)) {
2457                                 if (!(lflags & LDLM_FL_TEST_LOCK))
2458                                         ldlm_lock_decref(lockh, rc);
2459                                 RETURN(0);
2460                         }
2461                 }
2462                 if (!(lflags & LDLM_FL_TEST_LOCK) && mode != rc) {
2463                         ldlm_lock_addref(lockh, LCK_PR);
2464                         ldlm_lock_decref(lockh, LCK_PW);
2465                 }
2466                 RETURN(rc);
2467         }
2468         RETURN(rc);
2469 }
2470
2471 int osc_cancel_base(struct lustre_handle *lockh, __u32 mode)
2472 {
2473         ENTRY;
2474
2475         if (unlikely(mode == LCK_GROUP))
2476                 ldlm_lock_decref_and_cancel(lockh, mode);
2477         else
2478                 ldlm_lock_decref(lockh, mode);
2479
2480         RETURN(0);
2481 }
2482
2483 static int osc_statfs_interpret(const struct lu_env *env,
2484                                 struct ptlrpc_request *req,
2485                                 struct osc_async_args *aa, int rc)
2486 {
2487         struct obd_statfs *msfs;
2488         ENTRY;
2489
2490         if (rc == -EBADR)
2491                 /* The request has in fact never been sent
2492                  * due to issues at a higher level (LOV).
2493                  * Exit immediately since the caller is
2494                  * aware of the problem and takes care
2495                  * of the clean up */
2496                  RETURN(rc);
2497
2498         if ((rc == -ENOTCONN || rc == -EAGAIN) &&
2499             (aa->aa_oi->oi_flags & OBD_STATFS_NODELAY))
2500                 GOTO(out, rc = 0);
2501
2502         if (rc != 0)
2503                 GOTO(out, rc);
2504
2505         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
2506         if (msfs == NULL) {
2507                 GOTO(out, rc = -EPROTO);
2508         }
2509
2510         *aa->aa_oi->oi_osfs = *msfs;
2511 out:
2512         rc = aa->aa_oi->oi_cb_up(aa->aa_oi, rc);
2513         RETURN(rc);
2514 }
2515
2516 static int osc_statfs_async(struct obd_export *exp,
2517                             struct obd_info *oinfo, __u64 max_age,
2518                             struct ptlrpc_request_set *rqset)
2519 {
2520         struct obd_device     *obd = class_exp2obd(exp);
2521         struct ptlrpc_request *req;
2522         struct osc_async_args *aa;
2523         int                    rc;
2524         ENTRY;
2525
2526         /* We could possibly pass max_age in the request (as an absolute
2527          * timestamp or a "seconds.usec ago") so the target can avoid doing
2528          * extra calls into the filesystem if that isn't necessary (e.g.
2529          * during mount that would help a bit).  Having relative timestamps
2530          * is not so great if request processing is slow, while absolute
2531          * timestamps are not ideal because they need time synchronization. */
2532         req = ptlrpc_request_alloc(obd->u.cli.cl_import, &RQF_OST_STATFS);
2533         if (req == NULL)
2534                 RETURN(-ENOMEM);
2535
2536         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
2537         if (rc) {
2538                 ptlrpc_request_free(req);
2539                 RETURN(rc);
2540         }
2541         ptlrpc_request_set_replen(req);
2542         req->rq_request_portal = OST_CREATE_PORTAL;
2543         ptlrpc_at_set_req_timeout(req);
2544
2545         if (oinfo->oi_flags & OBD_STATFS_NODELAY) {
2546                 /* procfs requests not want stat in wait for avoid deadlock */
2547                 req->rq_no_resend = 1;
2548                 req->rq_no_delay = 1;
2549         }
2550
2551         req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_statfs_interpret;
2552         CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
2553         aa = ptlrpc_req_async_args(req);
2554         aa->aa_oi = oinfo;
2555
2556         ptlrpc_set_add_req(rqset, req);
2557         RETURN(0);
2558 }
2559
2560 static int osc_statfs(const struct lu_env *env, struct obd_export *exp,
2561                       struct obd_statfs *osfs, __u64 max_age, __u32 flags)
2562 {
2563         struct obd_device     *obd = class_exp2obd(exp);
2564         struct obd_statfs     *msfs;
2565         struct ptlrpc_request *req;
2566         struct obd_import     *imp = NULL;
2567         int rc;
2568         ENTRY;
2569
2570         /*Since the request might also come from lprocfs, so we need
2571          *sync this with client_disconnect_export Bug15684*/
2572         down_read(&obd->u.cli.cl_sem);
2573         if (obd->u.cli.cl_import)
2574                 imp = class_import_get(obd->u.cli.cl_import);
2575         up_read(&obd->u.cli.cl_sem);
2576         if (!imp)
2577                 RETURN(-ENODEV);
2578
2579         /* We could possibly pass max_age in the request (as an absolute
2580          * timestamp or a "seconds.usec ago") so the target can avoid doing
2581          * extra calls into the filesystem if that isn't necessary (e.g.
2582          * during mount that would help a bit).  Having relative timestamps
2583          * is not so great if request processing is slow, while absolute
2584          * timestamps are not ideal because they need time synchronization. */
2585         req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
2586
2587         class_import_put(imp);
2588
2589         if (req == NULL)
2590                 RETURN(-ENOMEM);
2591
2592         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
2593         if (rc) {
2594                 ptlrpc_request_free(req);
2595                 RETURN(rc);
2596         }
2597         ptlrpc_request_set_replen(req);
2598         req->rq_request_portal = OST_CREATE_PORTAL;
2599         ptlrpc_at_set_req_timeout(req);
2600
2601         if (flags & OBD_STATFS_NODELAY) {
2602                 /* procfs requests not want stat in wait for avoid deadlock */
2603                 req->rq_no_resend = 1;
2604                 req->rq_no_delay = 1;
2605         }
2606
2607         rc = ptlrpc_queue_wait(req);
2608         if (rc)
2609                 GOTO(out, rc);
2610
2611         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
2612         if (msfs == NULL) {
2613                 GOTO(out, rc = -EPROTO);
2614         }
2615
2616         *osfs = *msfs;
2617
2618         EXIT;
2619  out:
2620         ptlrpc_req_finished(req);
2621         return rc;
2622 }
2623
2624 static int osc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
2625                          void *karg, void *uarg)
2626 {
2627         struct obd_device *obd = exp->exp_obd;
2628         struct obd_ioctl_data *data = karg;
2629         int err = 0;
2630         ENTRY;
2631
2632         if (!try_module_get(THIS_MODULE)) {
2633                 CERROR("%s: cannot get module '%s'\n", obd->obd_name,
2634                        module_name(THIS_MODULE));
2635                 return -EINVAL;
2636         }
2637         switch (cmd) {
2638         case OBD_IOC_CLIENT_RECOVER:
2639                 err = ptlrpc_recover_import(obd->u.cli.cl_import,
2640                                             data->ioc_inlbuf1, 0);
2641                 if (err > 0)
2642                         err = 0;
2643                 GOTO(out, err);
2644         case IOC_OSC_SET_ACTIVE:
2645                 err = ptlrpc_set_import_active(obd->u.cli.cl_import,
2646                                                data->ioc_offset);
2647                 GOTO(out, err);
2648         case OBD_IOC_POLL_QUOTACHECK:
2649                 err = osc_quota_poll_check(exp, (struct if_quotacheck *)karg);
2650                 GOTO(out, err);
2651         case OBD_IOC_PING_TARGET:
2652                 err = ptlrpc_obd_ping(obd);
2653                 GOTO(out, err);
2654         default:
2655                 CDEBUG(D_INODE, "unrecognised ioctl %#x by %s\n",
2656                        cmd, current_comm());
2657                 GOTO(out, err = -ENOTTY);
2658         }
2659 out:
2660         module_put(THIS_MODULE);
2661         return err;
2662 }
2663
2664 static int osc_get_info(const struct lu_env *env, struct obd_export *exp,
2665                         obd_count keylen, void *key, __u32 *vallen, void *val,
2666                         struct lov_stripe_md *lsm)
2667 {
2668         ENTRY;
2669         if (!vallen || !val)
2670                 RETURN(-EFAULT);
2671
2672         if (KEY_IS(KEY_LOCK_TO_STRIPE)) {
2673                 __u32 *stripe = val;
2674                 *vallen = sizeof(*stripe);
2675                 *stripe = 0;
2676                 RETURN(0);
2677         } else if (KEY_IS(KEY_LAST_ID)) {
2678                 struct ptlrpc_request *req;
2679                 obd_id                *reply;
2680                 char                  *tmp;
2681                 int                    rc;
2682
2683                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
2684                                            &RQF_OST_GET_INFO_LAST_ID);
2685                 if (req == NULL)
2686                         RETURN(-ENOMEM);
2687
2688                 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
2689                                      RCL_CLIENT, keylen);
2690                 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GET_INFO);
2691                 if (rc) {
2692                         ptlrpc_request_free(req);
2693                         RETURN(rc);
2694                 }
2695
2696                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
2697                 memcpy(tmp, key, keylen);
2698
2699                 req->rq_no_delay = req->rq_no_resend = 1;
2700                 ptlrpc_request_set_replen(req);
2701                 rc = ptlrpc_queue_wait(req);
2702                 if (rc)
2703                         GOTO(out, rc);
2704
2705                 reply = req_capsule_server_get(&req->rq_pill, &RMF_OBD_ID);
2706                 if (reply == NULL)
2707                         GOTO(out, rc = -EPROTO);
2708
2709                 *((obd_id *)val) = *reply;
2710         out:
2711                 ptlrpc_req_finished(req);
2712                 RETURN(rc);
2713         } else if (KEY_IS(KEY_FIEMAP)) {
2714                 struct ll_fiemap_info_key *fm_key =
2715                                 (struct ll_fiemap_info_key *)key;
2716                 struct ldlm_res_id       res_id;
2717                 ldlm_policy_data_t       policy;
2718                 struct lustre_handle     lockh;
2719                 ldlm_mode_t              mode = 0;
2720                 struct ptlrpc_request   *req;
2721                 struct ll_user_fiemap   *reply;
2722                 char                    *tmp;
2723                 int                      rc;
2724
2725                 if (!(fm_key->fiemap.fm_flags & FIEMAP_FLAG_SYNC))
2726                         goto skip_locking;
2727
2728                 policy.l_extent.start = fm_key->fiemap.fm_start &
2729                                                 CFS_PAGE_MASK;
2730
2731                 if (OBD_OBJECT_EOF - fm_key->fiemap.fm_length <=
2732                     fm_key->fiemap.fm_start + PAGE_CACHE_SIZE - 1)
2733                         policy.l_extent.end = OBD_OBJECT_EOF;
2734                 else
2735                         policy.l_extent.end = (fm_key->fiemap.fm_start +
2736                                 fm_key->fiemap.fm_length +
2737                                 PAGE_CACHE_SIZE - 1) & CFS_PAGE_MASK;
2738
2739                 ostid_build_res_name(&fm_key->oa.o_oi, &res_id);
2740                 mode = ldlm_lock_match(exp->exp_obd->obd_namespace,
2741                                        LDLM_FL_BLOCK_GRANTED |
2742                                        LDLM_FL_LVB_READY,
2743                                        &res_id, LDLM_EXTENT, &policy,
2744                                        LCK_PR | LCK_PW, &lockh, 0);
2745                 if (mode) { /* lock is cached on client */
2746                         if (mode != LCK_PR) {
2747                                 ldlm_lock_addref(&lockh, LCK_PR);
2748                                 ldlm_lock_decref(&lockh, LCK_PW);
2749                         }
2750                 } else { /* no cached lock, needs acquire lock on server side */
2751                         fm_key->oa.o_valid |= OBD_MD_FLFLAGS;
2752                         fm_key->oa.o_flags |= OBD_FL_SRVLOCK;
2753                 }
2754
2755 skip_locking:
2756                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
2757                                            &RQF_OST_GET_INFO_FIEMAP);
2758                 if (req == NULL)
2759                         GOTO(drop_lock, rc = -ENOMEM);
2760
2761                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_KEY,
2762                                      RCL_CLIENT, keylen);
2763                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_VAL,
2764                                      RCL_CLIENT, *vallen);
2765                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_VAL,
2766                                      RCL_SERVER, *vallen);
2767
2768                 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GET_INFO);
2769                 if (rc) {
2770                         ptlrpc_request_free(req);
2771                         GOTO(drop_lock, rc);
2772                 }
2773
2774                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_FIEMAP_KEY);
2775                 memcpy(tmp, key, keylen);
2776                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_FIEMAP_VAL);
2777                 memcpy(tmp, val, *vallen);
2778
2779                 ptlrpc_request_set_replen(req);
2780                 rc = ptlrpc_queue_wait(req);
2781                 if (rc)
2782                         GOTO(fini_req, rc);
2783
2784                 reply = req_capsule_server_get(&req->rq_pill, &RMF_FIEMAP_VAL);
2785                 if (reply == NULL)
2786                         GOTO(fini_req, rc = -EPROTO);
2787
2788                 memcpy(val, reply, *vallen);
2789 fini_req:
2790                 ptlrpc_req_finished(req);
2791 drop_lock:
2792                 if (mode)
2793                         ldlm_lock_decref(&lockh, LCK_PR);
2794                 RETURN(rc);
2795         }
2796
2797         RETURN(-EINVAL);
2798 }
2799
2800 static int osc_set_info_async(const struct lu_env *env, struct obd_export *exp,
2801                               obd_count keylen, void *key, obd_count vallen,
2802                               void *val, struct ptlrpc_request_set *set)
2803 {
2804         struct ptlrpc_request *req;
2805         struct obd_device     *obd = exp->exp_obd;
2806         struct obd_import     *imp = class_exp2cliimp(exp);
2807         char                  *tmp;
2808         int                    rc;
2809         ENTRY;
2810
2811         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_SHUTDOWN, 10);
2812
2813         if (KEY_IS(KEY_CHECKSUM)) {
2814                 if (vallen != sizeof(int))
2815                         RETURN(-EINVAL);
2816                 exp->exp_obd->u.cli.cl_checksum = (*(int *)val) ? 1 : 0;
2817                 RETURN(0);
2818         }
2819
2820         if (KEY_IS(KEY_SPTLRPC_CONF)) {
2821                 sptlrpc_conf_client_adapt(obd);
2822                 RETURN(0);
2823         }
2824
2825         if (KEY_IS(KEY_FLUSH_CTX)) {
2826                 sptlrpc_import_flush_my_ctx(imp);
2827                 RETURN(0);
2828         }
2829
2830         if (KEY_IS(KEY_CACHE_SET)) {
2831                 struct client_obd *cli = &obd->u.cli;
2832
2833                 LASSERT(cli->cl_cache == NULL); /* only once */
2834                 cli->cl_cache = (struct cl_client_cache *)val;
2835                 atomic_inc(&cli->cl_cache->ccc_users);
2836                 cli->cl_lru_left = &cli->cl_cache->ccc_lru_left;
2837
2838                 /* add this osc into entity list */
2839                 LASSERT(list_empty(&cli->cl_lru_osc));
2840                 spin_lock(&cli->cl_cache->ccc_lru_lock);
2841                 list_add(&cli->cl_lru_osc, &cli->cl_cache->ccc_lru);
2842                 spin_unlock(&cli->cl_cache->ccc_lru_lock);
2843
2844                 RETURN(0);
2845         }
2846
2847         if (KEY_IS(KEY_CACHE_LRU_SHRINK)) {
2848                 struct client_obd *cli = &obd->u.cli;
2849                 int nr = atomic_read(&cli->cl_lru_in_list) >> 1;
2850                 int target = *(int *)val;
2851
2852                 nr = osc_lru_shrink(env, cli, min(nr, target), true);
2853                 *(int *)val -= nr;
2854                 RETURN(0);
2855         }
2856
2857         if (!set && !KEY_IS(KEY_GRANT_SHRINK))
2858                 RETURN(-EINVAL);
2859
2860         /* We pass all other commands directly to OST. Since nobody calls osc
2861            methods directly and everybody is supposed to go through LOV, we
2862            assume lov checked invalid values for us.
2863            The only recognised values so far are evict_by_nid and mds_conn.
2864            Even if something bad goes through, we'd get a -EINVAL from OST
2865            anyway. */
2866
2867         req = ptlrpc_request_alloc(imp, KEY_IS(KEY_GRANT_SHRINK) ?
2868                                                 &RQF_OST_SET_GRANT_INFO :
2869                                                 &RQF_OBD_SET_INFO);
2870         if (req == NULL)
2871                 RETURN(-ENOMEM);
2872
2873         req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
2874                              RCL_CLIENT, keylen);
2875         if (!KEY_IS(KEY_GRANT_SHRINK))
2876                 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_VAL,
2877                                      RCL_CLIENT, vallen);
2878         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SET_INFO);
2879         if (rc) {
2880                 ptlrpc_request_free(req);
2881                 RETURN(rc);
2882         }
2883
2884         tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
2885         memcpy(tmp, key, keylen);
2886         tmp = req_capsule_client_get(&req->rq_pill, KEY_IS(KEY_GRANT_SHRINK) ?
2887                                                         &RMF_OST_BODY :
2888                                                         &RMF_SETINFO_VAL);
2889         memcpy(tmp, val, vallen);
2890
2891         if (KEY_IS(KEY_GRANT_SHRINK)) {
2892                 struct osc_grant_args *aa;
2893                 struct obdo *oa;
2894
2895                 CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
2896                 aa = ptlrpc_req_async_args(req);
2897                 OBDO_ALLOC(oa);
2898                 if (!oa) {
2899                         ptlrpc_req_finished(req);
2900                         RETURN(-ENOMEM);
2901                 }
2902                 *oa = ((struct ost_body *)val)->oa;
2903                 aa->aa_oa = oa;
2904                 req->rq_interpret_reply = osc_shrink_grant_interpret;
2905         }
2906
2907         ptlrpc_request_set_replen(req);
2908         if (!KEY_IS(KEY_GRANT_SHRINK)) {
2909                 LASSERT(set != NULL);
2910                 ptlrpc_set_add_req(set, req);
2911                 ptlrpc_check_set(NULL, set);
2912         } else
2913                 ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
2914
2915         RETURN(0);
2916 }
2917
2918 static int osc_reconnect(const struct lu_env *env,
2919                          struct obd_export *exp, struct obd_device *obd,
2920                          struct obd_uuid *cluuid,
2921                          struct obd_connect_data *data,
2922                          void *localdata)
2923 {
2924         struct client_obd *cli = &obd->u.cli;
2925
2926         if (data != NULL && (data->ocd_connect_flags & OBD_CONNECT_GRANT)) {
2927                 long lost_grant;
2928
2929                 client_obd_list_lock(&cli->cl_loi_list_lock);
2930                 data->ocd_grant = (cli->cl_avail_grant +
2931                                   (cli->cl_dirty_pages << PAGE_CACHE_SHIFT)) ?:
2932                                   2 * cli_brw_size(obd);
2933                 lost_grant = cli->cl_lost_grant;
2934                 cli->cl_lost_grant = 0;
2935                 client_obd_list_unlock(&cli->cl_loi_list_lock);
2936
2937                 CDEBUG(D_RPCTRACE, "ocd_connect_flags: "LPX64" ocd_version: %d"
2938                        " ocd_grant: %d, lost: %ld.\n", data->ocd_connect_flags,
2939                        data->ocd_version, data->ocd_grant, lost_grant);
2940         }
2941
2942         RETURN(0);
2943 }
2944
2945 static int osc_disconnect(struct obd_export *exp)
2946 {
2947         struct obd_device *obd = class_exp2obd(exp);
2948         struct llog_ctxt  *ctxt;
2949         int rc;
2950
2951         ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
2952         if (ctxt) {
2953                 if (obd->u.cli.cl_conn_count == 1) {
2954                         /* Flush any remaining cancel messages out to the
2955                          * target */
2956                         llog_sync(ctxt, exp, 0);
2957                 }
2958                 llog_ctxt_put(ctxt);
2959         } else {
2960                 CDEBUG(D_HA, "No LLOG_SIZE_REPL_CTXT found in obd %p\n",
2961                        obd);
2962         }
2963
2964         rc = client_disconnect_export(exp);
2965         /**
2966          * Initially we put del_shrink_grant before disconnect_export, but it
2967          * causes the following problem if setup (connect) and cleanup
2968          * (disconnect) are tangled together.
2969          *      connect p1                     disconnect p2
2970          *   ptlrpc_connect_import
2971          *     ...............               class_manual_cleanup
2972          *                                     osc_disconnect
2973          *                                     del_shrink_grant
2974          *   ptlrpc_connect_interrupt
2975          *     init_grant_shrink
2976          *   add this client to shrink list
2977          *                                      cleanup_osc
2978          * Bang! pinger trigger the shrink.
2979          * So the osc should be disconnected from the shrink list, after we
2980          * are sure the import has been destroyed. BUG18662
2981          */
2982         if (obd->u.cli.cl_import == NULL)
2983                 osc_del_shrink_grant(&obd->u.cli);
2984         return rc;
2985 }
2986
2987 static int osc_import_event(struct obd_device *obd,
2988                             struct obd_import *imp,
2989                             enum obd_import_event event)
2990 {
2991         struct client_obd *cli;
2992         int rc = 0;
2993
2994         ENTRY;
2995         LASSERT(imp->imp_obd == obd);
2996
2997         switch (event) {
2998         case IMP_EVENT_DISCON: {
2999                 cli = &obd->u.cli;
3000                 client_obd_list_lock(&cli->cl_loi_list_lock);
3001                 cli->cl_avail_grant = 0;
3002                 cli->cl_lost_grant = 0;
3003                 client_obd_list_unlock(&cli->cl_loi_list_lock);
3004                 break;
3005         }
3006         case IMP_EVENT_INACTIVE: {
3007                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_INACTIVE, NULL);
3008                 break;
3009         }
3010         case IMP_EVENT_INVALIDATE: {
3011                 struct ldlm_namespace *ns = obd->obd_namespace;
3012                 struct lu_env         *env;
3013                 int                    refcheck;
3014
3015                 env = cl_env_get(&refcheck);
3016                 if (!IS_ERR(env)) {
3017                         /* Reset grants */
3018                         cli = &obd->u.cli;
3019                         /* all pages go to failing rpcs due to the invalid
3020                          * import */
3021                         osc_io_unplug(env, cli, NULL, PDL_POLICY_ROUND);
3022
3023                         ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
3024                         cl_env_put(env, &refcheck);
3025                 } else
3026                         rc = PTR_ERR(env);
3027                 break;
3028         }
3029         case IMP_EVENT_ACTIVE: {
3030                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE, NULL);
3031                 break;
3032         }
3033         case IMP_EVENT_OCD: {
3034                 struct obd_connect_data *ocd = &imp->imp_connect_data;
3035
3036                 if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT)
3037                         osc_init_grant(&obd->u.cli, ocd);
3038
3039                 /* See bug 7198 */
3040                 if (ocd->ocd_connect_flags & OBD_CONNECT_REQPORTAL)
3041                         imp->imp_client->cli_request_portal =OST_REQUEST_PORTAL;
3042
3043                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_OCD, NULL);
3044                 break;
3045         }
3046         case IMP_EVENT_DEACTIVATE: {
3047                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_DEACTIVATE, NULL);
3048                 break;
3049         }
3050         case IMP_EVENT_ACTIVATE: {
3051                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVATE, NULL);
3052                 break;
3053         }
3054         default:
3055                 CERROR("Unknown import event %d\n", event);
3056                 LBUG();
3057         }
3058         RETURN(rc);
3059 }
3060
3061 /**
3062  * Determine whether the lock can be canceled before replaying the lock
3063  * during recovery, see bug16774 for detailed information.
3064  *
3065  * \retval zero the lock can't be canceled
3066  * \retval other ok to cancel
3067  */
3068 static int osc_cancel_weight(struct ldlm_lock *lock)
3069 {
3070         /*
3071          * Cancel all unused and granted extent lock.
3072          */
3073         if (lock->l_resource->lr_type == LDLM_EXTENT &&
3074             lock->l_granted_mode == lock->l_req_mode &&
3075             osc_ldlm_weigh_ast(lock) == 0)
3076                 RETURN(1);
3077
3078         RETURN(0);
3079 }
3080
3081 static int brw_queue_work(const struct lu_env *env, void *data)
3082 {
3083         struct client_obd *cli = data;
3084
3085         CDEBUG(D_CACHE, "Run writeback work for client obd %p.\n", cli);
3086
3087         osc_io_unplug(env, cli, NULL, PDL_POLICY_SAME);
3088         RETURN(0);
3089 }
3090
3091 int osc_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
3092 {
3093         struct client_obd *cli = &obd->u.cli;
3094         struct obd_type   *type;
3095         void              *handler;
3096         int                rc;
3097         ENTRY;
3098
3099         rc = ptlrpcd_addref();
3100         if (rc)
3101                 RETURN(rc);
3102
3103         rc = client_obd_setup(obd, lcfg);
3104         if (rc)
3105                 GOTO(out_ptlrpcd, rc);
3106
3107         handler = ptlrpcd_alloc_work(cli->cl_import, brw_queue_work, cli);
3108         if (IS_ERR(handler))
3109                 GOTO(out_client_setup, rc = PTR_ERR(handler));
3110         cli->cl_writeback_work = handler;
3111
3112         handler = ptlrpcd_alloc_work(cli->cl_import, lru_queue_work, cli);
3113         if (IS_ERR(handler))
3114                 GOTO(out_ptlrpcd_work, rc = PTR_ERR(handler));
3115         cli->cl_lru_work = handler;
3116
3117         rc = osc_quota_setup(obd);
3118         if (rc)
3119                 GOTO(out_ptlrpcd_work, rc);
3120
3121         cli->cl_grant_shrink_interval = GRANT_SHRINK_INTERVAL;
3122
3123 #ifdef LPROCFS
3124         obd->obd_vars = lprocfs_osc_obd_vars;
3125 #endif
3126         /* If this is true then both client (osc) and server (osp) are on the
3127          * same node. The osp layer if loaded first will register the osc proc
3128          * directory. In that case this obd_device will be attached its proc
3129          * tree to type->typ_procsym instead of obd->obd_type->typ_procroot. */
3130         type = class_search_type(LUSTRE_OSP_NAME);
3131         if (type && type->typ_procsym) {
3132                 obd->obd_proc_entry = lprocfs_seq_register(obd->obd_name,
3133                                                            type->typ_procsym,
3134                                                            obd->obd_vars, obd);
3135                 if (IS_ERR(obd->obd_proc_entry)) {
3136                         rc = PTR_ERR(obd->obd_proc_entry);
3137                         CERROR("error %d setting up lprocfs for %s\n", rc,
3138                                obd->obd_name);
3139                         obd->obd_proc_entry = NULL;
3140                 }
3141         } else {
3142                 rc = lprocfs_seq_obd_setup(obd);
3143         }
3144
3145         /* If the basic OSC proc tree construction succeeded then
3146          * lets do the rest. */
3147         if (rc == 0) {
3148                 lproc_osc_attach_seqstat(obd);
3149                 sptlrpc_lprocfs_cliobd_attach(obd);
3150                 ptlrpc_lprocfs_register_obd(obd);
3151         }
3152
3153         /* We need to allocate a few requests more, because
3154          * brw_interpret tries to create new requests before freeing
3155          * previous ones, Ideally we want to have 2x max_rpcs_in_flight
3156          * reserved, but I'm afraid that might be too much wasted RAM
3157          * in fact, so 2 is just my guess and still should work. */
3158         cli->cl_import->imp_rq_pool =
3159                 ptlrpc_init_rq_pool(cli->cl_max_rpcs_in_flight + 2,
3160                                     OST_MAXREQSIZE,
3161                                     ptlrpc_add_rqs_to_pool);
3162
3163         INIT_LIST_HEAD(&cli->cl_grant_shrink_list);
3164         ns_register_cancel(obd->obd_namespace, osc_cancel_weight);
3165         RETURN(0);
3166
3167 out_ptlrpcd_work:
3168         if (cli->cl_writeback_work != NULL) {
3169                 ptlrpcd_destroy_work(cli->cl_writeback_work);
3170                 cli->cl_writeback_work = NULL;
3171         }
3172         if (cli->cl_lru_work != NULL) {
3173                 ptlrpcd_destroy_work(cli->cl_lru_work);
3174                 cli->cl_lru_work = NULL;
3175         }
3176 out_client_setup:
3177         client_obd_cleanup(obd);
3178 out_ptlrpcd:
3179         ptlrpcd_decref();
3180         RETURN(rc);
3181 }
3182
3183 static int osc_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
3184 {
3185         int rc = 0;
3186         ENTRY;
3187
3188         switch (stage) {
3189         case OBD_CLEANUP_EARLY: {
3190                 struct obd_import *imp;
3191                 imp = obd->u.cli.cl_import;
3192                 CDEBUG(D_HA, "Deactivating import %s\n", obd->obd_name);
3193                 /* ptlrpc_abort_inflight to stop an mds_lov_synchronize */
3194                 ptlrpc_deactivate_import(imp);
3195                 spin_lock(&imp->imp_lock);
3196                 imp->imp_pingable = 0;
3197                 spin_unlock(&imp->imp_lock);
3198                 break;
3199         }
3200         case OBD_CLEANUP_EXPORTS: {
3201                 struct client_obd *cli = &obd->u.cli;
3202                 /* LU-464
3203                  * for echo client, export may be on zombie list, wait for
3204                  * zombie thread to cull it, because cli.cl_import will be
3205                  * cleared in client_disconnect_export():
3206                  *   class_export_destroy() -> obd_cleanup() ->
3207                  *   echo_device_free() -> echo_client_cleanup() ->
3208                  *   obd_disconnect() -> osc_disconnect() ->
3209                  *   client_disconnect_export()
3210                  */
3211                 obd_zombie_barrier();
3212                 if (cli->cl_writeback_work) {
3213                         ptlrpcd_destroy_work(cli->cl_writeback_work);
3214                         cli->cl_writeback_work = NULL;
3215                 }
3216                 if (cli->cl_lru_work) {
3217                         ptlrpcd_destroy_work(cli->cl_lru_work);
3218                         cli->cl_lru_work = NULL;
3219                 }
3220                 obd_cleanup_client_import(obd);
3221                 ptlrpc_lprocfs_unregister_obd(obd);
3222                 lprocfs_obd_cleanup(obd);
3223                 rc = obd_llog_finish(obd, 0);
3224                 if (rc != 0)
3225                         CERROR("failed to cleanup llogging subsystems\n");
3226                 break;
3227                 }
3228         }
3229         RETURN(rc);
3230 }
3231
3232 int osc_cleanup(struct obd_device *obd)
3233 {
3234         struct client_obd *cli = &obd->u.cli;
3235         int rc;
3236
3237         ENTRY;
3238
3239         /* lru cleanup */
3240         if (cli->cl_cache != NULL) {
3241                 LASSERT(atomic_read(&cli->cl_cache->ccc_users) > 0);
3242                 spin_lock(&cli->cl_cache->ccc_lru_lock);
3243                 list_del_init(&cli->cl_lru_osc);
3244                 spin_unlock(&cli->cl_cache->ccc_lru_lock);
3245                 cli->cl_lru_left = NULL;
3246                 atomic_dec(&cli->cl_cache->ccc_users);
3247                 cli->cl_cache = NULL;
3248         }
3249
3250         /* free memory of osc quota cache */
3251         osc_quota_cleanup(obd);
3252
3253         rc = client_obd_cleanup(obd);
3254
3255         ptlrpcd_decref();
3256         RETURN(rc);
3257 }
3258
3259 int osc_process_config_base(struct obd_device *obd, struct lustre_cfg *lcfg)
3260 {
3261         int rc = class_process_proc_seq_param(PARAM_OSC, obd->obd_vars,
3262                                               lcfg, obd);
3263         return rc > 0 ? 0: rc;
3264 }
3265
3266 static int osc_process_config(struct obd_device *obd, obd_count len, void *buf)
3267 {
3268         return osc_process_config_base(obd, buf);
3269 }
3270
3271 struct obd_ops osc_obd_ops = {
3272         .o_owner                = THIS_MODULE,
3273         .o_setup                = osc_setup,
3274         .o_precleanup           = osc_precleanup,
3275         .o_cleanup              = osc_cleanup,
3276         .o_add_conn             = client_import_add_conn,
3277         .o_del_conn             = client_import_del_conn,
3278         .o_connect              = client_connect_import,
3279         .o_reconnect            = osc_reconnect,
3280         .o_disconnect           = osc_disconnect,
3281         .o_statfs               = osc_statfs,
3282         .o_statfs_async         = osc_statfs_async,
3283         .o_unpackmd             = osc_unpackmd,
3284         .o_create               = osc_create,
3285         .o_destroy              = osc_destroy,
3286         .o_getattr              = osc_getattr,
3287         .o_getattr_async        = osc_getattr_async,
3288         .o_setattr              = osc_setattr,
3289         .o_setattr_async        = osc_setattr_async,
3290         .o_change_cbdata        = osc_change_cbdata,
3291         .o_find_cbdata          = osc_find_cbdata,
3292         .o_iocontrol            = osc_iocontrol,
3293         .o_get_info             = osc_get_info,
3294         .o_set_info_async       = osc_set_info_async,
3295         .o_import_event         = osc_import_event,
3296         .o_process_config       = osc_process_config,
3297         .o_quotactl             = osc_quotactl,
3298         .o_quotacheck           = osc_quotacheck,
3299 };
3300
3301 extern struct lu_kmem_descr osc_caches[];
3302 extern spinlock_t osc_ast_guard;
3303 extern struct lock_class_key osc_ast_guard_class;
3304
3305 int __init osc_init(void)
3306 {
3307         bool enable_proc = true;
3308         struct obd_type *type;
3309         int rc;
3310         ENTRY;
3311
3312         /* print an address of _any_ initialized kernel symbol from this
3313          * module, to allow debugging with gdb that doesn't support data
3314          * symbols from modules.*/
3315         CDEBUG(D_INFO, "Lustre OSC module (%p).\n", &osc_caches);
3316
3317         rc = lu_kmem_init(osc_caches);
3318         if (rc)
3319                 RETURN(rc);
3320
3321         type = class_search_type(LUSTRE_OSP_NAME);
3322         if (type != NULL && type->typ_procsym != NULL)
3323                 enable_proc = false;
3324
3325         rc = class_register_type(&osc_obd_ops, NULL, enable_proc, NULL,
3326 #ifndef HAVE_ONLY_PROCFS_SEQ
3327                                  NULL,
3328 #endif
3329                                  LUSTRE_OSC_NAME, &osc_device_type);
3330         if (rc) {
3331                 lu_kmem_fini(osc_caches);
3332                 RETURN(rc);
3333         }
3334
3335         spin_lock_init(&osc_ast_guard);
3336         lockdep_set_class(&osc_ast_guard, &osc_ast_guard_class);
3337
3338         RETURN(rc);
3339 }
3340
3341 #ifdef __KERNEL__
3342 static void /*__exit*/ osc_exit(void)
3343 {
3344         class_unregister_type(LUSTRE_OSC_NAME);
3345         lu_kmem_fini(osc_caches);
3346 }
3347
3348 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
3349 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC)");
3350 MODULE_LICENSE("GPL");
3351
3352 cfs_module(osc, LUSTRE_VERSION_STRING, osc_init, osc_exit);
3353 #endif