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