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                 oti->oti_transno = lustre_msg_get_transno(req->rq_repmsg);
 507
 508                 if (oa->o_valid & OBD_MD_FLCOOKIE) {
 509                         if (!oti->oti_logcookies)
 510                                 oti_alloc_cookies(oti, 1);
 511                         *oti->oti_logcookies = oa->o_lcookie;
 512                 }
 513         }
 514
 515         CDEBUG(D_HA, "transno: "LPD64"\n",
 516                lustre_msg_get_transno(req->rq_repmsg));
 517 out_req:
 518         ptlrpc_req_finished(req);
 519 out:
 520         if (rc && !*ea)
 521                 obd_free_memmd(exp, &lsm);
 522         RETURN(rc);
 523 }
 524
 525 int osc_punch_base(struct obd_export *exp, struct obd_info *oinfo,
 526                    obd_enqueue_update_f upcall, void *cookie,
 527                    struct ptlrpc_request_set *rqset)
 528 {
 529         struct ptlrpc_request   *req;
 530         struct osc_setattr_args *sa;
 531         struct ost_body         *body;
 532         int                      rc;
 533         ENTRY;
 534
 535         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_PUNCH);
 536         if (req == NULL)
 537                 RETURN(-ENOMEM);
 538
 539         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
 540         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_PUNCH);
 541         if (rc) {
 542                 ptlrpc_request_free(req);
 543                 RETURN(rc);
 544         }
 545         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
 546         ptlrpc_at_set_req_timeout(req);
 547
 548         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 549         LASSERT(body);
 550         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa,
 551                              oinfo->oi_oa);
 552         osc_pack_capa(req, body, oinfo->oi_capa);
 553
 554         ptlrpc_request_set_replen(req);
 555
 556         req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_setattr_interpret;
 557         CLASSERT (sizeof(*sa) <= sizeof(req->rq_async_args));
 558         sa = ptlrpc_req_async_args(req);
 559         sa->sa_oa     = oinfo->oi_oa;
 560         sa->sa_upcall = upcall;
 561         sa->sa_cookie = cookie;
 562         if (rqset == PTLRPCD_SET)
 563                 ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
 564         else
 565                 ptlrpc_set_add_req(rqset, req);
 566
 567         RETURN(0);
 568 }
 569
 570 static int osc_sync_interpret(const struct lu_env *env,
 571                               struct ptlrpc_request *req,
 572                               void *arg, int rc)
 573 {
 574         struct osc_fsync_args *fa = arg;
 575         struct ost_body *body;
 576         ENTRY;
 577
 578         if (rc)
 579                 GOTO(out, rc);
 580
 581         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 582         if (body == NULL) {
 583                 CERROR ("can't unpack ost_body\n");
 584                 GOTO(out, rc = -EPROTO);
 585         }
 586
 587         *fa->fa_oi->oi_oa = body->oa;
 588 out:
 589         rc = fa->fa_upcall(fa->fa_cookie, rc);
 590         RETURN(rc);
 591 }
 592
 593 int osc_sync_base(struct obd_export *exp, struct obd_info *oinfo,
 594                   obd_enqueue_update_f upcall, void *cookie,
 595                   struct ptlrpc_request_set *rqset)
 596 {
 597         struct ptlrpc_request *req;
 598         struct ost_body       *body;
 599         struct osc_fsync_args *fa;
 600         int                    rc;
 601         ENTRY;
 602
 603         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SYNC);
 604         if (req == NULL)
 605                 RETURN(-ENOMEM);
 606
 607         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
 608         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SYNC);
 609         if (rc) {
 610                 ptlrpc_request_free(req);
 611                 RETURN(rc);
 612         }
 613
 614         /* overload the size and blocks fields in the oa with start/end */
 615         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 616         LASSERT(body);
 617         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa,
 618                              oinfo->oi_oa);
 619         osc_pack_capa(req, body, oinfo->oi_capa);
 620
 621         ptlrpc_request_set_replen(req);
 622         req->rq_interpret_reply = osc_sync_interpret;
 623
 624         CLASSERT(sizeof(*fa) <= sizeof(req->rq_async_args));
 625         fa = ptlrpc_req_async_args(req);
 626         fa->fa_oi = oinfo;
 627         fa->fa_upcall = upcall;
 628         fa->fa_cookie = cookie;
 629
 630         if (rqset == PTLRPCD_SET)
 631                 ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
 632         else
 633                 ptlrpc_set_add_req(rqset, req);
 634
 635         RETURN (0);
 636 }
 637
 638 /* Find and cancel locally locks matched by @mode in the resource found by
 639  * @objid. Found locks are added into @cancel list. Returns the amount of
 640  * locks added to @cancels list. */
 641 static int osc_resource_get_unused(struct obd_export *exp, struct obdo *oa,
 642                                    struct list_head *cancels,
 643                                    ldlm_mode_t mode, __u64 lock_flags)
 644 {
 645         struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
 646         struct ldlm_res_id res_id;
 647         struct ldlm_resource *res;
 648         int count;
 649         ENTRY;
 650
 651         /* Return, i.e. cancel nothing, only if ELC is supported (flag in
 652          * export) but disabled through procfs (flag in NS).
 653          *
 654          * This distinguishes from a case when ELC is not supported originally,
 655          * when we still want to cancel locks in advance and just cancel them
 656          * locally, without sending any RPC. */
 657         if (exp_connect_cancelset(exp) && !ns_connect_cancelset(ns))
 658                 RETURN(0);
 659
 660         ostid_build_res_name(&oa->o_oi, &res_id);
 661         res = ldlm_resource_get(ns, NULL, &res_id, 0, 0);
 662         if (res == NULL)
 663                 RETURN(0);
 664
 665         LDLM_RESOURCE_ADDREF(res);
 666         count = ldlm_cancel_resource_local(res, cancels, NULL, mode,
 667                                            lock_flags, 0, NULL);
 668         LDLM_RESOURCE_DELREF(res);
 669         ldlm_resource_putref(res);
 670         RETURN(count);
 671 }
 672
 673 static int osc_destroy_interpret(const struct lu_env *env,
 674                                  struct ptlrpc_request *req, void *data,
 675                                  int rc)
 676 {
 677         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
 678
 679         atomic_dec(&cli->cl_destroy_in_flight);
 680         wake_up(&cli->cl_destroy_waitq);
 681         return 0;
 682 }
 683
 684 static int osc_can_send_destroy(struct client_obd *cli)
 685 {
 686         if (atomic_inc_return(&cli->cl_destroy_in_flight) <=
 687             cli->cl_max_rpcs_in_flight) {
 688                 /* The destroy request can be sent */
 689                 return 1;
 690         }
 691         if (atomic_dec_return(&cli->cl_destroy_in_flight) <
 692             cli->cl_max_rpcs_in_flight) {
 693                 /*
 694                  * The counter has been modified between the two atomic
 695                  * operations.
 696                  */
 697                 wake_up(&cli->cl_destroy_waitq);
 698         }
 699         return 0;
 700 }
 701
 702 int osc_create(const struct lu_env *env, struct obd_export *exp,
 703                struct obdo *oa, struct lov_stripe_md **ea,
 704                struct obd_trans_info *oti)
 705 {
 706         int rc = 0;
 707         ENTRY;
 708
 709         LASSERT(oa);
 710         LASSERT(ea);
 711         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
 712
 713         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
 714             oa->o_flags == OBD_FL_RECREATE_OBJS) {
 715                 RETURN(osc_real_create(exp, oa, ea, oti));
 716         }
 717
 718         if (!fid_seq_is_mdt(ostid_seq(&oa->o_oi)))
 719                 RETURN(osc_real_create(exp, oa, ea, oti));
 720
 721         /* we should not get here anymore */
 722         LBUG();
 723
 724         RETURN(rc);
 725 }
 726
 727 /* Destroy requests can be async always on the client, and we don't even really
 728  * care about the return code since the client cannot do anything at all about
 729  * a destroy failure.
 730  * When the MDS is unlinking a filename, it saves the file objects into a
 731  * recovery llog, and these object records are cancelled when the OST reports
 732  * they were destroyed and sync'd to disk (i.e. transaction committed).
 733  * If the client dies, or the OST is down when the object should be destroyed,
 734  * the records are not cancelled, and when the OST reconnects to the MDS next,
 735  * it will retrieve the llog unlink logs and then sends the log cancellation
 736  * cookies to the MDS after committing destroy transactions. */
 737 static int osc_destroy(const struct lu_env *env, struct obd_export *exp,
 738                        struct obdo *oa, struct lov_stripe_md *ea,
 739                        struct obd_trans_info *oti, struct obd_export *md_export,
 740                        void *capa)
 741 {
 742         struct client_obd     *cli = &exp->exp_obd->u.cli;
 743         struct ptlrpc_request *req;
 744         struct ost_body       *body;
 745         struct list_head       cancels = LIST_HEAD_INIT(cancels);
 746         int rc, count;
 747         ENTRY;
 748
 749         if (!oa) {
 750                 CDEBUG(D_INFO, "oa NULL\n");
 751                 RETURN(-EINVAL);
 752         }
 753
 754         count = osc_resource_get_unused(exp, oa, &cancels, LCK_PW,
 755                                         LDLM_FL_DISCARD_DATA);
 756
 757         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_DESTROY);
 758         if (req == NULL) {
 759                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 760                 RETURN(-ENOMEM);
 761         }
 762
 763         osc_set_capa_size(req, &RMF_CAPA1, (struct obd_capa *)capa);
 764         rc = ldlm_prep_elc_req(exp, req, LUSTRE_OST_VERSION, OST_DESTROY,
 765                                0, &cancels, count);
 766         if (rc) {
 767                 ptlrpc_request_free(req);
 768                 RETURN(rc);
 769         }
 770
 771         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
 772         ptlrpc_at_set_req_timeout(req);
 773
 774         if (oti != NULL && oa->o_valid & OBD_MD_FLCOOKIE)
 775                 oa->o_lcookie = *oti->oti_logcookies;
 776         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 777         LASSERT(body);
 778         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
 779
 780         osc_pack_capa(req, body, (struct obd_capa *)capa);
 781         ptlrpc_request_set_replen(req);
 782
 783         /* If osc_destory is for destroying the unlink orphan,
 784          * sent from MDT to OST, which should not be blocked here,
 785          * because the process might be triggered by ptlrpcd, and
 786          * it is not good to block ptlrpcd thread (b=16006)*/
 787         if (!(oa->o_flags & OBD_FL_DELORPHAN)) {
 788                 req->rq_interpret_reply = osc_destroy_interpret;
 789                 if (!osc_can_send_destroy(cli)) {
 790                         struct l_wait_info lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP,
 791                                                           NULL);
 792
 793                         /*
 794                          * Wait until the number of on-going destroy RPCs drops
 795                          * under max_rpc_in_flight
 796                          */
 797                         l_wait_event_exclusive(cli->cl_destroy_waitq,
 798                                                osc_can_send_destroy(cli), &lwi);
 799                 }
 800         }
 801
 802         /* Do not wait for response */
 803         ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
 804         RETURN(0);
 805 }
 806
 807 static void osc_announce_cached(struct client_obd *cli, struct obdo *oa,
 808                                 long writing_bytes)
 809 {
 810         obd_flag bits = OBD_MD_FLBLOCKS|OBD_MD_FLGRANT;
 811
 812         LASSERT(!(oa->o_valid & bits));
 813
 814         oa->o_valid |= bits;
 815         client_obd_list_lock(&cli->cl_loi_list_lock);
 816         oa->o_dirty = cli->cl_dirty;
 817         if (unlikely(cli->cl_dirty - cli->cl_dirty_transit >
 818                      cli->cl_dirty_max)) {
 819                 CERROR("dirty %lu - %lu > dirty_max %lu\n",
 820                        cli->cl_dirty, cli->cl_dirty_transit, cli->cl_dirty_max);
 821                 oa->o_undirty = 0;
 822         } else if (unlikely(atomic_read(&obd_unstable_pages) +
 823                             atomic_read(&obd_dirty_pages) -
 824                             atomic_read(&obd_dirty_transit_pages) >
 825                             (long)(obd_max_dirty_pages + 1))) {
 826                 /* The atomic_read() allowing the atomic_inc() are
 827                  * not covered by a lock thus they may safely race and trip
 828                  * this CERROR() unless we add in a small fudge factor (+1). */
 829                 CERROR("%s: dirty %d + %d - %d > system dirty_max %d\n",
 830                        cli->cl_import->imp_obd->obd_name,
 831                        atomic_read(&obd_unstable_pages),
 832                        atomic_read(&obd_dirty_pages),
 833                        atomic_read(&obd_dirty_transit_pages),
 834                        obd_max_dirty_pages);
 835                 oa->o_undirty = 0;
 836         } else if (unlikely(cli->cl_dirty_max - cli->cl_dirty > 0x7fffffff)) {
 837                 CERROR("dirty %lu - dirty_max %lu too big???\n",
 838                        cli->cl_dirty, cli->cl_dirty_max);
 839                 oa->o_undirty = 0;
 840         } else {
 841                 long max_in_flight = (cli->cl_max_pages_per_rpc <<
 842                                       PAGE_CACHE_SHIFT) *
 843                                      (cli->cl_max_rpcs_in_flight + 1);
 844                 oa->o_undirty = max(cli->cl_dirty_max, 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 will drop
1044          * to 0 as inflight RPCs fail out; otherwise, it's avail_grant + dirty.
1045          *
1046          * race is tolerable here: if we're evicted, but imp_state already
1047          * left EVICTED state, then cl_dirty must be 0 already.
1048          */
1049         client_obd_list_lock(&cli->cl_loi_list_lock);
1050         if (cli->cl_import->imp_state == LUSTRE_IMP_EVICTED)
1051                 cli->cl_avail_grant = ocd->ocd_grant;
1052         else
1053                 cli->cl_avail_grant = ocd->ocd_grant - cli->cl_dirty;
1054
1055         if (cli->cl_avail_grant < 0) {
1056                 CWARN("%s: available grant < 0: avail/ocd/dirty %ld/%u/%ld\n",
1057                       cli->cl_import->imp_obd->obd_name, cli->cl_avail_grant,
1058                       ocd->ocd_grant, cli->cl_dirty);
1059                 /* workaround for servers which do not have the patch from
1060                  * LU-2679 */
1061                 cli->cl_avail_grant = ocd->ocd_grant;
1062         }
1063
1064         /* determine the appropriate chunk size used by osc_extent. */
1065         cli->cl_chunkbits = max_t(int, PAGE_CACHE_SHIFT, ocd->ocd_blocksize);
1066         client_obd_list_unlock(&cli->cl_loi_list_lock);
1067
1068         CDEBUG(D_CACHE, "%s, setting cl_avail_grant: %ld cl_lost_grant: %ld."
1069                 "chunk bits: %d.\n", cli->cl_import->imp_obd->obd_name,
1070                 cli->cl_avail_grant, cli->cl_lost_grant, cli->cl_chunkbits);
1071
1072         if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT_SHRINK &&
1073             list_empty(&cli->cl_grant_shrink_list))
1074                 osc_add_shrink_grant(cli);
1075 }
1076
1077 /* We assume that the reason this OSC got a short read is because it read
1078  * beyond the end of a stripe file; i.e. lustre is reading a sparse file
1079  * via the LOV, and it _knows_ it's reading inside the file, it's just that
1080  * this stripe never got written at or beyond this stripe offset yet. */
1081 static void handle_short_read(int nob_read, obd_count page_count,
1082                               struct brw_page **pga)
1083 {
1084         char *ptr;
1085         int i = 0;
1086
1087         /* skip bytes read OK */
1088         while (nob_read > 0) {
1089                 LASSERT (page_count > 0);
1090
1091                 if (pga[i]->count > nob_read) {
1092                         /* EOF inside this page */
1093                         ptr = kmap(pga[i]->pg) +
1094                                 (pga[i]->off & ~CFS_PAGE_MASK);
1095                         memset(ptr + nob_read, 0, pga[i]->count - nob_read);
1096                         kunmap(pga[i]->pg);
1097                         page_count--;
1098                         i++;
1099                         break;
1100                 }
1101
1102                 nob_read -= pga[i]->count;
1103                 page_count--;
1104                 i++;
1105         }
1106
1107         /* zero remaining pages */
1108         while (page_count-- > 0) {
1109                 ptr = kmap(pga[i]->pg) + (pga[i]->off & ~CFS_PAGE_MASK);
1110                 memset(ptr, 0, pga[i]->count);
1111                 kunmap(pga[i]->pg);
1112                 i++;
1113         }
1114 }
1115
1116 static int check_write_rcs(struct ptlrpc_request *req,
1117                            int requested_nob, int niocount,
1118                            obd_count page_count, struct brw_page **pga)
1119 {
1120         int     i;
1121         __u32   *remote_rcs;
1122
1123         remote_rcs = req_capsule_server_sized_get(&req->rq_pill, &RMF_RCS,
1124                                                   sizeof(*remote_rcs) *
1125                                                   niocount);
1126         if (remote_rcs == NULL) {
1127                 CDEBUG(D_INFO, "Missing/short RC vector on BRW_WRITE reply\n");
1128                 return(-EPROTO);
1129         }
1130
1131         /* return error if any niobuf was in error */
1132         for (i = 0; i < niocount; i++) {
1133                 if ((int)remote_rcs[i] < 0)
1134                         return(remote_rcs[i]);
1135
1136                 if (remote_rcs[i] != 0) {
1137                         CDEBUG(D_INFO, "rc[%d] invalid (%d) req %p\n",
1138                                 i, remote_rcs[i], req);
1139                         return(-EPROTO);
1140                 }
1141         }
1142
1143         if (req->rq_bulk->bd_nob_transferred != requested_nob) {
1144                 CERROR("Unexpected # bytes transferred: %d (requested %d)\n",
1145                        req->rq_bulk->bd_nob_transferred, requested_nob);
1146                 return(-EPROTO);
1147         }
1148
1149         return (0);
1150 }
1151
1152 static inline int can_merge_pages(struct brw_page *p1, struct brw_page *p2)
1153 {
1154         if (p1->flag != p2->flag) {
1155                 unsigned mask = ~(OBD_BRW_FROM_GRANT | OBD_BRW_NOCACHE |
1156                                   OBD_BRW_SYNC       | OBD_BRW_ASYNC   |
1157                                   OBD_BRW_NOQUOTA    | OBD_BRW_SOFT_SYNC);
1158
1159                 /* warn if we try to combine flags that we don't know to be
1160                  * safe to combine */
1161                 if (unlikely((p1->flag & mask) != (p2->flag & mask))) {
1162                         CWARN("Saw flags 0x%x and 0x%x in the same brw, please "
1163                               "report this at http://bugs.whamcloud.com/\n",
1164                               p1->flag, p2->flag);
1165                 }
1166                 return 0;
1167         }
1168
1169         return (p1->off + p1->count == p2->off);
1170 }
1171
1172 static obd_count osc_checksum_bulk(int nob, obd_count pg_count,
1173                                    struct brw_page **pga, int opc,
1174                                    cksum_type_t cksum_type)
1175 {
1176         __u32                           cksum;
1177         int                             i = 0;
1178         struct cfs_crypto_hash_desc     *hdesc;
1179         unsigned int                    bufsize;
1180         int                             err;
1181         unsigned char                   cfs_alg = cksum_obd2cfs(cksum_type);
1182
1183         LASSERT(pg_count > 0);
1184
1185         hdesc = cfs_crypto_hash_init(cfs_alg, NULL, 0);
1186         if (IS_ERR(hdesc)) {
1187                 CERROR("Unable to initialize checksum hash %s\n",
1188                        cfs_crypto_hash_name(cfs_alg));
1189                 return PTR_ERR(hdesc);
1190         }
1191
1192         while (nob > 0 && pg_count > 0) {
1193                 int count = pga[i]->count > nob ? nob : pga[i]->count;
1194
1195                 /* corrupt the data before we compute the checksum, to
1196                  * simulate an OST->client data error */
1197                 if (i == 0 && opc == OST_READ &&
1198                     OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_RECEIVE)) {
1199                         unsigned char *ptr = kmap(pga[i]->pg);
1200                         int off = pga[i]->off & ~CFS_PAGE_MASK;
1201                         memcpy(ptr + off, "bad1", min(4, nob));
1202                         kunmap(pga[i]->pg);
1203                 }
1204                 cfs_crypto_hash_update_page(hdesc, pga[i]->pg,
1205                                   pga[i]->off & ~CFS_PAGE_MASK,
1206                                   count);
1207                 LL_CDEBUG_PAGE(D_PAGE, pga[i]->pg, "off %d\n",
1208                                (int)(pga[i]->off & ~CFS_PAGE_MASK));
1209
1210                 nob -= pga[i]->count;
1211                 pg_count--;
1212                 i++;
1213         }
1214
1215         bufsize = 4;
1216         err = cfs_crypto_hash_final(hdesc, (unsigned char *)&cksum, &bufsize);
1217
1218         if (err)
1219                 cfs_crypto_hash_final(hdesc, NULL, NULL);
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->len += pg->count;
1348                 } else {
1349                         niobuf->offset = pg->off;
1350                         niobuf->len    = pg->count;
1351                         niobuf->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
1818                         /* Change file size if this is an out of quota or
1819                          * direct IO write and it extends the file size */
1820                         if (loi->loi_lvb.lvb_size < last_off) {
1821                                 attr->cat_size = last_off;
1822                                 valid |= CAT_SIZE;
1823                         }
1824                         /* Extend KMS if it's not a lockless write */
1825                         if (loi->loi_kms < last_off &&
1826                             oap2osc_page(last)->ops_srvlock == 0) {
1827                                 attr->cat_kms = last_off;
1828                                 valid |= CAT_KMS;
1829                         }
1830                 }
1831
1832                 if (valid != 0)
1833                         cl_object_attr_set(env, obj, attr, valid);
1834                 cl_object_attr_unlock(obj);
1835         }
1836         OBDO_FREE(aa->aa_oa);
1837
1838         list_for_each_entry_safe(ext, tmp, &aa->aa_exts, oe_link) {
1839                 list_del_init(&ext->oe_link);
1840                 osc_extent_finish(env, ext, 1, rc);
1841         }
1842         LASSERT(list_empty(&aa->aa_exts));
1843         LASSERT(list_empty(&aa->aa_oaps));
1844
1845         cl_req_completion(env, aa->aa_clerq, rc < 0 ? rc :
1846                           req->rq_bulk->bd_nob_transferred);
1847         osc_release_ppga(aa->aa_ppga, aa->aa_page_count);
1848         ptlrpc_lprocfs_brw(req, req->rq_bulk->bd_nob_transferred);
1849
1850         client_obd_list_lock(&cli->cl_loi_list_lock);
1851         /* We need to decrement before osc_ap_completion->osc_wake_cache_waiters
1852          * is called so we know whether to go to sync BRWs or wait for more
1853          * RPCs to complete */
1854         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE)
1855                 cli->cl_w_in_flight--;
1856         else
1857                 cli->cl_r_in_flight--;
1858         osc_wake_cache_waiters(cli);
1859         client_obd_list_unlock(&cli->cl_loi_list_lock);
1860
1861         osc_io_unplug(env, cli, NULL, PDL_POLICY_SAME);
1862         RETURN(rc);
1863 }
1864
1865 static void brw_commit(struct ptlrpc_request *req)
1866 {
1867         /* If osc_inc_unstable_pages (via osc_extent_finish) races with
1868          * this called via the rq_commit_cb, I need to ensure
1869          * osc_dec_unstable_pages is still called. Otherwise unstable
1870          * pages may be leaked. */
1871         spin_lock(&req->rq_lock);
1872         if (likely(req->rq_unstable)) {
1873                 req->rq_unstable = 0;
1874                 spin_unlock(&req->rq_lock);
1875
1876                 osc_dec_unstable_pages(req);
1877         } else {
1878                 req->rq_committed = 1;
1879                 spin_unlock(&req->rq_lock);
1880         }
1881 }
1882
1883 /**
1884  * Build an RPC by the list of extent @ext_list. The caller must ensure
1885  * that the total pages in this list are NOT over max pages per RPC.
1886  * Extents in the list must be in OES_RPC state.
1887  */
1888 int osc_build_rpc(const struct lu_env *env, struct client_obd *cli,
1889                   struct list_head *ext_list, int cmd, pdl_policy_t pol)
1890 {
1891         struct ptlrpc_request           *req = NULL;
1892         struct osc_extent               *ext;
1893         struct brw_page                 **pga = NULL;
1894         struct osc_brw_async_args       *aa = NULL;
1895         struct obdo                     *oa = NULL;
1896         struct osc_async_page           *oap;
1897         struct osc_async_page           *tmp;
1898         struct cl_req                   *clerq = NULL;
1899         enum cl_req_type                crt = (cmd & OBD_BRW_WRITE) ? CRT_WRITE :
1900                                                                       CRT_READ;
1901         struct ldlm_lock                *lock = NULL;
1902         struct cl_req_attr              *crattr = NULL;
1903         obd_off                         starting_offset = OBD_OBJECT_EOF;
1904         obd_off                         ending_offset = 0;
1905         int                             mpflag = 0;
1906         int                             mem_tight = 0;
1907         int                             page_count = 0;
1908         int                             i;
1909         int                             rc;
1910         struct list_head                rpc_list = LIST_HEAD_INIT(rpc_list);
1911
1912         ENTRY;
1913         LASSERT(!list_empty(ext_list));
1914
1915         /* add pages into rpc_list to build BRW rpc */
1916         list_for_each_entry(ext, ext_list, oe_link) {
1917                 LASSERT(ext->oe_state == OES_RPC);
1918                 mem_tight |= ext->oe_memalloc;
1919                 list_for_each_entry(oap, &ext->oe_pages, oap_pending_item) {
1920                         ++page_count;
1921                         list_add_tail(&oap->oap_rpc_item, &rpc_list);
1922                         if (starting_offset > oap->oap_obj_off)
1923                                 starting_offset = oap->oap_obj_off;
1924                         else
1925                                 LASSERT(oap->oap_page_off == 0);
1926                         if (ending_offset < oap->oap_obj_off + oap->oap_count)
1927                                 ending_offset = oap->oap_obj_off +
1928                                                 oap->oap_count;
1929                         else
1930                                 LASSERT(oap->oap_page_off + oap->oap_count ==
1931                                         PAGE_CACHE_SIZE);
1932                 }
1933         }
1934
1935         if (mem_tight)
1936                 mpflag = cfs_memory_pressure_get_and_set();
1937
1938         OBD_ALLOC(crattr, sizeof(*crattr));
1939         if (crattr == NULL)
1940                 GOTO(out, rc = -ENOMEM);
1941
1942         OBD_ALLOC(pga, sizeof(*pga) * page_count);
1943         if (pga == NULL)
1944                 GOTO(out, rc = -ENOMEM);
1945
1946         OBDO_ALLOC(oa);
1947         if (oa == NULL)
1948                 GOTO(out, rc = -ENOMEM);
1949
1950         i = 0;
1951         list_for_each_entry(oap, &rpc_list, oap_rpc_item) {
1952                 struct cl_page *page = oap2cl_page(oap);
1953                 if (clerq == NULL) {
1954                         clerq = cl_req_alloc(env, page, crt,
1955                                              1 /* only 1-object rpcs for now */);
1956                         if (IS_ERR(clerq))
1957                                 GOTO(out, rc = PTR_ERR(clerq));
1958                         lock = oap->oap_ldlm_lock;
1959                 }
1960                 if (mem_tight)
1961                         oap->oap_brw_flags |= OBD_BRW_MEMALLOC;
1962                 pga[i] = &oap->oap_brw_page;
1963                 pga[i]->off = oap->oap_obj_off + oap->oap_page_off;
1964                 CDEBUG(0, "put page %p index %lu oap %p flg %x to pga\n",
1965                        pga[i]->pg, page_index(oap->oap_page), oap,
1966                        pga[i]->flag);
1967                 i++;
1968                 cl_req_page_add(env, clerq, page);
1969         }
1970
1971         /* always get the data for the obdo for the rpc */
1972         LASSERT(clerq != NULL);
1973         crattr->cra_oa = oa;
1974         cl_req_attr_set(env, clerq, crattr, ~0ULL);
1975         if (lock) {
1976                 oa->o_handle = lock->l_remote_handle;
1977                 oa->o_valid |= OBD_MD_FLHANDLE;
1978         }
1979
1980         rc = cl_req_prep(env, clerq);
1981         if (rc != 0) {
1982                 CERROR("cl_req_prep failed: %d\n", rc);
1983                 GOTO(out, rc);
1984         }
1985
1986         sort_brw_pages(pga, page_count);
1987         rc = osc_brw_prep_request(cmd, cli, oa, NULL, page_count,
1988                         pga, &req, crattr->cra_capa, 1, 0);
1989         if (rc != 0) {
1990                 CERROR("prep_req failed: %d\n", rc);
1991                 GOTO(out, rc);
1992         }
1993
1994         req->rq_commit_cb = brw_commit;
1995         req->rq_interpret_reply = brw_interpret;
1996
1997         if (mem_tight != 0)
1998                 req->rq_memalloc = 1;
1999
2000         /* Need to update the timestamps after the request is built in case
2001          * we race with setattr (locally or in queue at OST).  If OST gets
2002          * later setattr before earlier BRW (as determined by the request xid),
2003          * the OST will not use BRW timestamps.  Sadly, there is no obvious
2004          * way to do this in a single call.  bug 10150 */
2005         cl_req_attr_set(env, clerq, crattr,
2006                         OBD_MD_FLMTIME|OBD_MD_FLCTIME|OBD_MD_FLATIME);
2007
2008         lustre_msg_set_jobid(req->rq_reqmsg, crattr->cra_jobid);
2009
2010         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
2011         aa = ptlrpc_req_async_args(req);
2012         INIT_LIST_HEAD(&aa->aa_oaps);
2013         list_splice_init(&rpc_list, &aa->aa_oaps);
2014         INIT_LIST_HEAD(&aa->aa_exts);
2015         list_splice_init(ext_list, &aa->aa_exts);
2016         aa->aa_clerq = clerq;
2017
2018         /* queued sync pages can be torn down while the pages
2019          * were between the pending list and the rpc */
2020         tmp = NULL;
2021         list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
2022                 /* only one oap gets a request reference */
2023                 if (tmp == NULL)
2024                         tmp = oap;
2025                 if (oap->oap_interrupted && !req->rq_intr) {
2026                         CDEBUG(D_INODE, "oap %p in req %p interrupted\n",
2027                                         oap, req);
2028                         ptlrpc_mark_interrupted(req);
2029                 }
2030         }
2031         if (tmp != NULL)
2032                 tmp->oap_request = ptlrpc_request_addref(req);
2033
2034         client_obd_list_lock(&cli->cl_loi_list_lock);
2035         starting_offset >>= PAGE_CACHE_SHIFT;
2036         if (cmd == OBD_BRW_READ) {
2037                 cli->cl_r_in_flight++;
2038                 lprocfs_oh_tally_log2(&cli->cl_read_page_hist, page_count);
2039                 lprocfs_oh_tally(&cli->cl_read_rpc_hist, cli->cl_r_in_flight);
2040                 lprocfs_oh_tally_log2(&cli->cl_read_offset_hist,
2041                                       starting_offset + 1);
2042         } else {
2043                 cli->cl_w_in_flight++;
2044                 lprocfs_oh_tally_log2(&cli->cl_write_page_hist, page_count);
2045                 lprocfs_oh_tally(&cli->cl_write_rpc_hist, cli->cl_w_in_flight);
2046                 lprocfs_oh_tally_log2(&cli->cl_write_offset_hist,
2047                                       starting_offset + 1);
2048         }
2049         client_obd_list_unlock(&cli->cl_loi_list_lock);
2050
2051         DEBUG_REQ(D_INODE, req, "%d pages, aa %p. now %dr/%dw in flight",
2052                   page_count, aa, cli->cl_r_in_flight,
2053                   cli->cl_w_in_flight);
2054
2055         /* XXX: Maybe the caller can check the RPC bulk descriptor to
2056          * see which CPU/NUMA node the majority of pages were allocated
2057          * on, and try to assign the async RPC to the CPU core
2058          * (PDL_POLICY_PREFERRED) to reduce cross-CPU memory traffic.
2059          *
2060          * But on the other hand, we expect that multiple ptlrpcd
2061          * threads and the initial write sponsor can run in parallel,
2062          * especially when data checksum is enabled, which is CPU-bound
2063          * operation and single ptlrpcd thread cannot process in time.
2064          * So more ptlrpcd threads sharing BRW load
2065          * (with PDL_POLICY_ROUND) seems better.
2066          */
2067         ptlrpcd_add_req(req, pol, -1);
2068         rc = 0;
2069         EXIT;
2070
2071 out:
2072         if (mem_tight != 0)
2073                 cfs_memory_pressure_restore(mpflag);
2074
2075         if (crattr != NULL) {
2076                 capa_put(crattr->cra_capa);
2077                 OBD_FREE(crattr, sizeof(*crattr));
2078         }
2079
2080         if (rc != 0) {
2081                 LASSERT(req == NULL);
2082
2083                 if (oa)
2084                         OBDO_FREE(oa);
2085                 if (pga)
2086                         OBD_FREE(pga, sizeof(*pga) * page_count);
2087                 /* this should happen rarely and is pretty bad, it makes the
2088                  * pending list not follow the dirty order */
2089                 while (!list_empty(ext_list)) {
2090                         ext = list_entry(ext_list->next, struct osc_extent,
2091                                          oe_link);
2092                         list_del_init(&ext->oe_link);
2093                         osc_extent_finish(env, ext, 0, rc);
2094                 }
2095                 if (clerq && !IS_ERR(clerq))
2096                         cl_req_completion(env, clerq, rc);
2097         }
2098         RETURN(rc);
2099 }
2100
2101 static int osc_set_lock_data_with_check(struct ldlm_lock *lock,
2102                                         struct ldlm_enqueue_info *einfo)
2103 {
2104         void *data = einfo->ei_cbdata;
2105         int set = 0;
2106
2107         LASSERT(lock != NULL);
2108         LASSERT(lock->l_blocking_ast == einfo->ei_cb_bl);
2109         LASSERT(lock->l_resource->lr_type == einfo->ei_type);
2110         LASSERT(lock->l_completion_ast == einfo->ei_cb_cp);
2111         LASSERT(lock->l_glimpse_ast == einfo->ei_cb_gl);
2112
2113         lock_res_and_lock(lock);
2114         spin_lock(&osc_ast_guard);
2115
2116         if (lock->l_ast_data == NULL)
2117                 lock->l_ast_data = data;
2118         if (lock->l_ast_data == data)
2119                 set = 1;
2120
2121         spin_unlock(&osc_ast_guard);
2122         unlock_res_and_lock(lock);
2123
2124         return set;
2125 }
2126
2127 static int osc_set_data_with_check(struct lustre_handle *lockh,
2128                                    struct ldlm_enqueue_info *einfo)
2129 {
2130         struct ldlm_lock *lock = ldlm_handle2lock(lockh);
2131         int set = 0;
2132
2133         if (lock != NULL) {
2134                 set = osc_set_lock_data_with_check(lock, einfo);
2135                 LDLM_LOCK_PUT(lock);
2136         } else
2137                 CERROR("lockh %p, data %p - client evicted?\n",
2138                        lockh, einfo->ei_cbdata);
2139         return set;
2140 }
2141
2142 static int osc_change_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
2143                              ldlm_iterator_t replace, void *data)
2144 {
2145         struct ldlm_res_id res_id;
2146         struct obd_device *obd = class_exp2obd(exp);
2147
2148         ostid_build_res_name(&lsm->lsm_oi, &res_id);
2149         ldlm_resource_iterate(obd->obd_namespace, &res_id, replace, data);
2150         return 0;
2151 }
2152
2153 /* find any ldlm lock of the inode in osc
2154  * return 0    not find
2155  *        1    find one
2156  *      < 0    error */
2157 static int osc_find_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
2158                            ldlm_iterator_t replace, void *data)
2159 {
2160         struct ldlm_res_id res_id;
2161         struct obd_device *obd = class_exp2obd(exp);
2162         int rc = 0;
2163
2164         ostid_build_res_name(&lsm->lsm_oi, &res_id);
2165         rc = ldlm_resource_iterate(obd->obd_namespace, &res_id, replace, data);
2166         if (rc == LDLM_ITER_STOP)
2167                 return(1);
2168         if (rc == LDLM_ITER_CONTINUE)
2169                 return(0);
2170         return(rc);
2171 }
2172
2173 static int osc_enqueue_fini(struct ptlrpc_request *req, struct ost_lvb *lvb,
2174                             obd_enqueue_update_f upcall, void *cookie,
2175                             __u64 *flags, int agl, int rc)
2176 {
2177         int intent = *flags & LDLM_FL_HAS_INTENT;
2178         ENTRY;
2179
2180         if (intent) {
2181                 /* The request was created before ldlm_cli_enqueue call. */
2182                 if (rc == ELDLM_LOCK_ABORTED) {
2183                         struct ldlm_reply *rep;
2184                         rep = req_capsule_server_get(&req->rq_pill,
2185                                                      &RMF_DLM_REP);
2186
2187                         LASSERT(rep != NULL);
2188                         rep->lock_policy_res1 =
2189                                 ptlrpc_status_ntoh(rep->lock_policy_res1);
2190                         if (rep->lock_policy_res1)
2191                                 rc = rep->lock_policy_res1;
2192                 }
2193         }
2194
2195         if ((intent != 0 && rc == ELDLM_LOCK_ABORTED && agl == 0) ||
2196             (rc == 0)) {
2197                 *flags |= LDLM_FL_LVB_READY;
2198                 CDEBUG(D_INODE,"got kms "LPU64" blocks "LPU64" mtime "LPU64"\n",
2199                        lvb->lvb_size, lvb->lvb_blocks, lvb->lvb_mtime);
2200         }
2201
2202         /* Call the update callback. */
2203         rc = (*upcall)(cookie, rc);
2204         RETURN(rc);
2205 }
2206
2207 static int osc_enqueue_interpret(const struct lu_env *env,
2208                                  struct ptlrpc_request *req,
2209                                  struct osc_enqueue_args *aa, int rc)
2210 {
2211         struct ldlm_lock *lock;
2212         struct lustre_handle handle;
2213         __u32 mode;
2214         struct ost_lvb *lvb;
2215         __u32 lvb_len;
2216         __u64 *flags = aa->oa_flags;
2217
2218         /* Make a local copy of a lock handle and a mode, because aa->oa_*
2219          * might be freed anytime after lock upcall has been called. */
2220         lustre_handle_copy(&handle, aa->oa_lockh);
2221         mode = aa->oa_ei->ei_mode;
2222
2223         /* ldlm_cli_enqueue is holding a reference on the lock, so it must
2224          * be valid. */
2225         lock = ldlm_handle2lock(&handle);
2226
2227         /* Take an additional reference so that a blocking AST that
2228          * ldlm_cli_enqueue_fini() might post for a failed lock, is guaranteed
2229          * to arrive after an upcall has been executed by
2230          * osc_enqueue_fini(). */
2231         ldlm_lock_addref(&handle, mode);
2232
2233         /* Let cl_lock_state_wait fail with -ERESTARTSYS to unuse sublocks. */
2234         OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_ENQUEUE_HANG, 2);
2235
2236         /* Let CP AST to grant the lock first. */
2237         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_ENQ_RACE, 1);
2238
2239         if (aa->oa_agl && rc == ELDLM_LOCK_ABORTED) {
2240                 lvb = NULL;
2241                 lvb_len = 0;
2242         } else {
2243                 lvb = aa->oa_lvb;
2244                 lvb_len = sizeof(*aa->oa_lvb);
2245         }
2246
2247         /* Complete obtaining the lock procedure. */
2248         rc = ldlm_cli_enqueue_fini(aa->oa_exp, req, aa->oa_ei->ei_type, 1,
2249                                    mode, flags, lvb, lvb_len, &handle, rc);
2250         /* Complete osc stuff. */
2251         rc = osc_enqueue_fini(req, aa->oa_lvb, aa->oa_upcall, aa->oa_cookie,
2252                               flags, aa->oa_agl, rc);
2253
2254         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_CANCEL_RACE, 10);
2255
2256         /* Release the lock for async request. */
2257         if (lustre_handle_is_used(&handle) && rc == ELDLM_OK)
2258                 /*
2259                  * Releases a reference taken by ldlm_cli_enqueue(), if it is
2260                  * not already released by
2261                  * ldlm_cli_enqueue_fini()->failed_lock_cleanup()
2262                  */
2263                 ldlm_lock_decref(&handle, mode);
2264
2265         LASSERTF(lock != NULL, "lockh %p, req %p, aa %p - client evicted?\n",
2266                  aa->oa_lockh, req, aa);
2267         ldlm_lock_decref(&handle, mode);
2268         LDLM_LOCK_PUT(lock);
2269         return rc;
2270 }
2271
2272 struct ptlrpc_request_set *PTLRPCD_SET = (void *)1;
2273
2274 /* When enqueuing asynchronously, locks are not ordered, we can obtain a lock
2275  * from the 2nd OSC before a lock from the 1st one. This does not deadlock with
2276  * other synchronous requests, however keeping some locks and trying to obtain
2277  * others may take a considerable amount of time in a case of ost failure; and
2278  * when other sync requests do not get released lock from a client, the client
2279  * is excluded from the cluster -- such scenarious make the life difficult, so
2280  * release locks just after they are obtained. */
2281 int osc_enqueue_base(struct obd_export *exp, struct ldlm_res_id *res_id,
2282                      __u64 *flags, ldlm_policy_data_t *policy,
2283                      struct ost_lvb *lvb, int kms_valid,
2284                      obd_enqueue_update_f upcall, void *cookie,
2285                      struct ldlm_enqueue_info *einfo,
2286                      struct lustre_handle *lockh,
2287                      struct ptlrpc_request_set *rqset, int async, int agl)
2288 {
2289         struct obd_device *obd = exp->exp_obd;
2290         struct ptlrpc_request *req = NULL;
2291         int intent = *flags & LDLM_FL_HAS_INTENT;
2292         __u64 match_lvb = (agl != 0 ? 0 : LDLM_FL_LVB_READY);
2293         ldlm_mode_t mode;
2294         int rc;
2295         ENTRY;
2296
2297         /* Filesystem lock extents are extended to page boundaries so that
2298          * dealing with the page cache is a little smoother.  */
2299         policy->l_extent.start -= policy->l_extent.start & ~CFS_PAGE_MASK;
2300         policy->l_extent.end |= ~CFS_PAGE_MASK;
2301
2302         /*
2303          * kms is not valid when either object is completely fresh (so that no
2304          * locks are cached), or object was evicted. In the latter case cached
2305          * lock cannot be used, because it would prime inode state with
2306          * potentially stale LVB.
2307          */
2308         if (!kms_valid)
2309                 goto no_match;
2310
2311         /* Next, search for already existing extent locks that will cover us */
2312         /* If we're trying to read, we also search for an existing PW lock.  The
2313          * VFS and page cache already protect us locally, so lots of readers/
2314          * writers can share a single PW lock.
2315          *
2316          * There are problems with conversion deadlocks, so instead of
2317          * converting a read lock to a write lock, we'll just enqueue a new
2318          * one.
2319          *
2320          * At some point we should cancel the read lock instead of making them
2321          * send us a blocking callback, but there are problems with canceling
2322          * locks out from other users right now, too. */
2323         mode = einfo->ei_mode;
2324         if (einfo->ei_mode == LCK_PR)
2325                 mode |= LCK_PW;
2326         mode = ldlm_lock_match(obd->obd_namespace, *flags | match_lvb, res_id,
2327                                einfo->ei_type, policy, mode, lockh, 0);
2328         if (mode) {
2329                 struct ldlm_lock *matched = ldlm_handle2lock(lockh);
2330
2331                 if ((agl != 0) && !ldlm_is_lvb_ready(matched)) {
2332                         /* For AGL, if enqueue RPC is sent but the lock is not
2333                          * granted, then skip to process this strpe.
2334                          * Return -ECANCELED to tell the caller. */
2335                         ldlm_lock_decref(lockh, mode);
2336                         LDLM_LOCK_PUT(matched);
2337                         RETURN(-ECANCELED);
2338                 } else if (osc_set_lock_data_with_check(matched, einfo)) {
2339                         *flags |= LDLM_FL_LVB_READY;
2340                         /* addref the lock only if not async requests and PW
2341                          * lock is matched whereas we asked for PR. */
2342                         if (!rqset && einfo->ei_mode != mode)
2343                                 ldlm_lock_addref(lockh, LCK_PR);
2344                         if (intent) {
2345                                 /* I would like to be able to ASSERT here that
2346                                  * rss <= kms, but I can't, for reasons which
2347                                  * are explained in lov_enqueue() */
2348                         }
2349
2350                         /* We already have a lock, and it's referenced.
2351                          *
2352                          * At this point, the cl_lock::cll_state is CLS_QUEUING,
2353                          * AGL upcall may change it to CLS_HELD directly. */
2354                         (*upcall)(cookie, ELDLM_OK);
2355
2356                         if (einfo->ei_mode != mode)
2357                                 ldlm_lock_decref(lockh, LCK_PW);
2358                         else if (rqset)
2359                                 /* For async requests, decref the lock. */
2360                                 ldlm_lock_decref(lockh, einfo->ei_mode);
2361                         LDLM_LOCK_PUT(matched);
2362                         RETURN(ELDLM_OK);
2363                 } else {
2364                         ldlm_lock_decref(lockh, mode);
2365                         LDLM_LOCK_PUT(matched);
2366                 }
2367         }
2368
2369  no_match:
2370         if (intent) {
2371                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
2372                                            &RQF_LDLM_ENQUEUE_LVB);
2373                 if (req == NULL)
2374                         RETURN(-ENOMEM);
2375
2376                 rc = ptlrpc_request_pack(req, LUSTRE_DLM_VERSION, LDLM_ENQUEUE);
2377                 if (rc < 0) {
2378                         ptlrpc_request_free(req);
2379                         RETURN(rc);
2380                 }
2381
2382                 req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
2383                                      sizeof *lvb);
2384                 ptlrpc_request_set_replen(req);
2385         }
2386
2387         /* users of osc_enqueue() can pass this flag for ldlm_lock_match() */
2388         *flags &= ~LDLM_FL_BLOCK_GRANTED;
2389
2390         rc = ldlm_cli_enqueue(exp, &req, einfo, res_id, policy, flags, lvb,
2391                               sizeof(*lvb), LVB_T_OST, lockh, async);
2392         if (rqset) {
2393                 if (!rc) {
2394                         struct osc_enqueue_args *aa;
2395                         CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
2396                         aa = ptlrpc_req_async_args(req);
2397                         aa->oa_ei = einfo;
2398                         aa->oa_exp = exp;
2399                         aa->oa_flags  = flags;
2400                         aa->oa_upcall = upcall;
2401                         aa->oa_cookie = cookie;
2402                         aa->oa_lvb    = lvb;
2403                         aa->oa_lockh  = lockh;
2404                         aa->oa_agl    = !!agl;
2405
2406                         req->rq_interpret_reply =
2407                                 (ptlrpc_interpterer_t)osc_enqueue_interpret;
2408                         if (rqset == PTLRPCD_SET)
2409                                 ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
2410                         else
2411                                 ptlrpc_set_add_req(rqset, req);
2412                 } else if (intent) {
2413                         ptlrpc_req_finished(req);
2414                 }
2415                 RETURN(rc);
2416         }
2417
2418         rc = osc_enqueue_fini(req, lvb, upcall, cookie, flags, agl, rc);
2419         if (intent)
2420                 ptlrpc_req_finished(req);
2421
2422         RETURN(rc);
2423 }
2424
2425 int osc_match_base(struct obd_export *exp, struct ldlm_res_id *res_id,
2426                    __u32 type, ldlm_policy_data_t *policy, __u32 mode,
2427                    __u64 *flags, void *data, struct lustre_handle *lockh,
2428                    int unref)
2429 {
2430         struct obd_device *obd = exp->exp_obd;
2431         __u64 lflags = *flags;
2432         ldlm_mode_t rc;
2433         ENTRY;
2434
2435         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_MATCH))
2436                 RETURN(-EIO);
2437
2438         /* Filesystem lock extents are extended to page boundaries so that
2439          * dealing with the page cache is a little smoother */
2440         policy->l_extent.start -= policy->l_extent.start & ~CFS_PAGE_MASK;
2441         policy->l_extent.end |= ~CFS_PAGE_MASK;
2442
2443         /* Next, search for already existing extent locks that will cover us */
2444         /* If we're trying to read, we also search for an existing PW lock.  The
2445          * VFS and page cache already protect us locally, so lots of readers/
2446          * writers can share a single PW lock. */
2447         rc = mode;
2448         if (mode == LCK_PR)
2449                 rc |= LCK_PW;
2450         rc = ldlm_lock_match(obd->obd_namespace, lflags,
2451                              res_id, type, policy, rc, lockh, unref);
2452         if (rc) {
2453                 if (data != NULL) {
2454                         if (!osc_set_data_with_check(lockh, data)) {
2455                                 if (!(lflags & LDLM_FL_TEST_LOCK))
2456                                         ldlm_lock_decref(lockh, rc);
2457                                 RETURN(0);
2458                         }
2459                 }
2460                 if (!(lflags & LDLM_FL_TEST_LOCK) && mode != rc) {
2461                         ldlm_lock_addref(lockh, LCK_PR);
2462                         ldlm_lock_decref(lockh, LCK_PW);
2463                 }
2464                 RETURN(rc);
2465         }
2466         RETURN(rc);
2467 }
2468
2469 int osc_cancel_base(struct lustre_handle *lockh, __u32 mode)
2470 {
2471         ENTRY;
2472
2473         if (unlikely(mode == LCK_GROUP))
2474                 ldlm_lock_decref_and_cancel(lockh, mode);
2475         else
2476                 ldlm_lock_decref(lockh, mode);
2477
2478         RETURN(0);
2479 }
2480
2481 static int osc_statfs_interpret(const struct lu_env *env,
2482                                 struct ptlrpc_request *req,
2483                                 struct osc_async_args *aa, int rc)
2484 {
2485         struct obd_statfs *msfs;
2486         ENTRY;
2487
2488         if (rc == -EBADR)
2489                 /* The request has in fact never been sent
2490                  * due to issues at a higher level (LOV).
2491                  * Exit immediately since the caller is
2492                  * aware of the problem and takes care
2493                  * of the clean up */
2494                  RETURN(rc);
2495
2496         if ((rc == -ENOTCONN || rc == -EAGAIN) &&
2497             (aa->aa_oi->oi_flags & OBD_STATFS_NODELAY))
2498                 GOTO(out, rc = 0);
2499
2500         if (rc != 0)
2501                 GOTO(out, rc);
2502
2503         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
2504         if (msfs == NULL) {
2505                 GOTO(out, rc = -EPROTO);
2506         }
2507
2508         *aa->aa_oi->oi_osfs = *msfs;
2509 out:
2510         rc = aa->aa_oi->oi_cb_up(aa->aa_oi, rc);
2511         RETURN(rc);
2512 }
2513
2514 static int osc_statfs_async(struct obd_export *exp,
2515                             struct obd_info *oinfo, __u64 max_age,
2516                             struct ptlrpc_request_set *rqset)
2517 {
2518         struct obd_device     *obd = class_exp2obd(exp);
2519         struct ptlrpc_request *req;
2520         struct osc_async_args *aa;
2521         int                    rc;
2522         ENTRY;
2523
2524         /* We could possibly pass max_age in the request (as an absolute
2525          * timestamp or a "seconds.usec ago") so the target can avoid doing
2526          * extra calls into the filesystem if that isn't necessary (e.g.
2527          * during mount that would help a bit).  Having relative timestamps
2528          * is not so great if request processing is slow, while absolute
2529          * timestamps are not ideal because they need time synchronization. */
2530         req = ptlrpc_request_alloc(obd->u.cli.cl_import, &RQF_OST_STATFS);
2531         if (req == NULL)
2532                 RETURN(-ENOMEM);
2533
2534         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
2535         if (rc) {
2536                 ptlrpc_request_free(req);
2537                 RETURN(rc);
2538         }
2539         ptlrpc_request_set_replen(req);
2540         req->rq_request_portal = OST_CREATE_PORTAL;
2541         ptlrpc_at_set_req_timeout(req);
2542
2543         if (oinfo->oi_flags & OBD_STATFS_NODELAY) {
2544                 /* procfs requests not want stat in wait for avoid deadlock */
2545                 req->rq_no_resend = 1;
2546                 req->rq_no_delay = 1;
2547         }
2548
2549         req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_statfs_interpret;
2550         CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
2551         aa = ptlrpc_req_async_args(req);
2552         aa->aa_oi = oinfo;
2553
2554         ptlrpc_set_add_req(rqset, req);
2555         RETURN(0);
2556 }
2557
2558 static int osc_statfs(const struct lu_env *env, struct obd_export *exp,
2559                       struct obd_statfs *osfs, __u64 max_age, __u32 flags)
2560 {
2561         struct obd_device     *obd = class_exp2obd(exp);
2562         struct obd_statfs     *msfs;
2563         struct ptlrpc_request *req;
2564         struct obd_import     *imp = NULL;
2565         int rc;
2566         ENTRY;
2567
2568         /*Since the request might also come from lprocfs, so we need
2569          *sync this with client_disconnect_export Bug15684*/
2570         down_read(&obd->u.cli.cl_sem);
2571         if (obd->u.cli.cl_import)
2572                 imp = class_import_get(obd->u.cli.cl_import);
2573         up_read(&obd->u.cli.cl_sem);
2574         if (!imp)
2575                 RETURN(-ENODEV);
2576
2577         /* We could possibly pass max_age in the request (as an absolute
2578          * timestamp or a "seconds.usec ago") so the target can avoid doing
2579          * extra calls into the filesystem if that isn't necessary (e.g.
2580          * during mount that would help a bit).  Having relative timestamps
2581          * is not so great if request processing is slow, while absolute
2582          * timestamps are not ideal because they need time synchronization. */
2583         req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
2584
2585         class_import_put(imp);
2586
2587         if (req == NULL)
2588                 RETURN(-ENOMEM);
2589
2590         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
2591         if (rc) {
2592                 ptlrpc_request_free(req);
2593                 RETURN(rc);
2594         }
2595         ptlrpc_request_set_replen(req);
2596         req->rq_request_portal = OST_CREATE_PORTAL;
2597         ptlrpc_at_set_req_timeout(req);
2598
2599         if (flags & OBD_STATFS_NODELAY) {
2600                 /* procfs requests not want stat in wait for avoid deadlock */
2601                 req->rq_no_resend = 1;
2602                 req->rq_no_delay = 1;
2603         }
2604
2605         rc = ptlrpc_queue_wait(req);
2606         if (rc)
2607                 GOTO(out, rc);
2608
2609         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
2610         if (msfs == NULL) {
2611                 GOTO(out, rc = -EPROTO);
2612         }
2613
2614         *osfs = *msfs;
2615
2616         EXIT;
2617  out:
2618         ptlrpc_req_finished(req);
2619         return rc;
2620 }
2621
2622 static int osc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
2623                          void *karg, void *uarg)
2624 {
2625         struct obd_device *obd = exp->exp_obd;
2626         struct obd_ioctl_data *data = karg;
2627         int err = 0;
2628         ENTRY;
2629
2630         if (!try_module_get(THIS_MODULE)) {
2631                 CERROR("Can't get module. Is it alive?");
2632                 return -EINVAL;
2633         }
2634         switch (cmd) {
2635         case OBD_IOC_CLIENT_RECOVER:
2636                 err = ptlrpc_recover_import(obd->u.cli.cl_import,
2637                                             data->ioc_inlbuf1, 0);
2638                 if (err > 0)
2639                         err = 0;
2640                 GOTO(out, err);
2641         case IOC_OSC_SET_ACTIVE:
2642                 err = ptlrpc_set_import_active(obd->u.cli.cl_import,
2643                                                data->ioc_offset);
2644                 GOTO(out, err);
2645         case OBD_IOC_POLL_QUOTACHECK:
2646                 err = osc_quota_poll_check(exp, (struct if_quotacheck *)karg);
2647                 GOTO(out, err);
2648         case OBD_IOC_PING_TARGET:
2649                 err = ptlrpc_obd_ping(obd);
2650                 GOTO(out, err);
2651         default:
2652                 CDEBUG(D_INODE, "unrecognised ioctl %#x by %s\n",
2653                        cmd, current_comm());
2654                 GOTO(out, err = -ENOTTY);
2655         }
2656 out:
2657         module_put(THIS_MODULE);
2658         return err;
2659 }
2660
2661 static int osc_get_info(const struct lu_env *env, struct obd_export *exp,
2662                         obd_count keylen, void *key, __u32 *vallen, void *val,
2663                         struct lov_stripe_md *lsm)
2664 {
2665         ENTRY;
2666         if (!vallen || !val)
2667                 RETURN(-EFAULT);
2668
2669         if (KEY_IS(KEY_LOCK_TO_STRIPE)) {
2670                 __u32 *stripe = val;
2671                 *vallen = sizeof(*stripe);
2672                 *stripe = 0;
2673                 RETURN(0);
2674         } else if (KEY_IS(KEY_LAST_ID)) {
2675                 struct ptlrpc_request *req;
2676                 obd_id                *reply;
2677                 char                  *tmp;
2678                 int                    rc;
2679
2680                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
2681                                            &RQF_OST_GET_INFO_LAST_ID);
2682                 if (req == NULL)
2683                         RETURN(-ENOMEM);
2684
2685                 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
2686                                      RCL_CLIENT, keylen);
2687                 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GET_INFO);
2688                 if (rc) {
2689                         ptlrpc_request_free(req);
2690                         RETURN(rc);
2691                 }
2692
2693                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
2694                 memcpy(tmp, key, keylen);
2695
2696                 req->rq_no_delay = req->rq_no_resend = 1;
2697                 ptlrpc_request_set_replen(req);
2698                 rc = ptlrpc_queue_wait(req);
2699                 if (rc)
2700                         GOTO(out, rc);
2701
2702                 reply = req_capsule_server_get(&req->rq_pill, &RMF_OBD_ID);
2703                 if (reply == NULL)
2704                         GOTO(out, rc = -EPROTO);
2705
2706                 *((obd_id *)val) = *reply;
2707         out:
2708                 ptlrpc_req_finished(req);
2709                 RETURN(rc);
2710         } else if (KEY_IS(KEY_FIEMAP)) {
2711                 struct ll_fiemap_info_key *fm_key =
2712                                 (struct ll_fiemap_info_key *)key;
2713                 struct ldlm_res_id       res_id;
2714                 ldlm_policy_data_t       policy;
2715                 struct lustre_handle     lockh;
2716                 ldlm_mode_t              mode = 0;
2717                 struct ptlrpc_request   *req;
2718                 struct ll_user_fiemap   *reply;
2719                 char                    *tmp;
2720                 int                      rc;
2721
2722                 if (!(fm_key->fiemap.fm_flags & FIEMAP_FLAG_SYNC))
2723                         goto skip_locking;
2724
2725                 policy.l_extent.start = fm_key->fiemap.fm_start &
2726                                                 CFS_PAGE_MASK;
2727
2728                 if (OBD_OBJECT_EOF - fm_key->fiemap.fm_length <=
2729                     fm_key->fiemap.fm_start + PAGE_CACHE_SIZE - 1)
2730                         policy.l_extent.end = OBD_OBJECT_EOF;
2731                 else
2732                         policy.l_extent.end = (fm_key->fiemap.fm_start +
2733                                 fm_key->fiemap.fm_length +
2734                                 PAGE_CACHE_SIZE - 1) & CFS_PAGE_MASK;
2735
2736                 ostid_build_res_name(&fm_key->oa.o_oi, &res_id);
2737                 mode = ldlm_lock_match(exp->exp_obd->obd_namespace,
2738                                        LDLM_FL_BLOCK_GRANTED |
2739                                        LDLM_FL_LVB_READY,
2740                                        &res_id, LDLM_EXTENT, &policy,
2741                                        LCK_PR | LCK_PW, &lockh, 0);
2742                 if (mode) { /* lock is cached on client */
2743                         if (mode != LCK_PR) {
2744                                 ldlm_lock_addref(&lockh, LCK_PR);
2745                                 ldlm_lock_decref(&lockh, LCK_PW);
2746                         }
2747                 } else { /* no cached lock, needs acquire lock on server side */
2748                         fm_key->oa.o_valid |= OBD_MD_FLFLAGS;
2749                         fm_key->oa.o_flags |= OBD_FL_SRVLOCK;
2750                 }
2751
2752 skip_locking:
2753                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
2754                                            &RQF_OST_GET_INFO_FIEMAP);
2755                 if (req == NULL)
2756                         GOTO(drop_lock, rc = -ENOMEM);
2757
2758                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_KEY,
2759                                      RCL_CLIENT, keylen);
2760                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_VAL,
2761                                      RCL_CLIENT, *vallen);
2762                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_VAL,
2763                                      RCL_SERVER, *vallen);
2764
2765                 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GET_INFO);
2766                 if (rc) {
2767                         ptlrpc_request_free(req);
2768                         GOTO(drop_lock, rc);
2769                 }
2770
2771                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_FIEMAP_KEY);
2772                 memcpy(tmp, key, keylen);
2773                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_FIEMAP_VAL);
2774                 memcpy(tmp, val, *vallen);
2775
2776                 ptlrpc_request_set_replen(req);
2777                 rc = ptlrpc_queue_wait(req);
2778                 if (rc)
2779                         GOTO(fini_req, rc);
2780
2781                 reply = req_capsule_server_get(&req->rq_pill, &RMF_FIEMAP_VAL);
2782                 if (reply == NULL)
2783                         GOTO(fini_req, rc = -EPROTO);
2784
2785                 memcpy(val, reply, *vallen);
2786 fini_req:
2787                 ptlrpc_req_finished(req);
2788 drop_lock:
2789                 if (mode)
2790                         ldlm_lock_decref(&lockh, LCK_PR);
2791                 RETURN(rc);
2792         }
2793
2794         RETURN(-EINVAL);
2795 }
2796
2797 static int osc_set_info_async(const struct lu_env *env, struct obd_export *exp,
2798                               obd_count keylen, void *key, obd_count vallen,
2799                               void *val, struct ptlrpc_request_set *set)
2800 {
2801         struct ptlrpc_request *req;
2802         struct obd_device     *obd = exp->exp_obd;
2803         struct obd_import     *imp = class_exp2cliimp(exp);
2804         char                  *tmp;
2805         int                    rc;
2806         ENTRY;
2807
2808         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_SHUTDOWN, 10);
2809
2810         if (KEY_IS(KEY_CHECKSUM)) {
2811                 if (vallen != sizeof(int))
2812                         RETURN(-EINVAL);
2813                 exp->exp_obd->u.cli.cl_checksum = (*(int *)val) ? 1 : 0;
2814                 RETURN(0);
2815         }
2816
2817         if (KEY_IS(KEY_SPTLRPC_CONF)) {
2818                 sptlrpc_conf_client_adapt(obd);
2819                 RETURN(0);
2820         }
2821
2822         if (KEY_IS(KEY_FLUSH_CTX)) {
2823                 sptlrpc_import_flush_my_ctx(imp);
2824                 RETURN(0);
2825         }
2826
2827         if (KEY_IS(KEY_CACHE_SET)) {
2828                 struct client_obd *cli = &obd->u.cli;
2829
2830                 LASSERT(cli->cl_cache == NULL); /* only once */
2831                 cli->cl_cache = (struct cl_client_cache *)val;
2832                 atomic_inc(&cli->cl_cache->ccc_users);
2833                 cli->cl_lru_left = &cli->cl_cache->ccc_lru_left;
2834
2835                 /* add this osc into entity list */
2836                 LASSERT(list_empty(&cli->cl_lru_osc));
2837                 spin_lock(&cli->cl_cache->ccc_lru_lock);
2838                 list_add(&cli->cl_lru_osc, &cli->cl_cache->ccc_lru);
2839                 spin_unlock(&cli->cl_cache->ccc_lru_lock);
2840
2841                 RETURN(0);
2842         }
2843
2844         if (KEY_IS(KEY_CACHE_LRU_SHRINK)) {
2845                 struct client_obd *cli = &obd->u.cli;
2846                 int nr = atomic_read(&cli->cl_lru_in_list) >> 1;
2847                 int target = *(int *)val;
2848
2849                 nr = osc_lru_shrink(env, cli, min(nr, target), true);
2850                 *(int *)val -= nr;
2851                 RETURN(0);
2852         }
2853
2854         if (!set && !KEY_IS(KEY_GRANT_SHRINK))
2855                 RETURN(-EINVAL);
2856
2857         /* We pass all other commands directly to OST. Since nobody calls osc
2858            methods directly and everybody is supposed to go through LOV, we
2859            assume lov checked invalid values for us.
2860            The only recognised values so far are evict_by_nid and mds_conn.
2861            Even if something bad goes through, we'd get a -EINVAL from OST
2862            anyway. */
2863
2864         req = ptlrpc_request_alloc(imp, KEY_IS(KEY_GRANT_SHRINK) ?
2865                                                 &RQF_OST_SET_GRANT_INFO :
2866                                                 &RQF_OBD_SET_INFO);
2867         if (req == NULL)
2868                 RETURN(-ENOMEM);
2869
2870         req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
2871                              RCL_CLIENT, keylen);
2872         if (!KEY_IS(KEY_GRANT_SHRINK))
2873                 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_VAL,
2874                                      RCL_CLIENT, vallen);
2875         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SET_INFO);
2876         if (rc) {
2877                 ptlrpc_request_free(req);
2878                 RETURN(rc);
2879         }
2880
2881         tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
2882         memcpy(tmp, key, keylen);
2883         tmp = req_capsule_client_get(&req->rq_pill, KEY_IS(KEY_GRANT_SHRINK) ?
2884                                                         &RMF_OST_BODY :
2885                                                         &RMF_SETINFO_VAL);
2886         memcpy(tmp, val, vallen);
2887
2888         if (KEY_IS(KEY_GRANT_SHRINK)) {
2889                 struct osc_grant_args *aa;
2890                 struct obdo *oa;
2891
2892                 CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
2893                 aa = ptlrpc_req_async_args(req);
2894                 OBDO_ALLOC(oa);
2895                 if (!oa) {
2896                         ptlrpc_req_finished(req);
2897                         RETURN(-ENOMEM);
2898                 }
2899                 *oa = ((struct ost_body *)val)->oa;
2900                 aa->aa_oa = oa;
2901                 req->rq_interpret_reply = osc_shrink_grant_interpret;
2902         }
2903
2904         ptlrpc_request_set_replen(req);
2905         if (!KEY_IS(KEY_GRANT_SHRINK)) {
2906                 LASSERT(set != NULL);
2907                 ptlrpc_set_add_req(set, req);
2908                 ptlrpc_check_set(NULL, set);
2909         } else
2910                 ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
2911
2912         RETURN(0);
2913 }
2914
2915 static int osc_reconnect(const struct lu_env *env,
2916                          struct obd_export *exp, struct obd_device *obd,
2917                          struct obd_uuid *cluuid,
2918                          struct obd_connect_data *data,
2919                          void *localdata)
2920 {
2921         struct client_obd *cli = &obd->u.cli;
2922
2923         if (data != NULL && (data->ocd_connect_flags & OBD_CONNECT_GRANT)) {
2924                 long lost_grant;
2925
2926                 client_obd_list_lock(&cli->cl_loi_list_lock);
2927                 data->ocd_grant = (cli->cl_avail_grant + cli->cl_dirty) ?:
2928                                 2 * cli_brw_size(obd);
2929                 lost_grant = cli->cl_lost_grant;
2930                 cli->cl_lost_grant = 0;
2931                 client_obd_list_unlock(&cli->cl_loi_list_lock);
2932
2933                 CDEBUG(D_RPCTRACE, "ocd_connect_flags: "LPX64" ocd_version: %d"
2934                        " ocd_grant: %d, lost: %ld.\n", data->ocd_connect_flags,
2935                        data->ocd_version, data->ocd_grant, lost_grant);
2936         }
2937
2938         RETURN(0);
2939 }
2940
2941 static int osc_disconnect(struct obd_export *exp)
2942 {
2943         struct obd_device *obd = class_exp2obd(exp);
2944         struct llog_ctxt  *ctxt;
2945         int rc;
2946
2947         ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
2948         if (ctxt) {
2949                 if (obd->u.cli.cl_conn_count == 1) {
2950                         /* Flush any remaining cancel messages out to the
2951                          * target */
2952                         llog_sync(ctxt, exp, 0);
2953                 }
2954                 llog_ctxt_put(ctxt);
2955         } else {
2956                 CDEBUG(D_HA, "No LLOG_SIZE_REPL_CTXT found in obd %p\n",
2957                        obd);
2958         }
2959
2960         rc = client_disconnect_export(exp);
2961         /**
2962          * Initially we put del_shrink_grant before disconnect_export, but it
2963          * causes the following problem if setup (connect) and cleanup
2964          * (disconnect) are tangled together.
2965          *      connect p1                     disconnect p2
2966          *   ptlrpc_connect_import
2967          *     ...............               class_manual_cleanup
2968          *                                     osc_disconnect
2969          *                                     del_shrink_grant
2970          *   ptlrpc_connect_interrupt
2971          *     init_grant_shrink
2972          *   add this client to shrink list
2973          *                                      cleanup_osc
2974          * Bang! pinger trigger the shrink.
2975          * So the osc should be disconnected from the shrink list, after we
2976          * are sure the import has been destroyed. BUG18662
2977          */
2978         if (obd->u.cli.cl_import == NULL)
2979                 osc_del_shrink_grant(&obd->u.cli);
2980         return rc;
2981 }
2982
2983 static int osc_import_event(struct obd_device *obd,
2984                             struct obd_import *imp,
2985                             enum obd_import_event event)
2986 {
2987         struct client_obd *cli;
2988         int rc = 0;
2989
2990         ENTRY;
2991         LASSERT(imp->imp_obd == obd);
2992
2993         switch (event) {
2994         case IMP_EVENT_DISCON: {
2995                 cli = &obd->u.cli;
2996                 client_obd_list_lock(&cli->cl_loi_list_lock);
2997                 cli->cl_avail_grant = 0;
2998                 cli->cl_lost_grant = 0;
2999                 client_obd_list_unlock(&cli->cl_loi_list_lock);
3000                 break;
3001         }
3002         case IMP_EVENT_INACTIVE: {
3003                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_INACTIVE, NULL);
3004                 break;
3005         }
3006         case IMP_EVENT_INVALIDATE: {
3007                 struct ldlm_namespace *ns = obd->obd_namespace;
3008                 struct lu_env         *env;
3009                 int                    refcheck;
3010
3011                 env = cl_env_get(&refcheck);
3012                 if (!IS_ERR(env)) {
3013                         /* Reset grants */
3014                         cli = &obd->u.cli;
3015                         /* all pages go to failing rpcs due to the invalid
3016                          * import */
3017                         osc_io_unplug(env, cli, NULL, PDL_POLICY_ROUND);
3018
3019                         ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
3020                         cl_env_put(env, &refcheck);
3021                 } else
3022                         rc = PTR_ERR(env);
3023                 break;
3024         }
3025         case IMP_EVENT_ACTIVE: {
3026                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE, NULL);
3027                 break;
3028         }
3029         case IMP_EVENT_OCD: {
3030                 struct obd_connect_data *ocd = &imp->imp_connect_data;
3031
3032                 if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT)
3033                         osc_init_grant(&obd->u.cli, ocd);
3034
3035                 /* See bug 7198 */
3036                 if (ocd->ocd_connect_flags & OBD_CONNECT_REQPORTAL)
3037                         imp->imp_client->cli_request_portal =OST_REQUEST_PORTAL;
3038
3039                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_OCD, NULL);
3040                 break;
3041         }
3042         case IMP_EVENT_DEACTIVATE: {
3043                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_DEACTIVATE, NULL);
3044                 break;
3045         }
3046         case IMP_EVENT_ACTIVATE: {
3047                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVATE, NULL);
3048                 break;
3049         }
3050         default:
3051                 CERROR("Unknown import event %d\n", event);
3052                 LBUG();
3053         }
3054         RETURN(rc);
3055 }
3056
3057 /**
3058  * Determine whether the lock can be canceled before replaying the lock
3059  * during recovery, see bug16774 for detailed information.
3060  *
3061  * \retval zero the lock can't be canceled
3062  * \retval other ok to cancel
3063  */
3064 static int osc_cancel_weight(struct ldlm_lock *lock)
3065 {
3066         /*
3067          * Cancel all unused and granted extent lock.
3068          */
3069         if (lock->l_resource->lr_type == LDLM_EXTENT &&
3070             lock->l_granted_mode == lock->l_req_mode &&
3071             osc_ldlm_weigh_ast(lock) == 0)
3072                 RETURN(1);
3073
3074         RETURN(0);
3075 }
3076
3077 static int brw_queue_work(const struct lu_env *env, void *data)
3078 {
3079         struct client_obd *cli = data;
3080
3081         CDEBUG(D_CACHE, "Run writeback work for client obd %p.\n", cli);
3082
3083         osc_io_unplug(env, cli, NULL, PDL_POLICY_SAME);
3084         RETURN(0);
3085 }
3086
3087 int osc_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
3088 {
3089         struct client_obd *cli = &obd->u.cli;
3090         struct obd_type   *type;
3091         void              *handler;
3092         int                rc;
3093         ENTRY;
3094
3095         rc = ptlrpcd_addref();
3096         if (rc)
3097                 RETURN(rc);
3098
3099         rc = client_obd_setup(obd, lcfg);
3100         if (rc)
3101                 GOTO(out_ptlrpcd, rc);
3102
3103         handler = ptlrpcd_alloc_work(cli->cl_import, brw_queue_work, cli);
3104         if (IS_ERR(handler))
3105                 GOTO(out_client_setup, rc = PTR_ERR(handler));
3106         cli->cl_writeback_work = handler;
3107
3108         handler = ptlrpcd_alloc_work(cli->cl_import, lru_queue_work, cli);
3109         if (IS_ERR(handler))
3110                 GOTO(out_ptlrpcd_work, rc = PTR_ERR(handler));
3111         cli->cl_lru_work = handler;
3112
3113         rc = osc_quota_setup(obd);
3114         if (rc)
3115                 GOTO(out_ptlrpcd_work, rc);
3116
3117         cli->cl_grant_shrink_interval = GRANT_SHRINK_INTERVAL;
3118
3119 #ifdef LPROCFS
3120         obd->obd_vars = lprocfs_osc_obd_vars;
3121 #endif
3122         /* If this is true then both client (osc) and server (osp) are on the
3123          * same node. The osp layer if loaded first will register the osc proc
3124          * directory. In that case this obd_device will be attached its proc
3125          * tree to type->typ_procsym instead of obd->obd_type->typ_procroot. */
3126         type = class_search_type(LUSTRE_OSP_NAME);
3127         if (type && type->typ_procsym) {
3128                 obd->obd_proc_entry = lprocfs_seq_register(obd->obd_name,
3129                                                            type->typ_procsym,
3130                                                            obd->obd_vars, obd);
3131                 if (IS_ERR(obd->obd_proc_entry)) {
3132                         rc = PTR_ERR(obd->obd_proc_entry);
3133                         CERROR("error %d setting up lprocfs for %s\n", rc,
3134                                obd->obd_name);
3135                         obd->obd_proc_entry = NULL;
3136                 }
3137         } else {
3138                 rc = lprocfs_seq_obd_setup(obd);
3139         }
3140
3141         /* If the basic OSC proc tree construction succeeded then
3142          * lets do the rest. */
3143         if (rc == 0) {
3144                 lproc_osc_attach_seqstat(obd);
3145                 sptlrpc_lprocfs_cliobd_attach(obd);
3146                 ptlrpc_lprocfs_register_obd(obd);
3147         }
3148
3149         /* We need to allocate a few requests more, because
3150          * brw_interpret tries to create new requests before freeing
3151          * previous ones, Ideally we want to have 2x max_rpcs_in_flight
3152          * reserved, but I'm afraid that might be too much wasted RAM
3153          * in fact, so 2 is just my guess and still should work. */
3154         cli->cl_import->imp_rq_pool =
3155                 ptlrpc_init_rq_pool(cli->cl_max_rpcs_in_flight + 2,
3156                                     OST_MAXREQSIZE,
3157                                     ptlrpc_add_rqs_to_pool);
3158
3159         INIT_LIST_HEAD(&cli->cl_grant_shrink_list);
3160         ns_register_cancel(obd->obd_namespace, osc_cancel_weight);
3161         RETURN(0);
3162
3163 out_ptlrpcd_work:
3164         if (cli->cl_writeback_work != NULL) {
3165                 ptlrpcd_destroy_work(cli->cl_writeback_work);
3166                 cli->cl_writeback_work = NULL;
3167         }
3168         if (cli->cl_lru_work != NULL) {
3169                 ptlrpcd_destroy_work(cli->cl_lru_work);
3170                 cli->cl_lru_work = NULL;
3171         }
3172 out_client_setup:
3173         client_obd_cleanup(obd);
3174 out_ptlrpcd:
3175         ptlrpcd_decref();
3176         RETURN(rc);
3177 }
3178
3179 static int osc_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
3180 {
3181         int rc = 0;
3182         ENTRY;
3183
3184         switch (stage) {
3185         case OBD_CLEANUP_EARLY: {
3186                 struct obd_import *imp;
3187                 imp = obd->u.cli.cl_import;
3188                 CDEBUG(D_HA, "Deactivating import %s\n", obd->obd_name);
3189                 /* ptlrpc_abort_inflight to stop an mds_lov_synchronize */
3190                 ptlrpc_deactivate_import(imp);
3191                 spin_lock(&imp->imp_lock);
3192                 imp->imp_pingable = 0;
3193                 spin_unlock(&imp->imp_lock);
3194                 break;
3195         }
3196         case OBD_CLEANUP_EXPORTS: {
3197                 struct client_obd *cli = &obd->u.cli;
3198                 /* LU-464
3199                  * for echo client, export may be on zombie list, wait for
3200                  * zombie thread to cull it, because cli.cl_import will be
3201                  * cleared in client_disconnect_export():
3202                  *   class_export_destroy() -> obd_cleanup() ->
3203                  *   echo_device_free() -> echo_client_cleanup() ->
3204                  *   obd_disconnect() -> osc_disconnect() ->
3205                  *   client_disconnect_export()
3206                  */
3207                 obd_zombie_barrier();
3208                 if (cli->cl_writeback_work) {
3209                         ptlrpcd_destroy_work(cli->cl_writeback_work);
3210                         cli->cl_writeback_work = NULL;
3211                 }
3212                 if (cli->cl_lru_work) {
3213                         ptlrpcd_destroy_work(cli->cl_lru_work);
3214                         cli->cl_lru_work = NULL;
3215                 }
3216                 obd_cleanup_client_import(obd);
3217                 ptlrpc_lprocfs_unregister_obd(obd);
3218                 lprocfs_obd_cleanup(obd);
3219                 rc = obd_llog_finish(obd, 0);
3220                 if (rc != 0)
3221                         CERROR("failed to cleanup llogging subsystems\n");
3222                 break;
3223                 }
3224         }
3225         RETURN(rc);
3226 }
3227
3228 int osc_cleanup(struct obd_device *obd)
3229 {
3230         struct client_obd *cli = &obd->u.cli;
3231         int rc;
3232
3233         ENTRY;
3234
3235         /* lru cleanup */
3236         if (cli->cl_cache != NULL) {
3237                 LASSERT(atomic_read(&cli->cl_cache->ccc_users) > 0);
3238                 spin_lock(&cli->cl_cache->ccc_lru_lock);
3239                 list_del_init(&cli->cl_lru_osc);
3240                 spin_unlock(&cli->cl_cache->ccc_lru_lock);
3241                 cli->cl_lru_left = NULL;
3242                 atomic_dec(&cli->cl_cache->ccc_users);
3243                 cli->cl_cache = NULL;
3244         }
3245
3246         /* free memory of osc quota cache */
3247         osc_quota_cleanup(obd);
3248
3249         rc = client_obd_cleanup(obd);
3250
3251         ptlrpcd_decref();
3252         RETURN(rc);
3253 }
3254
3255 int osc_process_config_base(struct obd_device *obd, struct lustre_cfg *lcfg)
3256 {
3257         int rc = class_process_proc_seq_param(PARAM_OSC, obd->obd_vars,
3258                                               lcfg, obd);
3259         return rc > 0 ? 0: rc;
3260 }
3261
3262 static int osc_process_config(struct obd_device *obd, obd_count len, void *buf)
3263 {
3264         return osc_process_config_base(obd, buf);
3265 }
3266
3267 struct obd_ops osc_obd_ops = {
3268         .o_owner                = THIS_MODULE,
3269         .o_setup                = osc_setup,
3270         .o_precleanup           = osc_precleanup,
3271         .o_cleanup              = osc_cleanup,
3272         .o_add_conn             = client_import_add_conn,
3273         .o_del_conn             = client_import_del_conn,
3274         .o_connect              = client_connect_import,
3275         .o_reconnect            = osc_reconnect,
3276         .o_disconnect           = osc_disconnect,
3277         .o_statfs               = osc_statfs,
3278         .o_statfs_async         = osc_statfs_async,
3279         .o_unpackmd             = osc_unpackmd,
3280         .o_create               = osc_create,
3281         .o_destroy              = osc_destroy,
3282         .o_getattr              = osc_getattr,
3283         .o_getattr_async        = osc_getattr_async,
3284         .o_setattr              = osc_setattr,
3285         .o_setattr_async        = osc_setattr_async,
3286         .o_change_cbdata        = osc_change_cbdata,
3287         .o_find_cbdata          = osc_find_cbdata,
3288         .o_iocontrol            = osc_iocontrol,
3289         .o_get_info             = osc_get_info,
3290         .o_set_info_async       = osc_set_info_async,
3291         .o_import_event         = osc_import_event,
3292         .o_process_config       = osc_process_config,
3293         .o_quotactl             = osc_quotactl,
3294         .o_quotacheck           = osc_quotacheck,
3295 };
3296
3297 extern struct lu_kmem_descr osc_caches[];
3298 extern spinlock_t osc_ast_guard;
3299 extern struct lock_class_key osc_ast_guard_class;
3300
3301 int __init osc_init(void)
3302 {
3303         bool enable_proc = true;
3304         struct obd_type *type;
3305         int rc;
3306         ENTRY;
3307
3308         /* print an address of _any_ initialized kernel symbol from this
3309          * module, to allow debugging with gdb that doesn't support data
3310          * symbols from modules.*/
3311         CDEBUG(D_INFO, "Lustre OSC module (%p).\n", &osc_caches);
3312
3313         rc = lu_kmem_init(osc_caches);
3314         if (rc)
3315                 RETURN(rc);
3316
3317         type = class_search_type(LUSTRE_OSP_NAME);
3318         if (type != NULL && type->typ_procsym != NULL)
3319                 enable_proc = false;
3320
3321         rc = class_register_type(&osc_obd_ops, NULL, enable_proc, NULL,
3322 #ifndef HAVE_ONLY_PROCFS_SEQ
3323                                  NULL,
3324 #endif
3325                                  LUSTRE_OSC_NAME, &osc_device_type);
3326         if (rc) {
3327                 lu_kmem_fini(osc_caches);
3328                 RETURN(rc);
3329         }
3330
3331         spin_lock_init(&osc_ast_guard);
3332         lockdep_set_class(&osc_ast_guard, &osc_ast_guard_class);
3333
3334         RETURN(rc);
3335 }
3336
3337 #ifdef __KERNEL__
3338 static void /*__exit*/ osc_exit(void)
3339 {
3340         class_unregister_type(LUSTRE_OSC_NAME);
3341         lu_kmem_fini(osc_caches);
3342 }
3343
3344 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
3345 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC)");
3346 MODULE_LICENSE("GPL");
3347
3348 cfs_module(osc, LUSTRE_VERSION_STRING, osc_init, osc_exit);
3349 #endif