lustre/osc/osc_request.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  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 #ifndef EXPORT_SYMTAB
  38 # define EXPORT_SYMTAB
  39 #endif
  40 #define DEBUG_SUBSYSTEM S_OSC
  41
  42 #include <libcfs/libcfs.h>
  43
  44 #ifndef __KERNEL__
  45 # include <liblustre.h>
  46 #endif
  47
  48 #include <lustre_dlm.h>
  49 #include <lustre_net.h>
  50 #include <lustre/lustre_user.h>
  51 #include <obd_cksum.h>
  52 #include <obd_ost.h>
  53 #include <obd_lov.h>
  54
  55 #ifdef  __CYGWIN__
  56 # include <ctype.h>
  57 #endif
  58
  59 #include <lustre_ha.h>
  60 #include <lprocfs_status.h>
  61 #include <lustre_log.h>
  62 #include <lustre_debug.h>
  63 #include <lustre_param.h>
  64 #include "osc_internal.h"
  65
  66 static quota_interface_t *quota_interface = NULL;
  67 extern quota_interface_t osc_quota_interface;
  68
  69 static void osc_release_ppga(struct brw_page **ppga, obd_count count);
  70 static int brw_interpret(const struct lu_env *env,
  71                          struct ptlrpc_request *req, void *data, int rc);
  72 int osc_cleanup(struct obd_device *obd);
  73
  74 /* Pack OSC object metadata for disk storage (LE byte order). */
  75 static int osc_packmd(struct obd_export *exp, struct lov_mds_md **lmmp,
  76                       struct lov_stripe_md *lsm)
  77 {
  78         int lmm_size;
  79         ENTRY;
  80
  81         lmm_size = sizeof(**lmmp);
  82         if (!lmmp)
  83                 RETURN(lmm_size);
  84
  85         if (*lmmp && !lsm) {
  86                 OBD_FREE(*lmmp, lmm_size);
  87                 *lmmp = NULL;
  88                 RETURN(0);
  89         }
  90
  91         if (!*lmmp) {
  92                 OBD_ALLOC(*lmmp, lmm_size);
  93                 if (!*lmmp)
  94                         RETURN(-ENOMEM);
  95         }
  96
  97         if (lsm) {
  98                 LASSERT(lsm->lsm_object_id);
  99                 LASSERT_SEQ_IS_MDT(lsm->lsm_object_seq);
 100                 (*lmmp)->lmm_object_id = cpu_to_le64(lsm->lsm_object_id);
 101                 (*lmmp)->lmm_object_seq = cpu_to_le64(lsm->lsm_object_seq);
 102         }
 103
 104         RETURN(lmm_size);
 105 }
 106
 107 /* Unpack OSC object metadata from disk storage (LE byte order). */
 108 static int osc_unpackmd(struct obd_export *exp, struct lov_stripe_md **lsmp,
 109                         struct lov_mds_md *lmm, int lmm_bytes)
 110 {
 111         int lsm_size;
 112         ENTRY;
 113
 114         if (lmm != NULL) {
 115                 if (lmm_bytes < sizeof (*lmm)) {
 116                         CERROR("lov_mds_md too small: %d, need %d\n",
 117                                lmm_bytes, (int)sizeof(*lmm));
 118                         RETURN(-EINVAL);
 119                 }
 120                 /* XXX LOV_MAGIC etc check? */
 121
 122                 if (lmm->lmm_object_id == 0) {
 123                         CERROR("lov_mds_md: zero lmm_object_id\n");
 124                         RETURN(-EINVAL);
 125                 }
 126         }
 127
 128         lsm_size = lov_stripe_md_size(1);
 129         if (lsmp == NULL)
 130                 RETURN(lsm_size);
 131
 132         if (*lsmp != NULL && lmm == NULL) {
 133                 OBD_FREE((*lsmp)->lsm_oinfo[0], sizeof(struct lov_oinfo));
 134                 OBD_FREE(*lsmp, lsm_size);
 135                 *lsmp = NULL;
 136                 RETURN(0);
 137         }
 138
 139         if (*lsmp == NULL) {
 140                 OBD_ALLOC(*lsmp, lsm_size);
 141                 if (*lsmp == NULL)
 142                         RETURN(-ENOMEM);
 143                 OBD_ALLOC((*lsmp)->lsm_oinfo[0], sizeof(struct lov_oinfo));
 144                 if ((*lsmp)->lsm_oinfo[0] == NULL) {
 145                         OBD_FREE(*lsmp, lsm_size);
 146                         RETURN(-ENOMEM);
 147                 }
 148                 loi_init((*lsmp)->lsm_oinfo[0]);
 149         }
 150
 151         if (lmm != NULL) {
 152                 /* XXX zero *lsmp? */
 153                 (*lsmp)->lsm_object_id = le64_to_cpu (lmm->lmm_object_id);
 154                 (*lsmp)->lsm_object_seq = le64_to_cpu (lmm->lmm_object_seq);
 155                 LASSERT((*lsmp)->lsm_object_id);
 156                 LASSERT_SEQ_IS_MDT((*lsmp)->lsm_object_seq);
 157         }
 158
 159         (*lsmp)->lsm_maxbytes = LUSTRE_STRIPE_MAXBYTES;
 160
 161         RETURN(lsm_size);
 162 }
 163
 164 static inline void osc_pack_capa(struct ptlrpc_request *req,
 165                                  struct ost_body *body, void *capa)
 166 {
 167         struct obd_capa *oc = (struct obd_capa *)capa;
 168         struct lustre_capa *c;
 169
 170         if (!capa)
 171                 return;
 172
 173         c = req_capsule_client_get(&req->rq_pill, &RMF_CAPA1);
 174         LASSERT(c);
 175         capa_cpy(c, oc);
 176         body->oa.o_valid |= OBD_MD_FLOSSCAPA;
 177         DEBUG_CAPA(D_SEC, c, "pack");
 178 }
 179
 180 static inline void osc_pack_req_body(struct ptlrpc_request *req,
 181                                      struct obd_info *oinfo)
 182 {
 183         struct ost_body *body;
 184
 185         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 186         LASSERT(body);
 187
 188         lustre_set_wire_obdo(&body->oa, oinfo->oi_oa);
 189         osc_pack_capa(req, body, oinfo->oi_capa);
 190 }
 191
 192 static inline void osc_set_capa_size(struct ptlrpc_request *req,
 193                                      const struct req_msg_field *field,
 194                                      struct obd_capa *oc)
 195 {
 196         if (oc == NULL)
 197                 req_capsule_set_size(&req->rq_pill, field, RCL_CLIENT, 0);
 198         else
 199                 /* it is already calculated as sizeof struct obd_capa */
 200                 ;
 201 }
 202
 203 static int osc_getattr_interpret(const struct lu_env *env,
 204                                  struct ptlrpc_request *req,
 205                                  struct osc_async_args *aa, int rc)
 206 {
 207         struct ost_body *body;
 208         ENTRY;
 209
 210         if (rc != 0)
 211                 GOTO(out, rc);
 212
 213         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 214         if (body) {
 215                 CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
 216                 lustre_get_wire_obdo(aa->aa_oi->oi_oa, &body->oa);
 217
 218                 /* This should really be sent by the OST */
 219                 aa->aa_oi->oi_oa->o_blksize = PTLRPC_MAX_BRW_SIZE;
 220                 aa->aa_oi->oi_oa->o_valid |= OBD_MD_FLBLKSZ;
 221         } else {
 222                 CDEBUG(D_INFO, "can't unpack ost_body\n");
 223                 rc = -EPROTO;
 224                 aa->aa_oi->oi_oa->o_valid = 0;
 225         }
 226 out:
 227         rc = aa->aa_oi->oi_cb_up(aa->aa_oi, rc);
 228         RETURN(rc);
 229 }
 230
 231 static int osc_getattr_async(struct obd_export *exp, struct obd_info *oinfo,
 232                              struct ptlrpc_request_set *set)
 233 {
 234         struct ptlrpc_request *req;
 235         struct osc_async_args *aa;
 236         int                    rc;
 237         ENTRY;
 238
 239         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_GETATTR);
 240         if (req == NULL)
 241                 RETURN(-ENOMEM);
 242
 243         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
 244         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GETATTR);
 245         if (rc) {
 246                 ptlrpc_request_free(req);
 247                 RETURN(rc);
 248         }
 249
 250         osc_pack_req_body(req, oinfo);
 251
 252         ptlrpc_request_set_replen(req);
 253         req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_getattr_interpret;
 254
 255         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
 256         aa = ptlrpc_req_async_args(req);
 257         aa->aa_oi = oinfo;
 258
 259         ptlrpc_set_add_req(set, req);
 260         RETURN(0);
 261 }
 262
 263 static int osc_getattr(struct obd_export *exp, struct obd_info *oinfo)
 264 {
 265         struct ptlrpc_request *req;
 266         struct ost_body       *body;
 267         int                    rc;
 268         ENTRY;
 269
 270         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_GETATTR);
 271         if (req == NULL)
 272                 RETURN(-ENOMEM);
 273
 274         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
 275         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GETATTR);
 276         if (rc) {
 277                 ptlrpc_request_free(req);
 278                 RETURN(rc);
 279         }
 280
 281         osc_pack_req_body(req, oinfo);
 282
 283         ptlrpc_request_set_replen(req);
 284
 285         rc = ptlrpc_queue_wait(req);
 286         if (rc)
 287                 GOTO(out, rc);
 288
 289         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 290         if (body == NULL)
 291                 GOTO(out, rc = -EPROTO);
 292
 293         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
 294         lustre_get_wire_obdo(oinfo->oi_oa, &body->oa);
 295
 296         /* This should really be sent by the OST */
 297         oinfo->oi_oa->o_blksize = PTLRPC_MAX_BRW_SIZE;
 298         oinfo->oi_oa->o_valid |= OBD_MD_FLBLKSZ;
 299
 300         EXIT;
 301  out:
 302         ptlrpc_req_finished(req);
 303         return rc;
 304 }
 305
 306 static int osc_setattr(struct obd_export *exp, struct obd_info *oinfo,
 307                        struct obd_trans_info *oti)
 308 {
 309         struct ptlrpc_request *req;
 310         struct ost_body       *body;
 311         int                    rc;
 312         ENTRY;
 313
 314         LASSERT(oinfo->oi_oa->o_valid & OBD_MD_FLGROUP);
 315
 316         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SETATTR);
 317         if (req == NULL)
 318                 RETURN(-ENOMEM);
 319
 320         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
 321         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SETATTR);
 322         if (rc) {
 323                 ptlrpc_request_free(req);
 324                 RETURN(rc);
 325         }
 326
 327         osc_pack_req_body(req, oinfo);
 328
 329         ptlrpc_request_set_replen(req);
 330
 331         rc = ptlrpc_queue_wait(req);
 332         if (rc)
 333                 GOTO(out, rc);
 334
 335         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 336         if (body == NULL)
 337                 GOTO(out, rc = -EPROTO);
 338
 339         lustre_get_wire_obdo(oinfo->oi_oa, &body->oa);
 340
 341         EXIT;
 342 out:
 343         ptlrpc_req_finished(req);
 344         RETURN(rc);
 345 }
 346
 347 static int osc_setattr_interpret(const struct lu_env *env,
 348                                  struct ptlrpc_request *req,
 349                                  struct osc_setattr_args *sa, int rc)
 350 {
 351         struct ost_body *body;
 352         ENTRY;
 353
 354         if (rc != 0)
 355                 GOTO(out, rc);
 356
 357         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 358         if (body == NULL)
 359                 GOTO(out, rc = -EPROTO);
 360
 361         lustre_get_wire_obdo(sa->sa_oa, &body->oa);
 362 out:
 363         rc = sa->sa_upcall(sa->sa_cookie, rc);
 364         RETURN(rc);
 365 }
 366
 367 int osc_setattr_async_base(struct obd_export *exp, struct obd_info *oinfo,
 368                            struct obd_trans_info *oti,
 369                            obd_enqueue_update_f upcall, void *cookie,
 370                            struct ptlrpc_request_set *rqset)
 371 {
 372         struct ptlrpc_request   *req;
 373         struct osc_setattr_args *sa;
 374         int                      rc;
 375         ENTRY;
 376
 377         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SETATTR);
 378         if (req == NULL)
 379                 RETURN(-ENOMEM);
 380
 381         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
 382         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SETATTR);
 383         if (rc) {
 384                 ptlrpc_request_free(req);
 385                 RETURN(rc);
 386         }
 387
 388         if (oti && oinfo->oi_oa->o_valid & OBD_MD_FLCOOKIE)
 389                 oinfo->oi_oa->o_lcookie = *oti->oti_logcookies;
 390
 391         osc_pack_req_body(req, oinfo);
 392
 393         ptlrpc_request_set_replen(req);
 394
 395         /* do mds to ost setattr asynchronously */
 396         if (!rqset) {
 397                 /* Do not wait for response. */
 398                 ptlrpcd_add_req(req, PSCOPE_OTHER);
 399         } else {
 400                 req->rq_interpret_reply =
 401                         (ptlrpc_interpterer_t)osc_setattr_interpret;
 402
 403                 CLASSERT (sizeof(*sa) <= sizeof(req->rq_async_args));
 404                 sa = ptlrpc_req_async_args(req);
 405                 sa->sa_oa = oinfo->oi_oa;
 406                 sa->sa_upcall = upcall;
 407                 sa->sa_cookie = cookie;
 408
 409                 if (rqset == PTLRPCD_SET)
 410                         ptlrpcd_add_req(req, PSCOPE_OTHER);
 411                 else
 412                         ptlrpc_set_add_req(rqset, req);
 413         }
 414
 415         RETURN(0);
 416 }
 417
 418 static int osc_setattr_async(struct obd_export *exp, struct obd_info *oinfo,
 419                              struct obd_trans_info *oti,
 420                              struct ptlrpc_request_set *rqset)
 421 {
 422         return osc_setattr_async_base(exp, oinfo, oti,
 423                                       oinfo->oi_cb_up, oinfo, rqset);
 424 }
 425
 426 int osc_real_create(struct obd_export *exp, struct obdo *oa,
 427                     struct lov_stripe_md **ea, struct obd_trans_info *oti)
 428 {
 429         struct ptlrpc_request *req;
 430         struct ost_body       *body;
 431         struct lov_stripe_md  *lsm;
 432         int                    rc;
 433         ENTRY;
 434
 435         LASSERT(oa);
 436         LASSERT(ea);
 437
 438         lsm = *ea;
 439         if (!lsm) {
 440                 rc = obd_alloc_memmd(exp, &lsm);
 441                 if (rc < 0)
 442                         RETURN(rc);
 443         }
 444
 445         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_CREATE);
 446         if (req == NULL)
 447                 GOTO(out, rc = -ENOMEM);
 448
 449         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_CREATE);
 450         if (rc) {
 451                 ptlrpc_request_free(req);
 452                 GOTO(out, rc);
 453         }
 454
 455         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 456         LASSERT(body);
 457         lustre_set_wire_obdo(&body->oa, oa);
 458
 459         ptlrpc_request_set_replen(req);
 460
 461         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
 462             oa->o_flags == OBD_FL_DELORPHAN) {
 463                 DEBUG_REQ(D_HA, req,
 464                           "delorphan from OST integration");
 465                 /* Don't resend the delorphan req */
 466                 req->rq_no_resend = req->rq_no_delay = 1;
 467         }
 468
 469         rc = ptlrpc_queue_wait(req);
 470         if (rc)
 471                 GOTO(out_req, rc);
 472
 473         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 474         if (body == NULL)
 475                 GOTO(out_req, rc = -EPROTO);
 476
 477         lustre_get_wire_obdo(oa, &body->oa);
 478
 479         /* This should really be sent by the OST */
 480         oa->o_blksize = PTLRPC_MAX_BRW_SIZE;
 481         oa->o_valid |= OBD_MD_FLBLKSZ;
 482
 483         /* XXX LOV STACKING: the lsm that is passed to us from LOV does not
 484          * have valid lsm_oinfo data structs, so don't go touching that.
 485          * This needs to be fixed in a big way.
 486          */
 487         lsm->lsm_object_id = oa->o_id;
 488         lsm->lsm_object_seq = oa->o_seq;
 489         *ea = lsm;
 490
 491         if (oti != NULL) {
 492                 oti->oti_transno = lustre_msg_get_transno(req->rq_repmsg);
 493
 494                 if (oa->o_valid & OBD_MD_FLCOOKIE) {
 495                         if (!oti->oti_logcookies)
 496                                 oti_alloc_cookies(oti, 1);
 497                         *oti->oti_logcookies = oa->o_lcookie;
 498                 }
 499         }
 500
 501         CDEBUG(D_HA, "transno: "LPD64"\n",
 502                lustre_msg_get_transno(req->rq_repmsg));
 503 out_req:
 504         ptlrpc_req_finished(req);
 505 out:
 506         if (rc && !*ea)
 507                 obd_free_memmd(exp, &lsm);
 508         RETURN(rc);
 509 }
 510
 511 int osc_punch_base(struct obd_export *exp, struct obd_info *oinfo,
 512                    obd_enqueue_update_f upcall, void *cookie,
 513                    struct ptlrpc_request_set *rqset)
 514 {
 515         struct ptlrpc_request   *req;
 516         struct osc_setattr_args *sa;
 517         struct ost_body         *body;
 518         int                      rc;
 519         ENTRY;
 520
 521         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_PUNCH);
 522         if (req == NULL)
 523                 RETURN(-ENOMEM);
 524
 525         osc_set_capa_size(req, &RMF_CAPA1, oinfo->oi_capa);
 526         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_PUNCH);
 527         if (rc) {
 528                 ptlrpc_request_free(req);
 529                 RETURN(rc);
 530         }
 531         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
 532         ptlrpc_at_set_req_timeout(req);
 533
 534         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 535         LASSERT(body);
 536         lustre_set_wire_obdo(&body->oa, oinfo->oi_oa);
 537         osc_pack_capa(req, body, oinfo->oi_capa);
 538
 539         ptlrpc_request_set_replen(req);
 540
 541
 542         req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_setattr_interpret;
 543         CLASSERT (sizeof(*sa) <= sizeof(req->rq_async_args));
 544         sa = ptlrpc_req_async_args(req);
 545         sa->sa_oa     = oinfo->oi_oa;
 546         sa->sa_upcall = upcall;
 547         sa->sa_cookie = cookie;
 548         if (rqset == PTLRPCD_SET)
 549                 ptlrpcd_add_req(req, PSCOPE_OTHER);
 550         else
 551                 ptlrpc_set_add_req(rqset, req);
 552
 553         RETURN(0);
 554 }
 555
 556 static int osc_punch(struct obd_export *exp, struct obd_info *oinfo,
 557                      struct obd_trans_info *oti,
 558                      struct ptlrpc_request_set *rqset)
 559 {
 560         oinfo->oi_oa->o_size   = oinfo->oi_policy.l_extent.start;
 561         oinfo->oi_oa->o_blocks = oinfo->oi_policy.l_extent.end;
 562         oinfo->oi_oa->o_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
 563         return osc_punch_base(exp, oinfo,
 564                               oinfo->oi_cb_up, oinfo, rqset);
 565 }
 566
 567 static int osc_sync(struct obd_export *exp, struct obdo *oa,
 568                     struct lov_stripe_md *md, obd_size start, obd_size end,
 569                     void *capa)
 570 {
 571         struct ptlrpc_request *req;
 572         struct ost_body       *body;
 573         int                    rc;
 574         ENTRY;
 575
 576         if (!oa) {
 577                 CDEBUG(D_INFO, "oa NULL\n");
 578                 RETURN(-EINVAL);
 579         }
 580
 581         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_SYNC);
 582         if (req == NULL)
 583                 RETURN(-ENOMEM);
 584
 585         osc_set_capa_size(req, &RMF_CAPA1, capa);
 586         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SYNC);
 587         if (rc) {
 588                 ptlrpc_request_free(req);
 589                 RETURN(rc);
 590         }
 591
 592         /* overload the size and blocks fields in the oa with start/end */
 593         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 594         LASSERT(body);
 595         lustre_set_wire_obdo(&body->oa, oa);
 596         body->oa.o_size = start;
 597         body->oa.o_blocks = end;
 598         body->oa.o_valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS);
 599         osc_pack_capa(req, body, capa);
 600
 601         ptlrpc_request_set_replen(req);
 602
 603         rc = ptlrpc_queue_wait(req);
 604         if (rc)
 605                 GOTO(out, rc);
 606
 607         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 608         if (body == NULL)
 609                 GOTO(out, rc = -EPROTO);
 610
 611         lustre_get_wire_obdo(oa, &body->oa);
 612
 613         EXIT;
 614  out:
 615         ptlrpc_req_finished(req);
 616         return rc;
 617 }
 618
 619 /* Find and cancel locally locks matched by @mode in the resource found by
 620  * @objid. Found locks are added into @cancel list. Returns the amount of
 621  * locks added to @cancels list. */
 622 static int osc_resource_get_unused(struct obd_export *exp, struct obdo *oa,
 623                                    cfs_list_t *cancels,
 624                                    ldlm_mode_t mode, int lock_flags)
 625 {
 626         struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
 627         struct ldlm_res_id res_id;
 628         struct ldlm_resource *res;
 629         int count;
 630         ENTRY;
 631
 632         osc_build_res_name(oa->o_id, oa->o_seq, &res_id);
 633         res = ldlm_resource_get(ns, NULL, &res_id, 0, 0);
 634         if (res == NULL)
 635                 RETURN(0);
 636
 637         LDLM_RESOURCE_ADDREF(res);
 638         count = ldlm_cancel_resource_local(res, cancels, NULL, mode,
 639                                            lock_flags, 0, NULL);
 640         LDLM_RESOURCE_DELREF(res);
 641         ldlm_resource_putref(res);
 642         RETURN(count);
 643 }
 644
 645 static int osc_destroy_interpret(const struct lu_env *env,
 646                                  struct ptlrpc_request *req, void *data,
 647                                  int rc)
 648 {
 649         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
 650
 651         cfs_atomic_dec(&cli->cl_destroy_in_flight);
 652         cfs_waitq_signal(&cli->cl_destroy_waitq);
 653         return 0;
 654 }
 655
 656 static int osc_can_send_destroy(struct client_obd *cli)
 657 {
 658         if (cfs_atomic_inc_return(&cli->cl_destroy_in_flight) <=
 659             cli->cl_max_rpcs_in_flight) {
 660                 /* The destroy request can be sent */
 661                 return 1;
 662         }
 663         if (cfs_atomic_dec_return(&cli->cl_destroy_in_flight) <
 664             cli->cl_max_rpcs_in_flight) {
 665                 /*
 666                  * The counter has been modified between the two atomic
 667                  * operations.
 668                  */
 669                 cfs_waitq_signal(&cli->cl_destroy_waitq);
 670         }
 671         return 0;
 672 }
 673
 674 /* Destroy requests can be async always on the client, and we don't even really
 675  * care about the return code since the client cannot do anything at all about
 676  * a destroy failure.
 677  * When the MDS is unlinking a filename, it saves the file objects into a
 678  * recovery llog, and these object records are cancelled when the OST reports
 679  * they were destroyed and sync'd to disk (i.e. transaction committed).
 680  * If the client dies, or the OST is down when the object should be destroyed,
 681  * the records are not cancelled, and when the OST reconnects to the MDS next,
 682  * it will retrieve the llog unlink logs and then sends the log cancellation
 683  * cookies to the MDS after committing destroy transactions. */
 684 static int osc_destroy(struct obd_export *exp, struct obdo *oa,
 685                        struct lov_stripe_md *ea, struct obd_trans_info *oti,
 686                        struct obd_export *md_export, void *capa)
 687 {
 688         struct client_obd     *cli = &exp->exp_obd->u.cli;
 689         struct ptlrpc_request *req;
 690         struct ost_body       *body;
 691         CFS_LIST_HEAD(cancels);
 692         int rc, count;
 693         ENTRY;
 694
 695         if (!oa) {
 696                 CDEBUG(D_INFO, "oa NULL\n");
 697                 RETURN(-EINVAL);
 698         }
 699
 700         count = osc_resource_get_unused(exp, oa, &cancels, LCK_PW,
 701                                         LDLM_FL_DISCARD_DATA);
 702
 703         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_DESTROY);
 704         if (req == NULL) {
 705                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
 706                 RETURN(-ENOMEM);
 707         }
 708
 709         osc_set_capa_size(req, &RMF_CAPA1, (struct obd_capa *)capa);
 710         rc = ldlm_prep_elc_req(exp, req, LUSTRE_OST_VERSION, OST_DESTROY,
 711                                0, &cancels, count);
 712         if (rc) {
 713                 ptlrpc_request_free(req);
 714                 RETURN(rc);
 715         }
 716
 717         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
 718         ptlrpc_at_set_req_timeout(req);
 719
 720         if (oti != NULL && oa->o_valid & OBD_MD_FLCOOKIE)
 721                 oa->o_lcookie = *oti->oti_logcookies;
 722         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 723         LASSERT(body);
 724         lustre_set_wire_obdo(&body->oa, oa);
 725
 726         osc_pack_capa(req, body, (struct obd_capa *)capa);
 727         ptlrpc_request_set_replen(req);
 728
 729         /* don't throttle destroy RPCs for the MDT */
 730         if (!(cli->cl_import->imp_connect_flags_orig & OBD_CONNECT_MDS)) {
 731                 req->rq_interpret_reply = osc_destroy_interpret;
 732                 if (!osc_can_send_destroy(cli)) {
 733                         struct l_wait_info lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP,
 734                                                           NULL);
 735
 736                         /*
 737                          * Wait until the number of on-going destroy RPCs drops
 738                          * under max_rpc_in_flight
 739                          */
 740                         l_wait_event_exclusive(cli->cl_destroy_waitq,
 741                                                osc_can_send_destroy(cli), &lwi);
 742                 }
 743         }
 744
 745         /* Do not wait for response */
 746         ptlrpcd_add_req(req, PSCOPE_OTHER);
 747         RETURN(0);
 748 }
 749
 750 static void osc_announce_cached(struct client_obd *cli, struct obdo *oa,
 751                                 long writing_bytes)
 752 {
 753         obd_flag bits = OBD_MD_FLBLOCKS|OBD_MD_FLGRANT;
 754
 755         LASSERT(!(oa->o_valid & bits));
 756
 757         oa->o_valid |= bits;
 758         client_obd_list_lock(&cli->cl_loi_list_lock);
 759         oa->o_dirty = cli->cl_dirty;
 760         if (cli->cl_dirty - cli->cl_dirty_transit > cli->cl_dirty_max) {
 761                 CERROR("dirty %lu - %lu > dirty_max %lu\n",
 762                        cli->cl_dirty, cli->cl_dirty_transit, cli->cl_dirty_max);
 763                 oa->o_undirty = 0;
 764         } else if (cfs_atomic_read(&obd_dirty_pages) -
 765                    cfs_atomic_read(&obd_dirty_transit_pages) >
 766                    obd_max_dirty_pages + 1){
 767                 /* The cfs_atomic_read() allowing the cfs_atomic_inc() are
 768                  * not covered by a lock thus they may safely race and trip
 769                  * this CERROR() unless we add in a small fudge factor (+1). */
 770                 CERROR("dirty %d - %d > system dirty_max %d\n",
 771                        cfs_atomic_read(&obd_dirty_pages),
 772                        cfs_atomic_read(&obd_dirty_transit_pages),
 773                        obd_max_dirty_pages);
 774                 oa->o_undirty = 0;
 775         } else if (cli->cl_dirty_max - cli->cl_dirty > 0x7fffffff) {
 776                 CERROR("dirty %lu - dirty_max %lu too big???\n",
 777                        cli->cl_dirty, cli->cl_dirty_max);
 778                 oa->o_undirty = 0;
 779         } else {
 780                 long max_in_flight = (cli->cl_max_pages_per_rpc << CFS_PAGE_SHIFT)*
 781                                 (cli->cl_max_rpcs_in_flight + 1);
 782                 oa->o_undirty = max(cli->cl_dirty_max, max_in_flight);
 783         }
 784         oa->o_grant = cli->cl_avail_grant;
 785         oa->o_dropped = cli->cl_lost_grant;
 786         cli->cl_lost_grant = 0;
 787         client_obd_list_unlock(&cli->cl_loi_list_lock);
 788         CDEBUG(D_CACHE,"dirty: "LPU64" undirty: %u dropped %u grant: "LPU64"\n",
 789                oa->o_dirty, oa->o_undirty, oa->o_dropped, oa->o_grant);
 790
 791 }
 792
 793 static void osc_update_next_shrink(struct client_obd *cli)
 794 {
 795         cli->cl_next_shrink_grant =
 796                 cfs_time_shift(cli->cl_grant_shrink_interval);
 797         CDEBUG(D_CACHE, "next time %ld to shrink grant \n",
 798                cli->cl_next_shrink_grant);
 799 }
 800
 801 /* caller must hold loi_list_lock */
 802 static void osc_consume_write_grant(struct client_obd *cli,
 803                                     struct brw_page *pga)
 804 {
 805         LASSERT_SPIN_LOCKED(&cli->cl_loi_list_lock.lock);
 806         LASSERT(!(pga->flag & OBD_BRW_FROM_GRANT));
 807         cfs_atomic_inc(&obd_dirty_pages);
 808         cli->cl_dirty += CFS_PAGE_SIZE;
 809         cli->cl_avail_grant -= CFS_PAGE_SIZE;
 810         pga->flag |= OBD_BRW_FROM_GRANT;
 811         CDEBUG(D_CACHE, "using %lu grant credits for brw %p page %p\n",
 812                CFS_PAGE_SIZE, pga, pga->pg);
 813         LASSERT(cli->cl_avail_grant >= 0);
 814         osc_update_next_shrink(cli);
 815 }
 816
 817 /* the companion to osc_consume_write_grant, called when a brw has completed.
 818  * must be called with the loi lock held. */
 819 static void osc_release_write_grant(struct client_obd *cli,
 820                                     struct brw_page *pga, int sent)
 821 {
 822         int blocksize = cli->cl_import->imp_obd->obd_osfs.os_bsize ? : 4096;
 823         ENTRY;
 824
 825         LASSERT_SPIN_LOCKED(&cli->cl_loi_list_lock.lock);
 826         if (!(pga->flag & OBD_BRW_FROM_GRANT)) {
 827                 EXIT;
 828                 return;
 829         }
 830
 831         pga->flag &= ~OBD_BRW_FROM_GRANT;
 832         cfs_atomic_dec(&obd_dirty_pages);
 833         cli->cl_dirty -= CFS_PAGE_SIZE;
 834         if (pga->flag & OBD_BRW_NOCACHE) {
 835                 pga->flag &= ~OBD_BRW_NOCACHE;
 836                 cfs_atomic_dec(&obd_dirty_transit_pages);
 837                 cli->cl_dirty_transit -= CFS_PAGE_SIZE;
 838         }
 839         if (!sent) {
 840                 cli->cl_lost_grant += CFS_PAGE_SIZE;
 841                 CDEBUG(D_CACHE, "lost grant: %lu avail grant: %lu dirty: %lu\n",
 842                        cli->cl_lost_grant, cli->cl_avail_grant, cli->cl_dirty);
 843         } else if (CFS_PAGE_SIZE != blocksize && pga->count != CFS_PAGE_SIZE) {
 844                 /* For short writes we shouldn't count parts of pages that
 845                  * span a whole block on the OST side, or our accounting goes
 846                  * wrong.  Should match the code in filter_grant_check. */
 847                 int offset = pga->off & ~CFS_PAGE_MASK;
 848                 int count = pga->count + (offset & (blocksize - 1));
 849                 int end = (offset + pga->count) & (blocksize - 1);
 850                 if (end)
 851                         count += blocksize - end;
 852
 853                 cli->cl_lost_grant += CFS_PAGE_SIZE - count;
 854                 CDEBUG(D_CACHE, "lost %lu grant: %lu avail: %lu dirty: %lu\n",
 855                        CFS_PAGE_SIZE - count, cli->cl_lost_grant,
 856                        cli->cl_avail_grant, cli->cl_dirty);
 857         }
 858
 859         EXIT;
 860 }
 861
 862 static unsigned long rpcs_in_flight(struct client_obd *cli)
 863 {
 864         return cli->cl_r_in_flight + cli->cl_w_in_flight;
 865 }
 866
 867 /* caller must hold loi_list_lock */
 868 void osc_wake_cache_waiters(struct client_obd *cli)
 869 {
 870         cfs_list_t *l, *tmp;
 871         struct osc_cache_waiter *ocw;
 872
 873         ENTRY;
 874         cfs_list_for_each_safe(l, tmp, &cli->cl_cache_waiters) {
 875                 /* if we can't dirty more, we must wait until some is written */
 876                 if ((cli->cl_dirty + CFS_PAGE_SIZE > cli->cl_dirty_max) ||
 877                    (cfs_atomic_read(&obd_dirty_pages) + 1 >
 878                     obd_max_dirty_pages)) {
 879                         CDEBUG(D_CACHE, "no dirty room: dirty: %ld "
 880                                "osc max %ld, sys max %d\n", cli->cl_dirty,
 881                                cli->cl_dirty_max, obd_max_dirty_pages);
 882                         return;
 883                 }
 884
 885                 /* if still dirty cache but no grant wait for pending RPCs that
 886                  * may yet return us some grant before doing sync writes */
 887                 if (cli->cl_w_in_flight && cli->cl_avail_grant < CFS_PAGE_SIZE) {
 888                         CDEBUG(D_CACHE, "%u BRW writes in flight, no grant\n",
 889                                cli->cl_w_in_flight);
 890                         return;
 891                 }
 892
 893                 ocw = cfs_list_entry(l, struct osc_cache_waiter, ocw_entry);
 894                 cfs_list_del_init(&ocw->ocw_entry);
 895                 if (cli->cl_avail_grant < CFS_PAGE_SIZE) {
 896                         /* no more RPCs in flight to return grant, do sync IO */
 897                         ocw->ocw_rc = -EDQUOT;
 898                         CDEBUG(D_INODE, "wake oap %p for sync\n", ocw->ocw_oap);
 899                 } else {
 900                         osc_consume_write_grant(cli,
 901                                                 &ocw->ocw_oap->oap_brw_page);
 902                 }
 903
 904                 cfs_waitq_signal(&ocw->ocw_waitq);
 905         }
 906
 907         EXIT;
 908 }
 909
 910 static void __osc_update_grant(struct client_obd *cli, obd_size grant)
 911 {
 912         client_obd_list_lock(&cli->cl_loi_list_lock);
 913         cli->cl_avail_grant += grant;
 914         client_obd_list_unlock(&cli->cl_loi_list_lock);
 915 }
 916
 917 static void osc_update_grant(struct client_obd *cli, struct ost_body *body)
 918 {
 919         if (body->oa.o_valid & OBD_MD_FLGRANT) {
 920                 CDEBUG(D_CACHE, "got "LPU64" extra grant\n", body->oa.o_grant);
 921                 __osc_update_grant(cli, body->oa.o_grant);
 922         }
 923 }
 924
 925 static int osc_set_info_async(struct obd_export *exp, obd_count keylen,
 926                               void *key, obd_count vallen, void *val,
 927                               struct ptlrpc_request_set *set);
 928
 929 static int osc_shrink_grant_interpret(const struct lu_env *env,
 930                                       struct ptlrpc_request *req,
 931                                       void *aa, int rc)
 932 {
 933         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
 934         struct obdo *oa = ((struct osc_grant_args *)aa)->aa_oa;
 935         struct ost_body *body;
 936
 937         if (rc != 0) {
 938                 __osc_update_grant(cli, oa->o_grant);
 939                 GOTO(out, rc);
 940         }
 941
 942         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 943         LASSERT(body);
 944         osc_update_grant(cli, body);
 945 out:
 946         OBD_FREE_PTR(oa);
 947         return rc;
 948 }
 949
 950 static void osc_shrink_grant_local(struct client_obd *cli, struct obdo *oa)
 951 {
 952         client_obd_list_lock(&cli->cl_loi_list_lock);
 953         oa->o_grant = cli->cl_avail_grant / 4;
 954         cli->cl_avail_grant -= oa->o_grant;
 955         client_obd_list_unlock(&cli->cl_loi_list_lock);
 956         oa->o_flags |= OBD_FL_SHRINK_GRANT;
 957         osc_update_next_shrink(cli);
 958 }
 959
 960 /* Shrink the current grant, either from some large amount to enough for a
 961  * full set of in-flight RPCs, or if we have already shrunk to that limit
 962  * then to enough for a single RPC.  This avoids keeping more grant than
 963  * needed, and avoids shrinking the grant piecemeal. */
 964 static int osc_shrink_grant(struct client_obd *cli)
 965 {
 966         long target = (cli->cl_max_rpcs_in_flight + 1) *
 967                       cli->cl_max_pages_per_rpc;
 968
 969         client_obd_list_lock(&cli->cl_loi_list_lock);
 970         if (cli->cl_avail_grant <= target)
 971                 target = cli->cl_max_pages_per_rpc;
 972         client_obd_list_unlock(&cli->cl_loi_list_lock);
 973
 974         return osc_shrink_grant_to_target(cli, target);
 975 }
 976
 977 int osc_shrink_grant_to_target(struct client_obd *cli, long target)
 978 {
 979         int    rc = 0;
 980         struct ost_body     *body;
 981         ENTRY;
 982
 983         client_obd_list_lock(&cli->cl_loi_list_lock);
 984         /* Don't shrink if we are already above or below the desired limit
 985          * We don't want to shrink below a single RPC, as that will negatively
 986          * impact block allocation and long-term performance. */
 987         if (target < cli->cl_max_pages_per_rpc)
 988                 target = cli->cl_max_pages_per_rpc;
 989
 990         if (target >= cli->cl_avail_grant) {
 991                 client_obd_list_unlock(&cli->cl_loi_list_lock);
 992                 RETURN(0);
 993         }
 994         client_obd_list_unlock(&cli->cl_loi_list_lock);
 995
 996         OBD_ALLOC_PTR(body);
 997         if (!body)
 998                 RETURN(-ENOMEM);
 999
1000         osc_announce_cached(cli, &body->oa, 0);
1001
1002         client_obd_list_lock(&cli->cl_loi_list_lock);
1003         body->oa.o_grant = cli->cl_avail_grant - target;
1004         cli->cl_avail_grant = target;
1005         client_obd_list_unlock(&cli->cl_loi_list_lock);
1006         body->oa.o_flags |= OBD_FL_SHRINK_GRANT;
1007         osc_update_next_shrink(cli);
1008
1009         rc = osc_set_info_async(cli->cl_import->imp_obd->obd_self_export,
1010                                 sizeof(KEY_GRANT_SHRINK), KEY_GRANT_SHRINK,
1011                                 sizeof(*body), body, NULL);
1012         if (rc != 0)
1013                 __osc_update_grant(cli, body->oa.o_grant);
1014         OBD_FREE_PTR(body);
1015         RETURN(rc);
1016 }
1017
1018 #define GRANT_SHRINK_LIMIT PTLRPC_MAX_BRW_SIZE
1019 static int osc_should_shrink_grant(struct client_obd *client)
1020 {
1021         cfs_time_t time = cfs_time_current();
1022         cfs_time_t next_shrink = client->cl_next_shrink_grant;
1023
1024         if ((client->cl_import->imp_connect_data.ocd_connect_flags &
1025              OBD_CONNECT_GRANT_SHRINK) == 0)
1026                 return 0;
1027
1028         if (cfs_time_aftereq(time, next_shrink - 5 * CFS_TICK)) {
1029                 if (client->cl_import->imp_state == LUSTRE_IMP_FULL &&
1030                     client->cl_avail_grant > GRANT_SHRINK_LIMIT)
1031                         return 1;
1032                 else
1033                         osc_update_next_shrink(client);
1034         }
1035         return 0;
1036 }
1037
1038 static int osc_grant_shrink_grant_cb(struct timeout_item *item, void *data)
1039 {
1040         struct client_obd *client;
1041
1042         cfs_list_for_each_entry(client, &item->ti_obd_list,
1043                                 cl_grant_shrink_list) {
1044                 if (osc_should_shrink_grant(client))
1045                         osc_shrink_grant(client);
1046         }
1047         return 0;
1048 }
1049
1050 static int osc_add_shrink_grant(struct client_obd *client)
1051 {
1052         int rc;
1053
1054         rc = ptlrpc_add_timeout_client(client->cl_grant_shrink_interval,
1055                                        TIMEOUT_GRANT,
1056                                        osc_grant_shrink_grant_cb, NULL,
1057                                        &client->cl_grant_shrink_list);
1058         if (rc) {
1059                 CERROR("add grant client %s error %d\n",
1060                         client->cl_import->imp_obd->obd_name, rc);
1061                 return rc;
1062         }
1063         CDEBUG(D_CACHE, "add grant client %s \n",
1064                client->cl_import->imp_obd->obd_name);
1065         osc_update_next_shrink(client);
1066         return 0;
1067 }
1068
1069 static int osc_del_shrink_grant(struct client_obd *client)
1070 {
1071         return ptlrpc_del_timeout_client(&client->cl_grant_shrink_list,
1072                                          TIMEOUT_GRANT);
1073 }
1074
1075 static void osc_init_grant(struct client_obd *cli, struct obd_connect_data *ocd)
1076 {
1077         /*
1078          * ocd_grant is the total grant amount we're expect to hold: if we've
1079          * been evicted, it's the new avail_grant amount, cl_dirty will drop
1080          * to 0 as inflight RPCs fail out; otherwise, it's avail_grant + dirty.
1081          *
1082          * race is tolerable here: if we're evicted, but imp_state already
1083          * left EVICTED state, then cl_dirty must be 0 already.
1084          */
1085         client_obd_list_lock(&cli->cl_loi_list_lock);
1086         if (cli->cl_import->imp_state == LUSTRE_IMP_EVICTED)
1087                 cli->cl_avail_grant = ocd->ocd_grant;
1088         else
1089                 cli->cl_avail_grant = ocd->ocd_grant - cli->cl_dirty;
1090         client_obd_list_unlock(&cli->cl_loi_list_lock);
1091
1092         CDEBUG(D_CACHE, "setting cl_avail_grant: %ld cl_lost_grant: %ld \n",
1093                cli->cl_avail_grant, cli->cl_lost_grant);
1094         LASSERT(cli->cl_avail_grant >= 0);
1095
1096         if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT_SHRINK &&
1097             cfs_list_empty(&cli->cl_grant_shrink_list))
1098                 osc_add_shrink_grant(cli);
1099 }
1100
1101 /* We assume that the reason this OSC got a short read is because it read
1102  * beyond the end of a stripe file; i.e. lustre is reading a sparse file
1103  * via the LOV, and it _knows_ it's reading inside the file, it's just that
1104  * this stripe never got written at or beyond this stripe offset yet. */
1105 static void handle_short_read(int nob_read, obd_count page_count,
1106                               struct brw_page **pga)
1107 {
1108         char *ptr;
1109         int i = 0;
1110
1111         /* skip bytes read OK */
1112         while (nob_read > 0) {
1113                 LASSERT (page_count > 0);
1114
1115                 if (pga[i]->count > nob_read) {
1116                         /* EOF inside this page */
1117                         ptr = cfs_kmap(pga[i]->pg) +
1118                                 (pga[i]->off & ~CFS_PAGE_MASK);
1119                         memset(ptr + nob_read, 0, pga[i]->count - nob_read);
1120                         cfs_kunmap(pga[i]->pg);
1121                         page_count--;
1122                         i++;
1123                         break;
1124                 }
1125
1126                 nob_read -= pga[i]->count;
1127                 page_count--;
1128                 i++;
1129         }
1130
1131         /* zero remaining pages */
1132         while (page_count-- > 0) {
1133                 ptr = cfs_kmap(pga[i]->pg) + (pga[i]->off & ~CFS_PAGE_MASK);
1134                 memset(ptr, 0, pga[i]->count);
1135                 cfs_kunmap(pga[i]->pg);
1136                 i++;
1137         }
1138 }
1139
1140 static int check_write_rcs(struct ptlrpc_request *req,
1141                            int requested_nob, int niocount,
1142                            obd_count page_count, struct brw_page **pga)
1143 {
1144         int     i;
1145         __u32   *remote_rcs;
1146
1147         remote_rcs = req_capsule_server_sized_get(&req->rq_pill, &RMF_RCS,
1148                                                   sizeof(*remote_rcs) *
1149                                                   niocount);
1150         if (remote_rcs == NULL) {
1151                 CDEBUG(D_INFO, "Missing/short RC vector on BRW_WRITE reply\n");
1152                 return(-EPROTO);
1153         }
1154
1155         /* return error if any niobuf was in error */
1156         for (i = 0; i < niocount; i++) {
1157                 if (remote_rcs[i] < 0)
1158                         return(remote_rcs[i]);
1159
1160                 if (remote_rcs[i] != 0) {
1161                         CDEBUG(D_INFO, "rc[%d] invalid (%d) req %p\n",
1162                                 i, remote_rcs[i], req);
1163                         return(-EPROTO);
1164                 }
1165         }
1166
1167         if (req->rq_bulk->bd_nob_transferred != requested_nob) {
1168                 CERROR("Unexpected # bytes transferred: %d (requested %d)\n",
1169                        req->rq_bulk->bd_nob_transferred, requested_nob);
1170                 return(-EPROTO);
1171         }
1172
1173         return (0);
1174 }
1175
1176 static inline int can_merge_pages(struct brw_page *p1, struct brw_page *p2)
1177 {
1178         if (p1->flag != p2->flag) {
1179                 unsigned mask = ~(OBD_BRW_FROM_GRANT|
1180                                   OBD_BRW_NOCACHE|OBD_BRW_SYNC);
1181
1182                 /* warn if we try to combine flags that we don't know to be
1183                  * safe to combine */
1184                 if ((p1->flag & mask) != (p2->flag & mask))
1185                         CERROR("is it ok to have flags 0x%x and 0x%x in the "
1186                                "same brw?\n", p1->flag, p2->flag);
1187                 return 0;
1188         }
1189
1190         return (p1->off + p1->count == p2->off);
1191 }
1192
1193 static obd_count osc_checksum_bulk(int nob, obd_count pg_count,
1194                                    struct brw_page **pga, int opc,
1195                                    cksum_type_t cksum_type)
1196 {
1197         __u32 cksum;
1198         int i = 0;
1199
1200         LASSERT (pg_count > 0);
1201         cksum = init_checksum(cksum_type);
1202         while (nob > 0 && pg_count > 0) {
1203                 unsigned char *ptr = cfs_kmap(pga[i]->pg);
1204                 int off = pga[i]->off & ~CFS_PAGE_MASK;
1205                 int count = pga[i]->count > nob ? nob : pga[i]->count;
1206
1207                 /* corrupt the data before we compute the checksum, to
1208                  * simulate an OST->client data error */
1209                 if (i == 0 && opc == OST_READ &&
1210                     OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_RECEIVE))
1211                         memcpy(ptr + off, "bad1", min(4, nob));
1212                 cksum = compute_checksum(cksum, ptr + off, count, cksum_type);
1213                 cfs_kunmap(pga[i]->pg);
1214                 LL_CDEBUG_PAGE(D_PAGE, pga[i]->pg, "off %d checksum %x\n",
1215                                off, cksum);
1216
1217                 nob -= pga[i]->count;
1218                 pg_count--;
1219                 i++;
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 {
1235         struct ptlrpc_request   *req;
1236         struct ptlrpc_bulk_desc *desc;
1237         struct ost_body         *body;
1238         struct obd_ioobj        *ioobj;
1239         struct niobuf_remote    *niobuf;
1240         int niocount, i, requested_nob, opc, rc;
1241         struct osc_brw_async_args *aa;
1242         struct req_capsule      *pill;
1243         struct brw_page *pg_prev;
1244
1245         ENTRY;
1246         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_PREP_REQ))
1247                 RETURN(-ENOMEM); /* Recoverable */
1248         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_PREP_REQ2))
1249                 RETURN(-EINVAL); /* Fatal */
1250
1251         if ((cmd & OBD_BRW_WRITE) != 0) {
1252                 opc = OST_WRITE;
1253                 req = ptlrpc_request_alloc_pool(cli->cl_import,
1254                                                 cli->cl_import->imp_rq_pool,
1255                                                 &RQF_OST_BRW_WRITE);
1256         } else {
1257                 opc = OST_READ;
1258                 req = ptlrpc_request_alloc(cli->cl_import, &RQF_OST_BRW_READ);
1259         }
1260         if (req == NULL)
1261                 RETURN(-ENOMEM);
1262
1263         for (niocount = i = 1; i < page_count; i++) {
1264                 if (!can_merge_pages(pga[i - 1], pga[i]))
1265                         niocount++;
1266         }
1267
1268         pill = &req->rq_pill;
1269         req_capsule_set_size(pill, &RMF_OBD_IOOBJ, RCL_CLIENT,
1270                              sizeof(*ioobj));
1271         req_capsule_set_size(pill, &RMF_NIOBUF_REMOTE, RCL_CLIENT,
1272                              niocount * sizeof(*niobuf));
1273         osc_set_capa_size(req, &RMF_CAPA1, ocapa);
1274
1275         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, opc);
1276         if (rc) {
1277                 ptlrpc_request_free(req);
1278                 RETURN(rc);
1279         }
1280         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
1281         ptlrpc_at_set_req_timeout(req);
1282
1283         if (opc == OST_WRITE)
1284                 desc = ptlrpc_prep_bulk_imp(req, page_count,
1285                                             BULK_GET_SOURCE, OST_BULK_PORTAL);
1286         else
1287                 desc = ptlrpc_prep_bulk_imp(req, page_count,
1288                                             BULK_PUT_SINK, OST_BULK_PORTAL);
1289
1290         if (desc == NULL)
1291                 GOTO(out, rc = -ENOMEM);
1292         /* NB request now owns desc and will free it when it gets freed */
1293
1294         body = req_capsule_client_get(pill, &RMF_OST_BODY);
1295         ioobj = req_capsule_client_get(pill, &RMF_OBD_IOOBJ);
1296         niobuf = req_capsule_client_get(pill, &RMF_NIOBUF_REMOTE);
1297         LASSERT(body != NULL && ioobj != NULL && niobuf != NULL);
1298
1299         lustre_set_wire_obdo(&body->oa, oa);
1300
1301         obdo_to_ioobj(oa, ioobj);
1302         ioobj->ioo_bufcnt = niocount;
1303         osc_pack_capa(req, body, ocapa);
1304         LASSERT (page_count > 0);
1305         pg_prev = pga[0];
1306         for (requested_nob = i = 0; i < page_count; i++, niobuf++) {
1307                 struct brw_page *pg = pga[i];
1308
1309                 LASSERT(pg->count > 0);
1310                 LASSERTF((pg->off & ~CFS_PAGE_MASK) + pg->count <= CFS_PAGE_SIZE,
1311                          "i: %d pg: %p off: "LPU64", count: %u\n", i, pg,
1312                          pg->off, pg->count);
1313 #ifdef __linux__
1314                 LASSERTF(i == 0 || pg->off > pg_prev->off,
1315                          "i %d p_c %u pg %p [pri %lu ind %lu] off "LPU64
1316                          " prev_pg %p [pri %lu ind %lu] off "LPU64"\n",
1317                          i, page_count,
1318                          pg->pg, page_private(pg->pg), pg->pg->index, pg->off,
1319                          pg_prev->pg, page_private(pg_prev->pg),
1320                          pg_prev->pg->index, pg_prev->off);
1321 #else
1322                 LASSERTF(i == 0 || pg->off > pg_prev->off,
1323                          "i %d p_c %u\n", i, page_count);
1324 #endif
1325                 LASSERT((pga[0]->flag & OBD_BRW_SRVLOCK) ==
1326                         (pg->flag & OBD_BRW_SRVLOCK));
1327
1328                 ptlrpc_prep_bulk_page(desc, pg->pg, pg->off & ~CFS_PAGE_MASK,
1329                                       pg->count);
1330                 requested_nob += pg->count;
1331
1332                 if (i > 0 && can_merge_pages(pg_prev, pg)) {
1333                         niobuf--;
1334                         niobuf->len += pg->count;
1335                 } else {
1336                         niobuf->offset = pg->off;
1337                         niobuf->len    = pg->count;
1338                         niobuf->flags  = pg->flag;
1339                 }
1340                 pg_prev = pg;
1341         }
1342
1343         LASSERTF((void *)(niobuf - niocount) ==
1344                 req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE),
1345                 "want %p - real %p\n", req_capsule_client_get(&req->rq_pill,
1346                 &RMF_NIOBUF_REMOTE), (void *)(niobuf - niocount));
1347
1348         osc_announce_cached(cli, &body->oa, opc == OST_WRITE ? requested_nob:0);
1349         if (osc_should_shrink_grant(cli))
1350                 osc_shrink_grant_local(cli, &body->oa);
1351
1352         /* size[REQ_REC_OFF] still sizeof (*body) */
1353         if (opc == OST_WRITE) {
1354                 if (unlikely(cli->cl_checksum) &&
1355                     !sptlrpc_flavor_has_bulk(&req->rq_flvr)) {
1356                         /* store cl_cksum_type in a local variable since
1357                          * it can be changed via lprocfs */
1358                         cksum_type_t cksum_type = cli->cl_cksum_type;
1359
1360                         if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0) {
1361                                 oa->o_flags &= OBD_FL_LOCAL_MASK;
1362                                 body->oa.o_flags = 0;
1363                         }
1364                         body->oa.o_flags |= cksum_type_pack(cksum_type);
1365                         body->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1366                         body->oa.o_cksum = osc_checksum_bulk(requested_nob,
1367                                                              page_count, pga,
1368                                                              OST_WRITE,
1369                                                              cksum_type);
1370                         CDEBUG(D_PAGE, "checksum at write origin: %x\n",
1371                                body->oa.o_cksum);
1372                         /* save this in 'oa', too, for later checking */
1373                         oa->o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1374                         oa->o_flags |= cksum_type_pack(cksum_type);
1375                 } else {
1376                         /* clear out the checksum flag, in case this is a
1377                          * resend but cl_checksum is no longer set. b=11238 */
1378                         oa->o_valid &= ~OBD_MD_FLCKSUM;
1379                 }
1380                 oa->o_cksum = body->oa.o_cksum;
1381                 /* 1 RC per niobuf */
1382                 req_capsule_set_size(pill, &RMF_RCS, RCL_SERVER,
1383                                      sizeof(__u32) * niocount);
1384         } else {
1385                 if (unlikely(cli->cl_checksum) &&
1386                     !sptlrpc_flavor_has_bulk(&req->rq_flvr)) {
1387                         if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0)
1388                                 body->oa.o_flags = 0;
1389                         body->oa.o_flags |= cksum_type_pack(cli->cl_cksum_type);
1390                         body->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1391                 }
1392         }
1393         ptlrpc_request_set_replen(req);
1394
1395         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
1396         aa = ptlrpc_req_async_args(req);
1397         aa->aa_oa = oa;
1398         aa->aa_requested_nob = requested_nob;
1399         aa->aa_nio_count = niocount;
1400         aa->aa_page_count = page_count;
1401         aa->aa_resends = 0;
1402         aa->aa_ppga = pga;
1403         aa->aa_cli = cli;
1404         CFS_INIT_LIST_HEAD(&aa->aa_oaps);
1405         if (ocapa && reserve)
1406                 aa->aa_ocapa = capa_get(ocapa);
1407
1408         *reqp = req;
1409         RETURN(0);
1410
1411  out:
1412         ptlrpc_req_finished(req);
1413         RETURN(rc);
1414 }
1415
1416 static int check_write_checksum(struct obdo *oa, const lnet_process_id_t *peer,
1417                                 __u32 client_cksum, __u32 server_cksum, int nob,
1418                                 obd_count page_count, struct brw_page **pga,
1419                                 cksum_type_t client_cksum_type)
1420 {
1421         __u32 new_cksum;
1422         char *msg;
1423         cksum_type_t cksum_type;
1424
1425         if (server_cksum == client_cksum) {
1426                 CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
1427                 return 0;
1428         }
1429
1430         if (oa->o_valid & OBD_MD_FLFLAGS)
1431                 cksum_type = cksum_type_unpack(oa->o_flags);
1432         else
1433                 cksum_type = OBD_CKSUM_CRC32;
1434
1435         new_cksum = osc_checksum_bulk(nob, page_count, pga, OST_WRITE,
1436                                       cksum_type);
1437
1438         if (cksum_type != client_cksum_type)
1439                 msg = "the server did not use the checksum type specified in "
1440                       "the original request - likely a protocol problem";
1441         else if (new_cksum == server_cksum)
1442                 msg = "changed on the client after we checksummed it - "
1443                       "likely false positive due to mmap IO (bug 11742)";
1444         else if (new_cksum == client_cksum)
1445                 msg = "changed in transit before arrival at OST";
1446         else
1447                 msg = "changed in transit AND doesn't match the original - "
1448                       "likely false positive due to mmap IO (bug 11742)";
1449
1450         LCONSOLE_ERROR_MSG(0x132, "BAD WRITE CHECKSUM: %s: from %s inode "DFID
1451                            " object "LPU64"/"LPU64" extent ["LPU64"-"LPU64"]\n",
1452                            msg, libcfs_nid2str(peer->nid),
1453                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_seq : (__u64)0,
1454                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_oid : 0,
1455                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_ver : 0,
1456                            oa->o_id,
1457                            oa->o_valid & OBD_MD_FLGROUP ? oa->o_seq : (__u64)0,
1458                            pga[0]->off,
1459                            pga[page_count-1]->off + pga[page_count-1]->count - 1);
1460         CERROR("original client csum %x (type %x), server csum %x (type %x), "
1461                "client csum now %x\n", client_cksum, client_cksum_type,
1462                server_cksum, cksum_type, new_cksum);
1463         return 1;
1464 }
1465
1466 /* Note rc enters this function as number of bytes transferred */
1467 static int osc_brw_fini_request(struct ptlrpc_request *req, int rc)
1468 {
1469         struct osc_brw_async_args *aa = (void *)&req->rq_async_args;
1470         const lnet_process_id_t *peer =
1471                         &req->rq_import->imp_connection->c_peer;
1472         struct client_obd *cli = aa->aa_cli;
1473         struct ost_body *body;
1474         __u32 client_cksum = 0;
1475         ENTRY;
1476
1477         if (rc < 0 && rc != -EDQUOT) {
1478                 DEBUG_REQ(D_INFO, req, "Failed request with rc = %d\n", rc);
1479                 RETURN(rc);
1480         }
1481
1482         LASSERTF(req->rq_repmsg != NULL, "rc = %d\n", rc);
1483         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
1484         if (body == NULL) {
1485                 DEBUG_REQ(D_INFO, req, "Can't unpack body\n");
1486                 RETURN(-EPROTO);
1487         }
1488
1489 #ifdef HAVE_QUOTA_SUPPORT
1490         /* set/clear over quota flag for a uid/gid */
1491         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE &&
1492             body->oa.o_valid & (OBD_MD_FLUSRQUOTA | OBD_MD_FLGRPQUOTA)) {
1493                 unsigned int qid[MAXQUOTAS] = { body->oa.o_uid, body->oa.o_gid };
1494
1495                 CDEBUG(D_QUOTA, "setdq for [%u %u] with valid "LPX64", flags %x\n",
1496                        body->oa.o_uid, body->oa.o_gid, body->oa.o_valid,
1497                        body->oa.o_flags);
1498                 lquota_setdq(quota_interface, cli, qid, body->oa.o_valid,
1499                              body->oa.o_flags);
1500         }
1501 #endif
1502
1503         osc_update_grant(cli, body);
1504
1505         if (rc < 0)
1506                 RETURN(rc);
1507
1508         if (aa->aa_oa->o_valid & OBD_MD_FLCKSUM)
1509                 client_cksum = aa->aa_oa->o_cksum; /* save for later */
1510
1511         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE) {
1512                 if (rc > 0) {
1513                         CERROR("Unexpected +ve rc %d\n", rc);
1514                         RETURN(-EPROTO);
1515                 }
1516                 LASSERT(req->rq_bulk->bd_nob == aa->aa_requested_nob);
1517
1518                 if (sptlrpc_cli_unwrap_bulk_write(req, req->rq_bulk))
1519                         RETURN(-EAGAIN);
1520
1521                 if ((aa->aa_oa->o_valid & OBD_MD_FLCKSUM) && client_cksum &&
1522                     check_write_checksum(&body->oa, peer, client_cksum,
1523                                          body->oa.o_cksum, aa->aa_requested_nob,
1524                                          aa->aa_page_count, aa->aa_ppga,
1525                                          cksum_type_unpack(aa->aa_oa->o_flags)))
1526                         RETURN(-EAGAIN);
1527
1528                 rc = check_write_rcs(req, aa->aa_requested_nob,aa->aa_nio_count,
1529                                      aa->aa_page_count, aa->aa_ppga);
1530                 GOTO(out, rc);
1531         }
1532
1533         /* The rest of this function executes only for OST_READs */
1534
1535         /* if unwrap_bulk failed, return -EAGAIN to retry */
1536         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk, rc);
1537         if (rc < 0)
1538                 GOTO(out, rc = -EAGAIN);
1539
1540         if (rc > aa->aa_requested_nob) {
1541                 CERROR("Unexpected rc %d (%d requested)\n", rc,
1542                        aa->aa_requested_nob);
1543                 RETURN(-EPROTO);
1544         }
1545
1546         if (rc != req->rq_bulk->bd_nob_transferred) {
1547                 CERROR ("Unexpected rc %d (%d transferred)\n",
1548                         rc, req->rq_bulk->bd_nob_transferred);
1549                 return (-EPROTO);
1550         }
1551
1552         if (rc < aa->aa_requested_nob)
1553                 handle_short_read(rc, aa->aa_page_count, aa->aa_ppga);
1554
1555         if (body->oa.o_valid & OBD_MD_FLCKSUM) {
1556                 static int cksum_counter;
1557                 __u32      server_cksum = body->oa.o_cksum;
1558                 char      *via;
1559                 char      *router;
1560                 cksum_type_t cksum_type;
1561
1562                 if (body->oa.o_valid & OBD_MD_FLFLAGS)
1563                         cksum_type = cksum_type_unpack(body->oa.o_flags);
1564                 else
1565                         cksum_type = OBD_CKSUM_CRC32;
1566                 client_cksum = osc_checksum_bulk(rc, aa->aa_page_count,
1567                                                  aa->aa_ppga, OST_READ,
1568                                                  cksum_type);
1569
1570                 if (peer->nid == req->rq_bulk->bd_sender) {
1571                         via = router = "";
1572                 } else {
1573                         via = " via ";
1574                         router = libcfs_nid2str(req->rq_bulk->bd_sender);
1575                 }
1576
1577                 if (server_cksum == ~0 && rc > 0) {
1578                         CERROR("Protocol error: server %s set the 'checksum' "
1579                                "bit, but didn't send a checksum.  Not fatal, "
1580                                "but please notify on http://bugzilla.lustre.org/\n",
1581                                libcfs_nid2str(peer->nid));
1582                 } else if (server_cksum != client_cksum) {
1583                         LCONSOLE_ERROR_MSG(0x133, "%s: BAD READ CHECKSUM: from "
1584                                            "%s%s%s inode "DFID" object "
1585                                            LPU64"/"LPU64" extent "
1586                                            "["LPU64"-"LPU64"]\n",
1587                                            req->rq_import->imp_obd->obd_name,
1588                                            libcfs_nid2str(peer->nid),
1589                                            via, router,
1590                                            body->oa.o_valid & OBD_MD_FLFID ?
1591                                                 body->oa.o_parent_seq : (__u64)0,
1592                                            body->oa.o_valid & OBD_MD_FLFID ?
1593                                                 body->oa.o_parent_oid : 0,
1594                                            body->oa.o_valid & OBD_MD_FLFID ?
1595                                                 body->oa.o_parent_ver : 0,
1596                                            body->oa.o_id,
1597                                            body->oa.o_valid & OBD_MD_FLGROUP ?
1598                                                 body->oa.o_seq : (__u64)0,
1599                                            aa->aa_ppga[0]->off,
1600                                            aa->aa_ppga[aa->aa_page_count-1]->off +
1601                                            aa->aa_ppga[aa->aa_page_count-1]->count -
1602                                                                         1);
1603                         CERROR("client %x, server %x, cksum_type %x\n",
1604                                client_cksum, server_cksum, cksum_type);
1605                         cksum_counter = 0;
1606                         aa->aa_oa->o_cksum = client_cksum;
1607                         rc = -EAGAIN;
1608                 } else {
1609                         cksum_counter++;
1610                         CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
1611                         rc = 0;
1612                 }
1613         } else if (unlikely(client_cksum)) {
1614                 static int cksum_missed;
1615
1616                 cksum_missed++;
1617                 if ((cksum_missed & (-cksum_missed)) == cksum_missed)
1618                         CERROR("Checksum %u requested from %s but not sent\n",
1619                                cksum_missed, libcfs_nid2str(peer->nid));
1620         } else {
1621                 rc = 0;
1622         }
1623 out:
1624         if (rc >= 0)
1625                 lustre_get_wire_obdo(aa->aa_oa, &body->oa);
1626
1627         RETURN(rc);
1628 }
1629
1630 static int osc_brw_internal(int cmd, struct obd_export *exp, struct obdo *oa,
1631                             struct lov_stripe_md *lsm,
1632                             obd_count page_count, struct brw_page **pga,
1633                             struct obd_capa *ocapa)
1634 {
1635         struct ptlrpc_request *req;
1636         int                    rc;
1637         cfs_waitq_t            waitq;
1638         int                    resends = 0;
1639         struct l_wait_info     lwi;
1640
1641         ENTRY;
1642
1643         cfs_waitq_init(&waitq);
1644
1645 restart_bulk:
1646         rc = osc_brw_prep_request(cmd, &exp->exp_obd->u.cli, oa, lsm,
1647                                   page_count, pga, &req, ocapa, 0);
1648         if (rc != 0)
1649                 return (rc);
1650
1651         rc = ptlrpc_queue_wait(req);
1652
1653         if (rc == -ETIMEDOUT && req->rq_resend) {
1654                 DEBUG_REQ(D_HA, req,  "BULK TIMEOUT");
1655                 ptlrpc_req_finished(req);
1656                 goto restart_bulk;
1657         }
1658
1659         rc = osc_brw_fini_request(req, rc);
1660
1661         ptlrpc_req_finished(req);
1662         if (osc_recoverable_error(rc)) {
1663                 resends++;
1664                 if (!osc_should_resend(resends, &exp->exp_obd->u.cli)) {
1665                         CERROR("too many resend retries, returning error\n");
1666                         RETURN(-EIO);
1667                 }
1668
1669                 lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(resends), NULL, NULL, NULL);
1670                 l_wait_event(waitq, 0, &lwi);
1671
1672                 goto restart_bulk;
1673         }
1674
1675         RETURN (rc);
1676 }
1677
1678 int osc_brw_redo_request(struct ptlrpc_request *request,
1679                          struct osc_brw_async_args *aa)
1680 {
1681         struct ptlrpc_request *new_req;
1682         struct ptlrpc_request_set *set = request->rq_set;
1683         struct osc_brw_async_args *new_aa;
1684         struct osc_async_page *oap;
1685         int rc = 0;
1686         ENTRY;
1687
1688         if (!osc_should_resend(aa->aa_resends, aa->aa_cli)) {
1689                 CERROR("too many resent retries, returning error\n");
1690                 RETURN(-EIO);
1691         }
1692
1693         DEBUG_REQ(D_ERROR, request, "redo for recoverable error");
1694
1695         rc = osc_brw_prep_request(lustre_msg_get_opc(request->rq_reqmsg) ==
1696                                         OST_WRITE ? OBD_BRW_WRITE :OBD_BRW_READ,
1697                                   aa->aa_cli, aa->aa_oa,
1698                                   NULL /* lsm unused by osc currently */,
1699                                   aa->aa_page_count, aa->aa_ppga,
1700                                   &new_req, aa->aa_ocapa, 0);
1701         if (rc)
1702                 RETURN(rc);
1703
1704         client_obd_list_lock(&aa->aa_cli->cl_loi_list_lock);
1705
1706         cfs_list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
1707                 if (oap->oap_request != NULL) {
1708                         LASSERTF(request == oap->oap_request,
1709                                  "request %p != oap_request %p\n",
1710                                  request, oap->oap_request);
1711                         if (oap->oap_interrupted) {
1712                                 client_obd_list_unlock(&aa->aa_cli->cl_loi_list_lock);
1713                                 ptlrpc_req_finished(new_req);
1714                                 RETURN(-EINTR);
1715                         }
1716                 }
1717         }
1718         /* New request takes over pga and oaps from old request.
1719          * Note that copying a list_head doesn't work, need to move it... */
1720         aa->aa_resends++;
1721         new_req->rq_interpret_reply = request->rq_interpret_reply;
1722         new_req->rq_async_args = request->rq_async_args;
1723         new_req->rq_sent = cfs_time_current_sec() + aa->aa_resends;
1724
1725         new_aa = ptlrpc_req_async_args(new_req);
1726
1727         CFS_INIT_LIST_HEAD(&new_aa->aa_oaps);
1728         cfs_list_splice(&aa->aa_oaps, &new_aa->aa_oaps);
1729         CFS_INIT_LIST_HEAD(&aa->aa_oaps);
1730
1731         cfs_list_for_each_entry(oap, &new_aa->aa_oaps, oap_rpc_item) {
1732                 if (oap->oap_request) {
1733                         ptlrpc_req_finished(oap->oap_request);
1734                         oap->oap_request = ptlrpc_request_addref(new_req);
1735                 }
1736         }
1737
1738         new_aa->aa_ocapa = aa->aa_ocapa;
1739         aa->aa_ocapa = NULL;
1740
1741         /* use ptlrpc_set_add_req is safe because interpret functions work
1742          * in check_set context. only one way exist with access to request
1743          * from different thread got -EINTR - this way protected with
1744          * cl_loi_list_lock */
1745         ptlrpc_set_add_req(set, new_req);
1746
1747         client_obd_list_unlock(&aa->aa_cli->cl_loi_list_lock);
1748
1749         DEBUG_REQ(D_INFO, new_req, "new request");
1750         RETURN(0);
1751 }
1752
1753 /*
1754  * ugh, we want disk allocation on the target to happen in offset order.  we'll
1755  * follow sedgewicks advice and stick to the dead simple shellsort -- it'll do
1756  * fine for our small page arrays and doesn't require allocation.  its an
1757  * insertion sort that swaps elements that are strides apart, shrinking the
1758  * stride down until its '1' and the array is sorted.
1759  */
1760 static void sort_brw_pages(struct brw_page **array, int num)
1761 {
1762         int stride, i, j;
1763         struct brw_page *tmp;
1764
1765         if (num == 1)
1766                 return;
1767         for (stride = 1; stride < num ; stride = (stride * 3) + 1)
1768                 ;
1769
1770         do {
1771                 stride /= 3;
1772                 for (i = stride ; i < num ; i++) {
1773                         tmp = array[i];
1774                         j = i;
1775                         while (j >= stride && array[j - stride]->off > tmp->off) {
1776                                 array[j] = array[j - stride];
1777                                 j -= stride;
1778                         }
1779                         array[j] = tmp;
1780                 }
1781         } while (stride > 1);
1782 }
1783
1784 static obd_count max_unfragmented_pages(struct brw_page **pg, obd_count pages)
1785 {
1786         int count = 1;
1787         int offset;
1788         int i = 0;
1789
1790         LASSERT (pages > 0);
1791         offset = pg[i]->off & ~CFS_PAGE_MASK;
1792
1793         for (;;) {
1794                 pages--;
1795                 if (pages == 0)         /* that's all */
1796                         return count;
1797
1798                 if (offset + pg[i]->count < CFS_PAGE_SIZE)
1799                         return count;   /* doesn't end on page boundary */
1800
1801                 i++;
1802                 offset = pg[i]->off & ~CFS_PAGE_MASK;
1803                 if (offset != 0)        /* doesn't start on page boundary */
1804                         return count;
1805
1806                 count++;
1807         }
1808 }
1809
1810 static struct brw_page **osc_build_ppga(struct brw_page *pga, obd_count count)
1811 {
1812         struct brw_page **ppga;
1813         int i;
1814
1815         OBD_ALLOC(ppga, sizeof(*ppga) * count);
1816         if (ppga == NULL)
1817                 return NULL;
1818
1819         for (i = 0; i < count; i++)
1820                 ppga[i] = pga + i;
1821         return ppga;
1822 }
1823
1824 static void osc_release_ppga(struct brw_page **ppga, obd_count count)
1825 {
1826         LASSERT(ppga != NULL);
1827         OBD_FREE(ppga, sizeof(*ppga) * count);
1828 }
1829
1830 static int osc_brw(int cmd, struct obd_export *exp, struct obd_info *oinfo,
1831                    obd_count page_count, struct brw_page *pga,
1832                    struct obd_trans_info *oti)
1833 {
1834         struct obdo *saved_oa = NULL;
1835         struct brw_page **ppga, **orig;
1836         struct obd_import *imp = class_exp2cliimp(exp);
1837         struct client_obd *cli;
1838         int rc, page_count_orig;
1839         ENTRY;
1840
1841         LASSERT((imp != NULL) && (imp->imp_obd != NULL));
1842         cli = &imp->imp_obd->u.cli;
1843
1844         if (cmd & OBD_BRW_CHECK) {
1845                 /* The caller just wants to know if there's a chance that this
1846                  * I/O can succeed */
1847
1848                 if (imp->imp_invalid)
1849                         RETURN(-EIO);
1850                 RETURN(0);
1851         }
1852
1853         /* test_brw with a failed create can trip this, maybe others. */
1854         LASSERT(cli->cl_max_pages_per_rpc);
1855
1856         rc = 0;
1857
1858         orig = ppga = osc_build_ppga(pga, page_count);
1859         if (ppga == NULL)
1860                 RETURN(-ENOMEM);
1861         page_count_orig = page_count;
1862
1863         sort_brw_pages(ppga, page_count);
1864         while (page_count) {
1865                 obd_count pages_per_brw;
1866
1867                 if (page_count > cli->cl_max_pages_per_rpc)
1868                         pages_per_brw = cli->cl_max_pages_per_rpc;
1869                 else
1870                         pages_per_brw = page_count;
1871
1872                 pages_per_brw = max_unfragmented_pages(ppga, pages_per_brw);
1873
1874                 if (saved_oa != NULL) {
1875                         /* restore previously saved oa */
1876                         *oinfo->oi_oa = *saved_oa;
1877                 } else if (page_count > pages_per_brw) {
1878                         /* save a copy of oa (brw will clobber it) */
1879                         OBDO_ALLOC(saved_oa);
1880                         if (saved_oa == NULL)
1881                                 GOTO(out, rc = -ENOMEM);
1882                         *saved_oa = *oinfo->oi_oa;
1883                 }
1884
1885                 rc = osc_brw_internal(cmd, exp, oinfo->oi_oa, oinfo->oi_md,
1886                                       pages_per_brw, ppga, oinfo->oi_capa);
1887
1888                 if (rc != 0)
1889                         break;
1890
1891                 page_count -= pages_per_brw;
1892                 ppga += pages_per_brw;
1893         }
1894
1895 out:
1896         osc_release_ppga(orig, page_count_orig);
1897
1898         if (saved_oa != NULL)
1899                 OBDO_FREE(saved_oa);
1900
1901         RETURN(rc);
1902 }
1903
1904 /* The companion to osc_enter_cache(), called when @oap is no longer part of
1905  * the dirty accounting.  Writeback completes or truncate happens before
1906  * writing starts.  Must be called with the loi lock held. */
1907 static void osc_exit_cache(struct client_obd *cli, struct osc_async_page *oap,
1908                            int sent)
1909 {
1910         osc_release_write_grant(cli, &oap->oap_brw_page, sent);
1911 }
1912
1913
1914 /* This maintains the lists of pending pages to read/write for a given object
1915  * (lop).  This is used by osc_check_rpcs->osc_next_loi() and loi_list_maint()
1916  * to quickly find objects that are ready to send an RPC. */
1917 static int lop_makes_rpc(struct client_obd *cli, struct loi_oap_pages *lop,
1918                          int cmd)
1919 {
1920         int optimal;
1921         ENTRY;
1922
1923         if (lop->lop_num_pending == 0)
1924                 RETURN(0);
1925
1926         /* if we have an invalid import we want to drain the queued pages
1927          * by forcing them through rpcs that immediately fail and complete
1928          * the pages.  recovery relies on this to empty the queued pages
1929          * before canceling the locks and evicting down the llite pages */
1930         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
1931                 RETURN(1);
1932
1933         /* stream rpcs in queue order as long as as there is an urgent page
1934          * queued.  this is our cheap solution for good batching in the case
1935          * where writepage marks some random page in the middle of the file
1936          * as urgent because of, say, memory pressure */
1937         if (!cfs_list_empty(&lop->lop_urgent)) {
1938                 CDEBUG(D_CACHE, "urgent request forcing RPC\n");
1939                 RETURN(1);
1940         }
1941         /* fire off rpcs when we have 'optimal' rpcs as tuned for the wire. */
1942         optimal = cli->cl_max_pages_per_rpc;
1943         if (cmd & OBD_BRW_WRITE) {
1944                 /* trigger a write rpc stream as long as there are dirtiers
1945                  * waiting for space.  as they're waiting, they're not going to
1946                  * create more pages to coallesce with what's waiting.. */
1947                 if (!cfs_list_empty(&cli->cl_cache_waiters)) {
1948                         CDEBUG(D_CACHE, "cache waiters forcing RPC\n");
1949                         RETURN(1);
1950                 }
1951                 /* +16 to avoid triggering rpcs that would want to include pages
1952                  * that are being queued but which can't be made ready until
1953                  * the queuer finishes with the page. this is a wart for
1954                  * llite::commit_write() */
1955                 optimal += 16;
1956         }
1957         if (lop->lop_num_pending >= optimal)
1958                 RETURN(1);
1959
1960         RETURN(0);
1961 }
1962
1963 static int lop_makes_hprpc(struct loi_oap_pages *lop)
1964 {
1965         struct osc_async_page *oap;
1966         ENTRY;
1967
1968         if (cfs_list_empty(&lop->lop_urgent))
1969                 RETURN(0);
1970
1971         oap = cfs_list_entry(lop->lop_urgent.next,
1972                          struct osc_async_page, oap_urgent_item);
1973
1974         if (oap->oap_async_flags & ASYNC_HP) {
1975                 CDEBUG(D_CACHE, "hp request forcing RPC\n");
1976                 RETURN(1);
1977         }
1978
1979         RETURN(0);
1980 }
1981
1982 static void on_list(cfs_list_t *item, cfs_list_t *list,
1983                     int should_be_on)
1984 {
1985         if (cfs_list_empty(item) && should_be_on)
1986                 cfs_list_add_tail(item, list);
1987         else if (!cfs_list_empty(item) && !should_be_on)
1988                 cfs_list_del_init(item);
1989 }
1990
1991 /* maintain the loi's cli list membership invariants so that osc_send_oap_rpc
1992  * can find pages to build into rpcs quickly */
1993 void loi_list_maint(struct client_obd *cli, struct lov_oinfo *loi)
1994 {
1995         if (lop_makes_hprpc(&loi->loi_write_lop) ||
1996             lop_makes_hprpc(&loi->loi_read_lop)) {
1997                 /* HP rpc */
1998                 on_list(&loi->loi_ready_item, &cli->cl_loi_ready_list, 0);
1999                 on_list(&loi->loi_hp_ready_item, &cli->cl_loi_hp_ready_list, 1);
2000         } else {
2001                 on_list(&loi->loi_hp_ready_item, &cli->cl_loi_hp_ready_list, 0);
2002                 on_list(&loi->loi_ready_item, &cli->cl_loi_ready_list,
2003                         lop_makes_rpc(cli, &loi->loi_write_lop, OBD_BRW_WRITE)||
2004                         lop_makes_rpc(cli, &loi->loi_read_lop, OBD_BRW_READ));
2005         }
2006
2007         on_list(&loi->loi_write_item, &cli->cl_loi_write_list,
2008                 loi->loi_write_lop.lop_num_pending);
2009
2010         on_list(&loi->loi_read_item, &cli->cl_loi_read_list,
2011                 loi->loi_read_lop.lop_num_pending);
2012 }
2013
2014 static void lop_update_pending(struct client_obd *cli,
2015                                struct loi_oap_pages *lop, int cmd, int delta)
2016 {
2017         lop->lop_num_pending += delta;
2018         if (cmd & OBD_BRW_WRITE)
2019                 cli->cl_pending_w_pages += delta;
2020         else
2021                 cli->cl_pending_r_pages += delta;
2022 }
2023
2024 /**
2025  * this is called when a sync waiter receives an interruption.  Its job is to
2026  * get the caller woken as soon as possible.  If its page hasn't been put in an
2027  * rpc yet it can dequeue immediately.  Otherwise it has to mark the rpc as
2028  * desiring interruption which will forcefully complete the rpc once the rpc
2029  * has timed out.
2030  */
2031 int osc_oap_interrupted(const struct lu_env *env, struct osc_async_page *oap)
2032 {
2033         struct loi_oap_pages *lop;
2034         struct lov_oinfo *loi;
2035         int rc = -EBUSY;
2036         ENTRY;
2037
2038         LASSERT(!oap->oap_interrupted);
2039         oap->oap_interrupted = 1;
2040
2041         /* ok, it's been put in an rpc. only one oap gets a request reference */
2042         if (oap->oap_request != NULL) {
2043                 ptlrpc_mark_interrupted(oap->oap_request);
2044                 ptlrpcd_wake(oap->oap_request);
2045                 ptlrpc_req_finished(oap->oap_request);
2046                 oap->oap_request = NULL;
2047         }
2048
2049         /*
2050          * page completion may be called only if ->cpo_prep() method was
2051          * executed by osc_io_submit(), that also adds page the to pending list
2052          */
2053         if (!cfs_list_empty(&oap->oap_pending_item)) {
2054                 cfs_list_del_init(&oap->oap_pending_item);
2055                 cfs_list_del_init(&oap->oap_urgent_item);
2056
2057                 loi = oap->oap_loi;
2058                 lop = (oap->oap_cmd & OBD_BRW_WRITE) ?
2059                         &loi->loi_write_lop : &loi->loi_read_lop;
2060                 lop_update_pending(oap->oap_cli, lop, oap->oap_cmd, -1);
2061                 loi_list_maint(oap->oap_cli, oap->oap_loi);
2062                 rc = oap->oap_caller_ops->ap_completion(env,
2063                                           oap->oap_caller_data,
2064                                           oap->oap_cmd, NULL, -EINTR);
2065         }
2066
2067         RETURN(rc);
2068 }
2069
2070 /* this is trying to propogate async writeback errors back up to the
2071  * application.  As an async write fails we record the error code for later if
2072  * the app does an fsync.  As long as errors persist we force future rpcs to be
2073  * sync so that the app can get a sync error and break the cycle of queueing
2074  * pages for which writeback will fail. */
2075 static void osc_process_ar(struct osc_async_rc *ar, __u64 xid,
2076                            int rc)
2077 {
2078         if (rc) {
2079                 if (!ar->ar_rc)
2080                         ar->ar_rc = rc;
2081
2082                 ar->ar_force_sync = 1;
2083                 ar->ar_min_xid = ptlrpc_sample_next_xid();
2084                 return;
2085
2086         }
2087
2088         if (ar->ar_force_sync && (xid >= ar->ar_min_xid))
2089                 ar->ar_force_sync = 0;
2090 }
2091
2092 void osc_oap_to_pending(struct osc_async_page *oap)
2093 {
2094         struct loi_oap_pages *lop;
2095
2096         if (oap->oap_cmd & OBD_BRW_WRITE)
2097                 lop = &oap->oap_loi->loi_write_lop;
2098         else
2099                 lop = &oap->oap_loi->loi_read_lop;
2100
2101         if (oap->oap_async_flags & ASYNC_HP)
2102                 cfs_list_add(&oap->oap_urgent_item, &lop->lop_urgent);
2103         else if (oap->oap_async_flags & ASYNC_URGENT)
2104                 cfs_list_add_tail(&oap->oap_urgent_item, &lop->lop_urgent);
2105         cfs_list_add_tail(&oap->oap_pending_item, &lop->lop_pending);
2106         lop_update_pending(oap->oap_cli, lop, oap->oap_cmd, 1);
2107 }
2108
2109 /* this must be called holding the loi list lock to give coverage to exit_cache,
2110  * async_flag maintenance, and oap_request */
2111 static void osc_ap_completion(const struct lu_env *env,
2112                               struct client_obd *cli, struct obdo *oa,
2113                               struct osc_async_page *oap, int sent, int rc)
2114 {
2115         __u64 xid = 0;
2116
2117         ENTRY;
2118         if (oap->oap_request != NULL) {
2119                 xid = ptlrpc_req_xid(oap->oap_request);
2120                 ptlrpc_req_finished(oap->oap_request);
2121                 oap->oap_request = NULL;
2122         }
2123
2124         cfs_spin_lock(&oap->oap_lock);
2125         oap->oap_async_flags = 0;
2126         cfs_spin_unlock(&oap->oap_lock);
2127         oap->oap_interrupted = 0;
2128
2129         if (oap->oap_cmd & OBD_BRW_WRITE) {
2130                 osc_process_ar(&cli->cl_ar, xid, rc);
2131                 osc_process_ar(&oap->oap_loi->loi_ar, xid, rc);
2132         }
2133
2134         if (rc == 0 && oa != NULL) {
2135                 if (oa->o_valid & OBD_MD_FLBLOCKS)
2136                         oap->oap_loi->loi_lvb.lvb_blocks = oa->o_blocks;
2137                 if (oa->o_valid & OBD_MD_FLMTIME)
2138                         oap->oap_loi->loi_lvb.lvb_mtime = oa->o_mtime;
2139                 if (oa->o_valid & OBD_MD_FLATIME)
2140                         oap->oap_loi->loi_lvb.lvb_atime = oa->o_atime;
2141                 if (oa->o_valid & OBD_MD_FLCTIME)
2142                         oap->oap_loi->loi_lvb.lvb_ctime = oa->o_ctime;
2143         }
2144
2145         rc = oap->oap_caller_ops->ap_completion(env, oap->oap_caller_data,
2146                                                 oap->oap_cmd, oa, rc);
2147
2148         /* ll_ap_completion (from llite) drops PG_locked. so, a new
2149          * I/O on the page could start, but OSC calls it under lock
2150          * and thus we can add oap back to pending safely */
2151         if (rc)
2152                 /* upper layer wants to leave the page on pending queue */
2153                 osc_oap_to_pending(oap);
2154         else
2155                 osc_exit_cache(cli, oap, sent);
2156         EXIT;
2157 }
2158
2159 static int brw_interpret(const struct lu_env *env,
2160                          struct ptlrpc_request *req, void *data, int rc)
2161 {
2162         struct osc_brw_async_args *aa = data;
2163         struct client_obd *cli;
2164         int async;
2165         ENTRY;
2166
2167         rc = osc_brw_fini_request(req, rc);
2168         CDEBUG(D_INODE, "request %p aa %p rc %d\n", req, aa, rc);
2169         if (osc_recoverable_error(rc)) {
2170                 rc = osc_brw_redo_request(req, aa);
2171                 if (rc == 0)
2172                         RETURN(0);
2173         }
2174
2175         if (aa->aa_ocapa) {
2176                 capa_put(aa->aa_ocapa);
2177                 aa->aa_ocapa = NULL;
2178         }
2179
2180         cli = aa->aa_cli;
2181
2182         client_obd_list_lock(&cli->cl_loi_list_lock);
2183
2184         /* We need to decrement before osc_ap_completion->osc_wake_cache_waiters
2185          * is called so we know whether to go to sync BRWs or wait for more
2186          * RPCs to complete */
2187         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE)
2188                 cli->cl_w_in_flight--;
2189         else
2190                 cli->cl_r_in_flight--;
2191
2192         async = cfs_list_empty(&aa->aa_oaps);
2193         if (!async) { /* from osc_send_oap_rpc() */
2194                 struct osc_async_page *oap, *tmp;
2195                 /* the caller may re-use the oap after the completion call so
2196                  * we need to clean it up a little */
2197                 cfs_list_for_each_entry_safe(oap, tmp, &aa->aa_oaps,
2198                                              oap_rpc_item) {
2199                         cfs_list_del_init(&oap->oap_rpc_item);
2200                         osc_ap_completion(env, cli, aa->aa_oa, oap, 1, rc);
2201                 }
2202                 OBDO_FREE(aa->aa_oa);
2203         } else { /* from async_internal() */
2204                 int i;
2205                 for (i = 0; i < aa->aa_page_count; i++)
2206                         osc_release_write_grant(aa->aa_cli, aa->aa_ppga[i], 1);
2207
2208                 if (aa->aa_oa->o_flags & OBD_FL_TEMPORARY)
2209                         OBDO_FREE(aa->aa_oa);
2210         }
2211         osc_wake_cache_waiters(cli);
2212         osc_check_rpcs(env, cli);
2213         client_obd_list_unlock(&cli->cl_loi_list_lock);
2214         if (!async)
2215                 cl_req_completion(env, aa->aa_clerq, rc);
2216         osc_release_ppga(aa->aa_ppga, aa->aa_page_count);
2217         RETURN(rc);
2218 }
2219
2220 static struct ptlrpc_request *osc_build_req(const struct lu_env *env,
2221                                             struct client_obd *cli,
2222                                             cfs_list_t *rpc_list,
2223                                             int page_count, int cmd)
2224 {
2225         struct ptlrpc_request *req;
2226         struct brw_page **pga = NULL;
2227         struct osc_brw_async_args *aa;
2228         struct obdo *oa = NULL;
2229         const struct obd_async_page_ops *ops = NULL;
2230         void *caller_data = NULL;
2231         struct osc_async_page *oap;
2232         struct osc_async_page *tmp;
2233         struct ost_body *body;
2234         struct cl_req *clerq = NULL;
2235         enum cl_req_type crt = (cmd & OBD_BRW_WRITE) ? CRT_WRITE : CRT_READ;
2236         struct ldlm_lock *lock = NULL;
2237         struct cl_req_attr crattr;
2238         int i, rc;
2239
2240         ENTRY;
2241         LASSERT(!cfs_list_empty(rpc_list));
2242
2243         memset(&crattr, 0, sizeof crattr);
2244         OBD_ALLOC(pga, sizeof(*pga) * page_count);
2245         if (pga == NULL)
2246                 GOTO(out, req = ERR_PTR(-ENOMEM));
2247
2248         OBDO_ALLOC(oa);
2249         if (oa == NULL)
2250                 GOTO(out, req = ERR_PTR(-ENOMEM));
2251
2252         i = 0;
2253         cfs_list_for_each_entry(oap, rpc_list, oap_rpc_item) {
2254                 struct cl_page *page = osc_oap2cl_page(oap);
2255                 if (ops == NULL) {
2256                         ops = oap->oap_caller_ops;
2257                         caller_data = oap->oap_caller_data;
2258
2259                         clerq = cl_req_alloc(env, page, crt,
2260                                              1 /* only 1-object rpcs for
2261                                                 * now */);
2262                         if (IS_ERR(clerq))
2263                                 GOTO(out, req = (void *)clerq);
2264                         lock = oap->oap_ldlm_lock;
2265                 }
2266                 pga[i] = &oap->oap_brw_page;
2267                 pga[i]->off = oap->oap_obj_off + oap->oap_page_off;
2268                 CDEBUG(0, "put page %p index %lu oap %p flg %x to pga\n",
2269                        pga[i]->pg, cfs_page_index(oap->oap_page), oap, pga[i]->flag);
2270                 i++;
2271                 cl_req_page_add(env, clerq, page);
2272         }
2273
2274         /* always get the data for the obdo for the rpc */
2275         LASSERT(ops != NULL);
2276         crattr.cra_oa = oa;
2277         crattr.cra_capa = NULL;
2278         cl_req_attr_set(env, clerq, &crattr, ~0ULL);
2279         if (lock) {
2280                 oa->o_handle = lock->l_remote_handle;
2281                 oa->o_valid |= OBD_MD_FLHANDLE;
2282         }
2283
2284         rc = cl_req_prep(env, clerq);
2285         if (rc != 0) {
2286                 CERROR("cl_req_prep failed: %d\n", rc);
2287                 GOTO(out, req = ERR_PTR(rc));
2288         }
2289
2290         sort_brw_pages(pga, page_count);
2291         rc = osc_brw_prep_request(cmd, cli, oa, NULL, page_count,
2292                                   pga, &req, crattr.cra_capa, 1);
2293         if (rc != 0) {
2294                 CERROR("prep_req failed: %d\n", rc);
2295                 GOTO(out, req = ERR_PTR(rc));
2296         }
2297
2298         /* Need to update the timestamps after the request is built in case
2299          * we race with setattr (locally or in queue at OST).  If OST gets
2300          * later setattr before earlier BRW (as determined by the request xid),
2301          * the OST will not use BRW timestamps.  Sadly, there is no obvious
2302          * way to do this in a single call.  bug 10150 */
2303         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
2304         cl_req_attr_set(env, clerq, &crattr,
2305                         OBD_MD_FLMTIME|OBD_MD_FLCTIME|OBD_MD_FLATIME);
2306
2307         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
2308         aa = ptlrpc_req_async_args(req);
2309         CFS_INIT_LIST_HEAD(&aa->aa_oaps);
2310         cfs_list_splice(rpc_list, &aa->aa_oaps);
2311         CFS_INIT_LIST_HEAD(rpc_list);
2312         aa->aa_clerq = clerq;
2313 out:
2314         capa_put(crattr.cra_capa);
2315         if (IS_ERR(req)) {
2316                 if (oa)
2317                         OBDO_FREE(oa);
2318                 if (pga)
2319                         OBD_FREE(pga, sizeof(*pga) * page_count);
2320                 /* this should happen rarely and is pretty bad, it makes the
2321                  * pending list not follow the dirty order */
2322                 client_obd_list_lock(&cli->cl_loi_list_lock);
2323                 cfs_list_for_each_entry_safe(oap, tmp, rpc_list, oap_rpc_item) {
2324                         cfs_list_del_init(&oap->oap_rpc_item);
2325
2326                         /* queued sync pages can be torn down while the pages
2327                          * were between the pending list and the rpc */
2328                         if (oap->oap_interrupted) {
2329                                 CDEBUG(D_INODE, "oap %p interrupted\n", oap);
2330                                 osc_ap_completion(env, cli, NULL, oap, 0,
2331                                                   oap->oap_count);
2332                                 continue;
2333                         }
2334                         osc_ap_completion(env, cli, NULL, oap, 0, PTR_ERR(req));
2335                 }
2336                 if (clerq && !IS_ERR(clerq))
2337                         cl_req_completion(env, clerq, PTR_ERR(req));
2338         }
2339         RETURN(req);
2340 }
2341
2342 /**
2343  * prepare pages for ASYNC io and put pages in send queue.
2344  *
2345  * \param cmd OBD_BRW_* macroses
2346  * \param lop pending pages
2347  *
2348  * \return zero if pages successfully add to send queue.
2349  * \return not zere if error occurring.
2350  */
2351 static int
2352 osc_send_oap_rpc(const struct lu_env *env, struct client_obd *cli,
2353                  struct lov_oinfo *loi,
2354                  int cmd, struct loi_oap_pages *lop)
2355 {
2356         struct ptlrpc_request *req;
2357         obd_count page_count = 0;
2358         struct osc_async_page *oap = NULL, *tmp;
2359         struct osc_brw_async_args *aa;
2360         const struct obd_async_page_ops *ops;
2361         CFS_LIST_HEAD(rpc_list);
2362         CFS_LIST_HEAD(tmp_list);
2363         unsigned int ending_offset;
2364         unsigned  starting_offset = 0;
2365         int srvlock = 0;
2366         struct cl_object *clob = NULL;
2367         ENTRY;
2368
2369         /* ASYNC_HP pages first. At present, when the lock the pages is
2370          * to be canceled, the pages covered by the lock will be sent out
2371          * with ASYNC_HP. We have to send out them as soon as possible. */
2372         cfs_list_for_each_entry_safe(oap, tmp, &lop->lop_urgent, oap_urgent_item) {
2373                 if (oap->oap_async_flags & ASYNC_HP)
2374                         cfs_list_move(&oap->oap_pending_item, &tmp_list);
2375                 else
2376                         cfs_list_move_tail(&oap->oap_pending_item, &tmp_list);
2377                 if (++page_count >= cli->cl_max_pages_per_rpc)
2378                         break;
2379         }
2380
2381         cfs_list_splice(&tmp_list, &lop->lop_pending);
2382         page_count = 0;
2383
2384         /* first we find the pages we're allowed to work with */
2385         cfs_list_for_each_entry_safe(oap, tmp, &lop->lop_pending,
2386                                      oap_pending_item) {
2387                 ops = oap->oap_caller_ops;
2388
2389                 LASSERTF(oap->oap_magic == OAP_MAGIC, "Bad oap magic: oap %p, "
2390                          "magic 0x%x\n", oap, oap->oap_magic);
2391
2392                 if (clob == NULL) {
2393                         /* pin object in memory, so that completion call-backs
2394                          * can be safely called under client_obd_list lock. */
2395                         clob = osc_oap2cl_page(oap)->cp_obj;
2396                         cl_object_get(clob);
2397                 }
2398
2399                 if (page_count != 0 &&
2400                     srvlock != !!(oap->oap_brw_flags & OBD_BRW_SRVLOCK)) {
2401                         CDEBUG(D_PAGE, "SRVLOCK flag mismatch,"
2402                                " oap %p, page %p, srvlock %u\n",
2403                                oap, oap->oap_brw_page.pg, (unsigned)!srvlock);
2404                         break;
2405                 }
2406
2407                 /* If there is a gap at the start of this page, it can't merge
2408                  * with any previous page, so we'll hand the network a
2409                  * "fragmented" page array that it can't transfer in 1 RDMA */
2410                 if (page_count != 0 && oap->oap_page_off != 0)
2411                         break;
2412
2413                 /* in llite being 'ready' equates to the page being locked
2414                  * until completion unlocks it.  commit_write submits a page
2415                  * as not ready because its unlock will happen unconditionally
2416                  * as the call returns.  if we race with commit_write giving
2417                  * us that page we dont' want to create a hole in the page
2418                  * stream, so we stop and leave the rpc to be fired by
2419                  * another dirtier or kupdated interval (the not ready page
2420                  * will still be on the dirty list).  we could call in
2421                  * at the end of ll_file_write to process the queue again. */
2422                 if (!(oap->oap_async_flags & ASYNC_READY)) {
2423                         int rc = ops->ap_make_ready(env, oap->oap_caller_data,
2424                                                     cmd);
2425                         if (rc < 0)
2426                                 CDEBUG(D_INODE, "oap %p page %p returned %d "
2427                                                 "instead of ready\n", oap,
2428                                                 oap->oap_page, rc);
2429                         switch (rc) {
2430                         case -EAGAIN:
2431                                 /* llite is telling us that the page is still
2432                                  * in commit_write and that we should try
2433                                  * and put it in an rpc again later.  we
2434                                  * break out of the loop so we don't create
2435                                  * a hole in the sequence of pages in the rpc
2436                                  * stream.*/
2437                                 oap = NULL;
2438                                 break;
2439                         case -EINTR:
2440                                 /* the io isn't needed.. tell the checks
2441                                  * below to complete the rpc with EINTR */
2442                                 cfs_spin_lock(&oap->oap_lock);
2443                                 oap->oap_async_flags |= ASYNC_COUNT_STABLE;
2444                                 cfs_spin_unlock(&oap->oap_lock);
2445                                 oap->oap_count = -EINTR;
2446                                 break;
2447                         case 0:
2448                                 cfs_spin_lock(&oap->oap_lock);
2449                                 oap->oap_async_flags |= ASYNC_READY;
2450                                 cfs_spin_unlock(&oap->oap_lock);
2451                                 break;
2452                         default:
2453                                 LASSERTF(0, "oap %p page %p returned %d "
2454                                             "from make_ready\n", oap,
2455                                             oap->oap_page, rc);
2456                                 break;
2457                         }
2458                 }
2459                 if (oap == NULL)
2460                         break;
2461                 /*
2462                  * Page submitted for IO has to be locked. Either by
2463                  * ->ap_make_ready() or by higher layers.
2464                  */
2465 #if defined(__KERNEL__) && defined(__linux__)
2466                 {
2467                         struct cl_page *page;
2468
2469                         page = osc_oap2cl_page(oap);
2470
2471                         if (page->cp_type == CPT_CACHEABLE &&
2472                             !(PageLocked(oap->oap_page) &&
2473                               (CheckWriteback(oap->oap_page, cmd)))) {
2474                                 CDEBUG(D_PAGE, "page %p lost wb %lx/%x\n",
2475                                        oap->oap_page,
2476                                        (long)oap->oap_page->flags,
2477                                        oap->oap_async_flags);
2478                                 LBUG();
2479                         }
2480                 }
2481 #endif
2482
2483                 /* take the page out of our book-keeping */
2484                 cfs_list_del_init(&oap->oap_pending_item);
2485                 lop_update_pending(cli, lop, cmd, -1);
2486                 cfs_list_del_init(&oap->oap_urgent_item);
2487
2488                 if (page_count == 0)
2489                         starting_offset = (oap->oap_obj_off+oap->oap_page_off) &
2490                                           (PTLRPC_MAX_BRW_SIZE - 1);
2491
2492                 /* ask the caller for the size of the io as the rpc leaves. */
2493                 if (!(oap->oap_async_flags & ASYNC_COUNT_STABLE)) {
2494                         oap->oap_count =
2495                                 ops->ap_refresh_count(env, oap->oap_caller_data,
2496                                                       cmd);
2497                         LASSERT(oap->oap_page_off + oap->oap_count <= CFS_PAGE_SIZE);
2498                 }
2499                 if (oap->oap_count <= 0) {
2500                         CDEBUG(D_CACHE, "oap %p count %d, completing\n", oap,
2501                                oap->oap_count);
2502                         osc_ap_completion(env, cli, NULL,
2503                                           oap, 0, oap->oap_count);
2504                         continue;
2505                 }
2506
2507                 /* now put the page back in our accounting */
2508                 cfs_list_add_tail(&oap->oap_rpc_item, &rpc_list);
2509                 if (page_count == 0)
2510                         srvlock = !!(oap->oap_brw_flags & OBD_BRW_SRVLOCK);
2511                 if (++page_count >= cli->cl_max_pages_per_rpc)
2512                         break;
2513
2514                 /* End on a PTLRPC_MAX_BRW_SIZE boundary.  We want full-sized
2515                  * RPCs aligned on PTLRPC_MAX_BRW_SIZE boundaries to help reads
2516                  * have the same alignment as the initial writes that allocated
2517                  * extents on the server. */
2518                 ending_offset = (oap->oap_obj_off + oap->oap_page_off +
2519                                  oap->oap_count) & (PTLRPC_MAX_BRW_SIZE - 1);
2520                 if (ending_offset == 0)
2521                         break;
2522
2523                 /* If there is a gap at the end of this page, it can't merge
2524                  * with any subsequent pages, so we'll hand the network a
2525                  * "fragmented" page array that it can't transfer in 1 RDMA */
2526                 if (oap->oap_page_off + oap->oap_count < CFS_PAGE_SIZE)
2527                         break;
2528         }
2529
2530         osc_wake_cache_waiters(cli);
2531
2532         loi_list_maint(cli, loi);
2533
2534         client_obd_list_unlock(&cli->cl_loi_list_lock);
2535
2536         if (clob != NULL)
2537                 cl_object_put(env, clob);
2538
2539         if (page_count == 0) {
2540                 client_obd_list_lock(&cli->cl_loi_list_lock);
2541                 RETURN(0);
2542         }
2543
2544         req = osc_build_req(env, cli, &rpc_list, page_count, cmd);
2545         if (IS_ERR(req)) {
2546                 LASSERT(cfs_list_empty(&rpc_list));
2547                 loi_list_maint(cli, loi);
2548                 RETURN(PTR_ERR(req));
2549         }
2550
2551         aa = ptlrpc_req_async_args(req);
2552
2553         if (cmd == OBD_BRW_READ) {
2554                 lprocfs_oh_tally_log2(&cli->cl_read_page_hist, page_count);
2555                 lprocfs_oh_tally(&cli->cl_read_rpc_hist, cli->cl_r_in_flight);
2556                 lprocfs_oh_tally_log2(&cli->cl_read_offset_hist,
2557                                       (starting_offset >> CFS_PAGE_SHIFT) + 1);
2558         } else {
2559                 lprocfs_oh_tally_log2(&cli->cl_write_page_hist, page_count);
2560                 lprocfs_oh_tally(&cli->cl_write_rpc_hist,
2561                                  cli->cl_w_in_flight);
2562                 lprocfs_oh_tally_log2(&cli->cl_write_offset_hist,
2563                                       (starting_offset >> CFS_PAGE_SHIFT) + 1);
2564         }
2565         ptlrpc_lprocfs_brw(req, aa->aa_requested_nob);
2566
2567         client_obd_list_lock(&cli->cl_loi_list_lock);
2568
2569         if (cmd == OBD_BRW_READ)
2570                 cli->cl_r_in_flight++;
2571         else
2572                 cli->cl_w_in_flight++;
2573
2574         /* queued sync pages can be torn down while the pages
2575          * were between the pending list and the rpc */
2576         tmp = NULL;
2577         cfs_list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
2578                 /* only one oap gets a request reference */
2579                 if (tmp == NULL)
2580                         tmp = oap;
2581                 if (oap->oap_interrupted && !req->rq_intr) {
2582                         CDEBUG(D_INODE, "oap %p in req %p interrupted\n",
2583                                oap, req);
2584                         ptlrpc_mark_interrupted(req);
2585                 }
2586         }
2587         if (tmp != NULL)
2588                 tmp->oap_request = ptlrpc_request_addref(req);
2589
2590         DEBUG_REQ(D_INODE, req, "%d pages, aa %p. now %dr/%dw in flight",
2591                   page_count, aa, cli->cl_r_in_flight, cli->cl_w_in_flight);
2592
2593         req->rq_interpret_reply = brw_interpret;
2594         ptlrpcd_add_req(req, PSCOPE_BRW);
2595         RETURN(1);
2596 }
2597
2598 #define LOI_DEBUG(LOI, STR, args...)                                     \
2599         CDEBUG(D_INODE, "loi ready %d wr %d:%d rd %d:%d " STR,           \
2600                !cfs_list_empty(&(LOI)->loi_ready_item) ||                \
2601                !cfs_list_empty(&(LOI)->loi_hp_ready_item),               \
2602                (LOI)->loi_write_lop.lop_num_pending,                     \
2603                !cfs_list_empty(&(LOI)->loi_write_lop.lop_urgent),        \
2604                (LOI)->loi_read_lop.lop_num_pending,                      \
2605                !cfs_list_empty(&(LOI)->loi_read_lop.lop_urgent),         \
2606                args)                                                     \
2607
2608 /* This is called by osc_check_rpcs() to find which objects have pages that
2609  * we could be sending.  These lists are maintained by lop_makes_rpc(). */
2610 struct lov_oinfo *osc_next_loi(struct client_obd *cli)
2611 {
2612         ENTRY;
2613
2614         /* First return objects that have blocked locks so that they
2615          * will be flushed quickly and other clients can get the lock,
2616          * then objects which have pages ready to be stuffed into RPCs */
2617         if (!cfs_list_empty(&cli->cl_loi_hp_ready_list))
2618                 RETURN(cfs_list_entry(cli->cl_loi_hp_ready_list.next,
2619                                       struct lov_oinfo, loi_hp_ready_item));
2620         if (!cfs_list_empty(&cli->cl_loi_ready_list))
2621                 RETURN(cfs_list_entry(cli->cl_loi_ready_list.next,
2622                                       struct lov_oinfo, loi_ready_item));
2623
2624         /* then if we have cache waiters, return all objects with queued
2625          * writes.  This is especially important when many small files
2626          * have filled up the cache and not been fired into rpcs because
2627          * they don't pass the nr_pending/object threshhold */
2628         if (!cfs_list_empty(&cli->cl_cache_waiters) &&
2629             !cfs_list_empty(&cli->cl_loi_write_list))
2630                 RETURN(cfs_list_entry(cli->cl_loi_write_list.next,
2631                                       struct lov_oinfo, loi_write_item));
2632
2633         /* then return all queued objects when we have an invalid import
2634          * so that they get flushed */
2635         if (cli->cl_import == NULL || cli->cl_import->imp_invalid) {
2636                 if (!cfs_list_empty(&cli->cl_loi_write_list))
2637                         RETURN(cfs_list_entry(cli->cl_loi_write_list.next,
2638                                               struct lov_oinfo,
2639                                               loi_write_item));
2640                 if (!cfs_list_empty(&cli->cl_loi_read_list))
2641                         RETURN(cfs_list_entry(cli->cl_loi_read_list.next,
2642                                               struct lov_oinfo, loi_read_item));
2643         }
2644         RETURN(NULL);
2645 }
2646
2647 static int osc_max_rpc_in_flight(struct client_obd *cli, struct lov_oinfo *loi)
2648 {
2649         struct osc_async_page *oap;
2650         int hprpc = 0;
2651
2652         if (!cfs_list_empty(&loi->loi_write_lop.lop_urgent)) {
2653                 oap = cfs_list_entry(loi->loi_write_lop.lop_urgent.next,
2654                                      struct osc_async_page, oap_urgent_item);
2655                 hprpc = !!(oap->oap_async_flags & ASYNC_HP);
2656         }
2657
2658         if (!hprpc && !cfs_list_empty(&loi->loi_read_lop.lop_urgent)) {
2659                 oap = cfs_list_entry(loi->loi_read_lop.lop_urgent.next,
2660                                      struct osc_async_page, oap_urgent_item);
2661                 hprpc = !!(oap->oap_async_flags & ASYNC_HP);
2662         }
2663
2664         return rpcs_in_flight(cli) >= cli->cl_max_rpcs_in_flight + hprpc;
2665 }
2666
2667 /* called with the loi list lock held */
2668 void osc_check_rpcs(const struct lu_env *env, struct client_obd *cli)
2669 {
2670         struct lov_oinfo *loi;
2671         int rc = 0, race_counter = 0;
2672         ENTRY;
2673
2674         while ((loi = osc_next_loi(cli)) != NULL) {
2675                 LOI_DEBUG(loi, "%lu in flight\n", rpcs_in_flight(cli));
2676
2677                 if (osc_max_rpc_in_flight(cli, loi))
2678                         break;
2679
2680                 /* attempt some read/write balancing by alternating between
2681                  * reads and writes in an object.  The makes_rpc checks here
2682                  * would be redundant if we were getting read/write work items
2683                  * instead of objects.  we don't want send_oap_rpc to drain a
2684                  * partial read pending queue when we're given this object to
2685                  * do io on writes while there are cache waiters */
2686                 if (lop_makes_rpc(cli, &loi->loi_write_lop, OBD_BRW_WRITE)) {
2687                         rc = osc_send_oap_rpc(env, cli, loi, OBD_BRW_WRITE,
2688                                               &loi->loi_write_lop);
2689                         if (rc < 0) {
2690                                 CERROR("Write request failed with %d\n", rc);
2691
2692                                 /* osc_send_oap_rpc failed, mostly because of
2693                                  * memory pressure.
2694                                  *
2695                                  * It can't break here, because if:
2696                                  *  - a page was submitted by osc_io_submit, so
2697                                  *    page locked;
2698                                  *  - no request in flight
2699                                  *  - no subsequent request
2700                                  * The system will be in live-lock state,
2701                                  * because there is no chance to call
2702                                  * osc_io_unplug() and osc_check_rpcs() any
2703                                  * more. pdflush can't help in this case,
2704                                  * because it might be blocked at grabbing
2705                                  * the page lock as we mentioned.
2706                                  *
2707                                  * Anyway, continue to drain pages. */
2708                                 /* break; */
2709                         }
2710
2711                         if (rc > 0)
2712                                 race_counter = 0;
2713                         else
2714                                 race_counter++;
2715                 }
2716                 if (lop_makes_rpc(cli, &loi->loi_read_lop, OBD_BRW_READ)) {
2717                         rc = osc_send_oap_rpc(env, cli, loi, OBD_BRW_READ,
2718                                               &loi->loi_read_lop);
2719                         if (rc < 0)
2720                                 CERROR("Read request failed with %d\n", rc);
2721
2722                         if (rc > 0)
2723                                 race_counter = 0;
2724                         else
2725                                 race_counter++;
2726                 }
2727
2728                 /* attempt some inter-object balancing by issueing rpcs
2729                  * for each object in turn */
2730                 if (!cfs_list_empty(&loi->loi_hp_ready_item))
2731                         cfs_list_del_init(&loi->loi_hp_ready_item);
2732                 if (!cfs_list_empty(&loi->loi_ready_item))
2733                         cfs_list_del_init(&loi->loi_ready_item);
2734                 if (!cfs_list_empty(&loi->loi_write_item))
2735                         cfs_list_del_init(&loi->loi_write_item);
2736                 if (!cfs_list_empty(&loi->loi_read_item))
2737                         cfs_list_del_init(&loi->loi_read_item);
2738
2739                 loi_list_maint(cli, loi);
2740
2741                 /* send_oap_rpc fails with 0 when make_ready tells it to
2742                  * back off.  llite's make_ready does this when it tries
2743                  * to lock a page queued for write that is already locked.
2744                  * we want to try sending rpcs from many objects, but we
2745                  * don't want to spin failing with 0.  */
2746                 if (race_counter == 10)
2747                         break;
2748         }
2749         EXIT;
2750 }
2751
2752 /* we're trying to queue a page in the osc so we're subject to the
2753  * 'cl_dirty_max' limit on the number of pages that can be queued in the osc.
2754  * If the osc's queued pages are already at that limit, then we want to sleep
2755  * until there is space in the osc's queue for us.  We also may be waiting for
2756  * write credits from the OST if there are RPCs in flight that may return some
2757  * before we fall back to sync writes.
2758  *
2759  * We need this know our allocation was granted in the presence of signals */
2760 static int ocw_granted(struct client_obd *cli, struct osc_cache_waiter *ocw)
2761 {
2762         int rc;
2763         ENTRY;
2764         client_obd_list_lock(&cli->cl_loi_list_lock);
2765         rc = cfs_list_empty(&ocw->ocw_entry) || rpcs_in_flight(cli) == 0;
2766         client_obd_list_unlock(&cli->cl_loi_list_lock);
2767         RETURN(rc);
2768 };
2769
2770 /**
2771  * Non-blocking version of osc_enter_cache() that consumes grant only when it
2772  * is available.
2773  */
2774 int osc_enter_cache_try(const struct lu_env *env,
2775                         struct client_obd *cli, struct lov_oinfo *loi,
2776                         struct osc_async_page *oap, int transient)
2777 {
2778         int has_grant;
2779
2780         has_grant = cli->cl_avail_grant >= CFS_PAGE_SIZE;
2781         if (has_grant) {
2782                 osc_consume_write_grant(cli, &oap->oap_brw_page);
2783                 if (transient) {
2784                         cli->cl_dirty_transit += CFS_PAGE_SIZE;
2785                         cfs_atomic_inc(&obd_dirty_transit_pages);
2786                         oap->oap_brw_flags |= OBD_BRW_NOCACHE;
2787                 }
2788         }
2789         return has_grant;
2790 }
2791
2792 /* Caller must hold loi_list_lock - we drop/regain it if we need to wait for
2793  * grant or cache space. */
2794 static int osc_enter_cache(const struct lu_env *env,
2795                            struct client_obd *cli, struct lov_oinfo *loi,
2796                            struct osc_async_page *oap)
2797 {
2798         struct osc_cache_waiter ocw;
2799         struct l_wait_info lwi = { 0 };
2800
2801         ENTRY;
2802
2803         CDEBUG(D_CACHE, "dirty: %ld/%d dirty_max: %ld/%d dropped: %lu "
2804                "grant: %lu\n", cli->cl_dirty, cfs_atomic_read(&obd_dirty_pages),
2805                cli->cl_dirty_max, obd_max_dirty_pages,
2806                cli->cl_lost_grant, cli->cl_avail_grant);
2807
2808         /* force the caller to try sync io.  this can jump the list
2809          * of queued writes and create a discontiguous rpc stream */
2810         if (cli->cl_dirty_max < CFS_PAGE_SIZE || cli->cl_ar.ar_force_sync ||
2811             loi->loi_ar.ar_force_sync)
2812                 RETURN(-EDQUOT);
2813
2814         /* Hopefully normal case - cache space and write credits available */
2815         if (cli->cl_dirty + CFS_PAGE_SIZE <= cli->cl_dirty_max &&
2816             cfs_atomic_read(&obd_dirty_pages) + 1 <= obd_max_dirty_pages &&
2817             osc_enter_cache_try(env, cli, loi, oap, 0))
2818                 RETURN(0);
2819
2820         /* It is safe to block as a cache waiter as long as there is grant
2821          * space available or the hope of additional grant being returned
2822          * when an in flight write completes.  Using the write back cache
2823          * if possible is preferable to sending the data synchronously
2824          * because write pages can then be merged in to large requests.
2825          * The addition of this cache waiter will causing pending write
2826          * pages to be sent immediately. */
2827         if (cli->cl_w_in_flight || cli->cl_avail_grant >= CFS_PAGE_SIZE) {
2828                 cfs_list_add_tail(&ocw.ocw_entry, &cli->cl_cache_waiters);
2829                 cfs_waitq_init(&ocw.ocw_waitq);
2830                 ocw.ocw_oap = oap;
2831                 ocw.ocw_rc = 0;
2832
2833                 loi_list_maint(cli, loi);
2834                 osc_check_rpcs(env, cli);
2835                 client_obd_list_unlock(&cli->cl_loi_list_lock);
2836
2837                 CDEBUG(D_CACHE, "sleeping for cache space\n");
2838                 l_wait_event(ocw.ocw_waitq, ocw_granted(cli, &ocw), &lwi);
2839
2840                 client_obd_list_lock(&cli->cl_loi_list_lock);
2841                 if (!cfs_list_empty(&ocw.ocw_entry)) {
2842                         cfs_list_del(&ocw.ocw_entry);
2843                         RETURN(-EINTR);
2844                 }
2845                 RETURN(ocw.ocw_rc);
2846         }
2847
2848         RETURN(-EDQUOT);
2849 }
2850
2851
2852 int osc_prep_async_page(struct obd_export *exp, struct lov_stripe_md *lsm,
2853                         struct lov_oinfo *loi, cfs_page_t *page,
2854                         obd_off offset, const struct obd_async_page_ops *ops,
2855                         void *data, void **res, int nocache,
2856                         struct lustre_handle *lockh)
2857 {
2858         struct osc_async_page *oap;
2859
2860         ENTRY;
2861
2862         if (!page)
2863                 return cfs_size_round(sizeof(*oap));
2864
2865         oap = *res;
2866         oap->oap_magic = OAP_MAGIC;
2867         oap->oap_cli = &exp->exp_obd->u.cli;
2868         oap->oap_loi = loi;
2869
2870         oap->oap_caller_ops = ops;
2871         oap->oap_caller_data = data;
2872
2873         oap->oap_page = page;
2874         oap->oap_obj_off = offset;
2875         if (!client_is_remote(exp) &&
2876             cfs_capable(CFS_CAP_SYS_RESOURCE))
2877                 oap->oap_brw_flags = OBD_BRW_NOQUOTA;
2878
2879         LASSERT(!(offset & ~CFS_PAGE_MASK));
2880
2881         CFS_INIT_LIST_HEAD(&oap->oap_pending_item);
2882         CFS_INIT_LIST_HEAD(&oap->oap_urgent_item);
2883         CFS_INIT_LIST_HEAD(&oap->oap_rpc_item);
2884         CFS_INIT_LIST_HEAD(&oap->oap_page_list);
2885
2886         cfs_spin_lock_init(&oap->oap_lock);
2887         CDEBUG(D_CACHE, "oap %p page %p obj off "LPU64"\n", oap, page, offset);
2888         RETURN(0);
2889 }
2890
2891 struct osc_async_page *oap_from_cookie(void *cookie)
2892 {
2893         struct osc_async_page *oap = cookie;
2894         if (oap->oap_magic != OAP_MAGIC)
2895                 return ERR_PTR(-EINVAL);
2896         return oap;
2897 };
2898
2899 int osc_queue_async_io(const struct lu_env *env,
2900                        struct obd_export *exp, struct lov_stripe_md *lsm,
2901                        struct lov_oinfo *loi, void *cookie,
2902                        int cmd, obd_off off, int count,
2903                        obd_flag brw_flags, enum async_flags async_flags)
2904 {
2905         struct client_obd *cli = &exp->exp_obd->u.cli;
2906         struct osc_async_page *oap;
2907         int rc = 0;
2908         ENTRY;
2909
2910         oap = oap_from_cookie(cookie);
2911         if (IS_ERR(oap))
2912                 RETURN(PTR_ERR(oap));
2913
2914         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
2915                 RETURN(-EIO);
2916
2917         if (!cfs_list_empty(&oap->oap_pending_item) ||
2918             !cfs_list_empty(&oap->oap_urgent_item) ||
2919             !cfs_list_empty(&oap->oap_rpc_item))
2920                 RETURN(-EBUSY);
2921
2922         /* check if the file's owner/group is over quota */
2923         if ((cmd & OBD_BRW_WRITE) && !(cmd & OBD_BRW_NOQUOTA)) {
2924                 struct cl_object *obj;
2925                 struct cl_attr    attr; /* XXX put attr into thread info */
2926                 unsigned int qid[MAXQUOTAS];
2927
2928                 obj = cl_object_top(osc_oap2cl_page(oap)->cp_obj);
2929
2930                 cl_object_attr_lock(obj);
2931                 rc = cl_object_attr_get(env, obj, &attr);
2932                 cl_object_attr_unlock(obj);
2933
2934                 qid[USRQUOTA] = attr.cat_uid;
2935                 qid[GRPQUOTA] = attr.cat_gid;
2936                 if (rc == 0 &&
2937                     lquota_chkdq(quota_interface, cli, qid) == NO_QUOTA)
2938                         rc = -EDQUOT;
2939                 if (rc)
2940                         RETURN(rc);
2941         }
2942
2943         if (loi == NULL)
2944                 loi = lsm->lsm_oinfo[0];
2945
2946         client_obd_list_lock(&cli->cl_loi_list_lock);
2947
2948         LASSERT(off + count <= CFS_PAGE_SIZE);
2949         oap->oap_cmd = cmd;
2950         oap->oap_page_off = off;
2951         oap->oap_count = count;
2952         oap->oap_brw_flags = brw_flags;
2953         /* Give a hint to OST that requests are coming from kswapd - bug19529 */
2954         if (libcfs_memory_pressure_get())
2955                 oap->oap_brw_flags |= OBD_BRW_MEMALLOC;
2956         cfs_spin_lock(&oap->oap_lock);
2957         oap->oap_async_flags = async_flags;
2958         cfs_spin_unlock(&oap->oap_lock);
2959
2960         if (cmd & OBD_BRW_WRITE) {
2961                 rc = osc_enter_cache(env, cli, loi, oap);
2962                 if (rc) {
2963                         client_obd_list_unlock(&cli->cl_loi_list_lock);
2964                         RETURN(rc);
2965                 }
2966         }
2967
2968         osc_oap_to_pending(oap);
2969         loi_list_maint(cli, loi);
2970
2971         LOI_DEBUG(loi, "oap %p page %p added for cmd %d\n", oap, oap->oap_page,
2972                   cmd);
2973
2974         osc_check_rpcs(env, cli);
2975         client_obd_list_unlock(&cli->cl_loi_list_lock);
2976
2977         RETURN(0);
2978 }
2979
2980 /* aka (~was & now & flag), but this is more clear :) */
2981 #define SETTING(was, now, flag) (!(was & flag) && (now & flag))
2982
2983 int osc_set_async_flags_base(struct client_obd *cli,
2984                              struct lov_oinfo *loi, struct osc_async_page *oap,
2985                              obd_flag async_flags)
2986 {
2987         struct loi_oap_pages *lop;
2988         int flags = 0;
2989         ENTRY;
2990
2991         LASSERT(!cfs_list_empty(&oap->oap_pending_item));
2992
2993         if (oap->oap_cmd & OBD_BRW_WRITE) {
2994                 lop = &loi->loi_write_lop;
2995         } else {
2996                 lop = &loi->loi_read_lop;
2997         }
2998
2999         if ((oap->oap_async_flags & async_flags) == async_flags)
3000                 RETURN(0);
3001
3002         if (SETTING(oap->oap_async_flags, async_flags, ASYNC_READY))
3003                 flags |= ASYNC_READY;
3004
3005         if (SETTING(oap->oap_async_flags, async_flags, ASYNC_URGENT) &&
3006             cfs_list_empty(&oap->oap_rpc_item)) {
3007                 if (oap->oap_async_flags & ASYNC_HP)
3008                         cfs_list_add(&oap->oap_urgent_item, &lop->lop_urgent);
3009                 else
3010                         cfs_list_add_tail(&oap->oap_urgent_item,
3011                                           &lop->lop_urgent);
3012                 flags |= ASYNC_URGENT;
3013                 loi_list_maint(cli, loi);
3014         }
3015         cfs_spin_lock(&oap->oap_lock);
3016         oap->oap_async_flags |= flags;
3017         cfs_spin_unlock(&oap->oap_lock);
3018
3019         LOI_DEBUG(loi, "oap %p page %p has flags %x\n", oap, oap->oap_page,
3020                         oap->oap_async_flags);
3021         RETURN(0);
3022 }
3023
3024 int osc_teardown_async_page(struct obd_export *exp,
3025                             struct lov_stripe_md *lsm,
3026                             struct lov_oinfo *loi, void *cookie)
3027 {
3028         struct client_obd *cli = &exp->exp_obd->u.cli;
3029         struct loi_oap_pages *lop;
3030         struct osc_async_page *oap;
3031         int rc = 0;
3032         ENTRY;
3033
3034         oap = oap_from_cookie(cookie);
3035         if (IS_ERR(oap))
3036                 RETURN(PTR_ERR(oap));
3037
3038         if (loi == NULL)
3039                 loi = lsm->lsm_oinfo[0];
3040
3041         if (oap->oap_cmd & OBD_BRW_WRITE) {
3042                 lop = &loi->loi_write_lop;
3043         } else {
3044                 lop = &loi->loi_read_lop;
3045         }
3046
3047         client_obd_list_lock(&cli->cl_loi_list_lock);
3048
3049         if (!cfs_list_empty(&oap->oap_rpc_item))
3050                 GOTO(out, rc = -EBUSY);
3051
3052         osc_exit_cache(cli, oap, 0);
3053         osc_wake_cache_waiters(cli);
3054
3055         if (!cfs_list_empty(&oap->oap_urgent_item)) {
3056                 cfs_list_del_init(&oap->oap_urgent_item);
3057                 cfs_spin_lock(&oap->oap_lock);
3058                 oap->oap_async_flags &= ~(ASYNC_URGENT | ASYNC_HP);
3059                 cfs_spin_unlock(&oap->oap_lock);
3060         }
3061         if (!cfs_list_empty(&oap->oap_pending_item)) {
3062                 cfs_list_del_init(&oap->oap_pending_item);
3063                 lop_update_pending(cli, lop, oap->oap_cmd, -1);
3064         }
3065         loi_list_maint(cli, loi);
3066         LOI_DEBUG(loi, "oap %p page %p torn down\n", oap, oap->oap_page);
3067 out:
3068         client_obd_list_unlock(&cli->cl_loi_list_lock);
3069         RETURN(rc);
3070 }
3071
3072 static void osc_set_lock_data_with_check(struct ldlm_lock *lock,
3073                                          struct ldlm_enqueue_info *einfo,
3074                                          int flags)
3075 {
3076         void *data = einfo->ei_cbdata;
3077
3078         LASSERT(lock != NULL);
3079         LASSERT(lock->l_blocking_ast == einfo->ei_cb_bl);
3080         LASSERT(lock->l_resource->lr_type == einfo->ei_type);
3081         LASSERT(lock->l_completion_ast == einfo->ei_cb_cp);
3082         LASSERT(lock->l_glimpse_ast == einfo->ei_cb_gl);
3083
3084         lock_res_and_lock(lock);
3085         cfs_spin_lock(&osc_ast_guard);
3086         LASSERT(lock->l_ast_data == NULL || lock->l_ast_data == data);
3087         lock->l_ast_data = data;
3088         cfs_spin_unlock(&osc_ast_guard);
3089         unlock_res_and_lock(lock);
3090 }
3091
3092 static void osc_set_data_with_check(struct lustre_handle *lockh,
3093                                     struct ldlm_enqueue_info *einfo,
3094                                     int flags)
3095 {
3096         struct ldlm_lock *lock = ldlm_handle2lock(lockh);
3097
3098         if (lock != NULL) {
3099                 osc_set_lock_data_with_check(lock, einfo, flags);
3100                 LDLM_LOCK_PUT(lock);
3101         } else
3102                 CERROR("lockh %p, data %p - client evicted?\n",
3103                        lockh, einfo->ei_cbdata);
3104 }
3105
3106 static int osc_change_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
3107                              ldlm_iterator_t replace, void *data)
3108 {
3109         struct ldlm_res_id res_id;
3110         struct obd_device *obd = class_exp2obd(exp);
3111
3112         osc_build_res_name(lsm->lsm_object_id, lsm->lsm_object_seq, &res_id);
3113         ldlm_resource_iterate(obd->obd_namespace, &res_id, replace, data);
3114         return 0;
3115 }
3116
3117 /* find any ldlm lock of the inode in osc
3118  * return 0    not find
3119  *        1    find one
3120  *      < 0    error */
3121 static int osc_find_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
3122                            ldlm_iterator_t replace, void *data)
3123 {
3124         struct ldlm_res_id res_id;
3125         struct obd_device *obd = class_exp2obd(exp);
3126         int rc = 0;
3127
3128         osc_build_res_name(lsm->lsm_object_id, lsm->lsm_object_seq, &res_id);
3129         rc = ldlm_resource_iterate(obd->obd_namespace, &res_id, replace, data);
3130         if (rc == LDLM_ITER_STOP)
3131                 return(1);
3132         if (rc == LDLM_ITER_CONTINUE)
3133                 return(0);
3134         return(rc);
3135 }
3136
3137 static int osc_enqueue_fini(struct ptlrpc_request *req, struct ost_lvb *lvb,
3138                             obd_enqueue_update_f upcall, void *cookie,
3139                             int *flags, int rc)
3140 {
3141         int intent = *flags & LDLM_FL_HAS_INTENT;
3142         ENTRY;
3143
3144         if (intent) {
3145                 /* The request was created before ldlm_cli_enqueue call. */
3146                 if (rc == ELDLM_LOCK_ABORTED) {
3147                         struct ldlm_reply *rep;
3148                         rep = req_capsule_server_get(&req->rq_pill,
3149                                                      &RMF_DLM_REP);
3150
3151                         LASSERT(rep != NULL);
3152                         if (rep->lock_policy_res1)
3153                                 rc = rep->lock_policy_res1;
3154                 }
3155         }
3156
3157         if ((intent && rc == ELDLM_LOCK_ABORTED) || !rc) {
3158                 *flags |= LDLM_FL_LVB_READY;
3159                 CDEBUG(D_INODE,"got kms "LPU64" blocks "LPU64" mtime "LPU64"\n",
3160                        lvb->lvb_size, lvb->lvb_blocks, lvb->lvb_mtime);
3161         }
3162
3163         /* Call the update callback. */
3164         rc = (*upcall)(cookie, rc);
3165         RETURN(rc);
3166 }
3167
3168 static int osc_enqueue_interpret(const struct lu_env *env,
3169                                  struct ptlrpc_request *req,
3170                                  struct osc_enqueue_args *aa, int rc)
3171 {
3172         struct ldlm_lock *lock;
3173         struct lustre_handle handle;
3174         __u32 mode;
3175
3176         /* Make a local copy of a lock handle and a mode, because aa->oa_*
3177          * might be freed anytime after lock upcall has been called. */
3178         lustre_handle_copy(&handle, aa->oa_lockh);
3179         mode = aa->oa_ei->ei_mode;
3180
3181         /* ldlm_cli_enqueue is holding a reference on the lock, so it must
3182          * be valid. */
3183         lock = ldlm_handle2lock(&handle);
3184
3185         /* Take an additional reference so that a blocking AST that
3186          * ldlm_cli_enqueue_fini() might post for a failed lock, is guaranteed
3187          * to arrive after an upcall has been executed by
3188          * osc_enqueue_fini(). */
3189         ldlm_lock_addref(&handle, mode);
3190
3191         /* Complete obtaining the lock procedure. */
3192         rc = ldlm_cli_enqueue_fini(aa->oa_exp, req, aa->oa_ei->ei_type, 1,
3193                                    mode, aa->oa_flags, aa->oa_lvb,
3194                                    sizeof(*aa->oa_lvb), &handle, rc);
3195         /* Complete osc stuff. */
3196         rc = osc_enqueue_fini(req, aa->oa_lvb,
3197                               aa->oa_upcall, aa->oa_cookie, aa->oa_flags, rc);
3198
3199         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_CANCEL_RACE, 10);
3200
3201         /* Release the lock for async request. */
3202         if (lustre_handle_is_used(&handle) && rc == ELDLM_OK)
3203                 /*
3204                  * Releases a reference taken by ldlm_cli_enqueue(), if it is
3205                  * not already released by
3206                  * ldlm_cli_enqueue_fini()->failed_lock_cleanup()
3207                  */
3208                 ldlm_lock_decref(&handle, mode);
3209
3210         LASSERTF(lock != NULL, "lockh %p, req %p, aa %p - client evicted?\n",
3211                  aa->oa_lockh, req, aa);
3212         ldlm_lock_decref(&handle, mode);
3213         LDLM_LOCK_PUT(lock);
3214         return rc;
3215 }
3216
3217 void osc_update_enqueue(struct lustre_handle *lov_lockhp,
3218                         struct lov_oinfo *loi, int flags,
3219                         struct ost_lvb *lvb, __u32 mode, int rc)
3220 {
3221         if (rc == ELDLM_OK) {
3222                 struct ldlm_lock *lock = ldlm_handle2lock(lov_lockhp);
3223                 __u64 tmp;
3224
3225                 LASSERT(lock != NULL);
3226                 loi->loi_lvb = *lvb;
3227                 tmp = loi->loi_lvb.lvb_size;
3228                 /* Extend KMS up to the end of this lock and no further
3229                  * A lock on [x,y] means a KMS of up to y + 1 bytes! */
3230                 if (tmp > lock->l_policy_data.l_extent.end)
3231                         tmp = lock->l_policy_data.l_extent.end + 1;
3232                 if (tmp >= loi->loi_kms) {
3233                         LDLM_DEBUG(lock, "lock acquired, setting rss="LPU64
3234                                    ", kms="LPU64, loi->loi_lvb.lvb_size, tmp);
3235                         loi_kms_set(loi, tmp);
3236                 } else {
3237                         LDLM_DEBUG(lock, "lock acquired, setting rss="
3238                                    LPU64"; leaving kms="LPU64", end="LPU64,
3239                                    loi->loi_lvb.lvb_size, loi->loi_kms,
3240                                    lock->l_policy_data.l_extent.end);
3241                 }
3242                 ldlm_lock_allow_match(lock);
3243                 LDLM_LOCK_PUT(lock);
3244         } else if (rc == ELDLM_LOCK_ABORTED && (flags & LDLM_FL_HAS_INTENT)) {
3245                 loi->loi_lvb = *lvb;
3246                 CDEBUG(D_INODE, "glimpsed, setting rss="LPU64"; leaving"
3247                        " kms="LPU64"\n", loi->loi_lvb.lvb_size, loi->loi_kms);
3248                 rc = ELDLM_OK;
3249         }
3250 }
3251 EXPORT_SYMBOL(osc_update_enqueue);
3252
3253 struct ptlrpc_request_set *PTLRPCD_SET = (void *)1;
3254
3255 /* When enqueuing asynchronously, locks are not ordered, we can obtain a lock
3256  * from the 2nd OSC before a lock from the 1st one. This does not deadlock with
3257  * other synchronous requests, however keeping some locks and trying to obtain
3258  * others may take a considerable amount of time in a case of ost failure; and
3259  * when other sync requests do not get released lock from a client, the client
3260  * is excluded from the cluster -- such scenarious make the life difficult, so
3261  * release locks just after they are obtained. */
3262 int osc_enqueue_base(struct obd_export *exp, struct ldlm_res_id *res_id,
3263                      int *flags, ldlm_policy_data_t *policy,
3264                      struct ost_lvb *lvb, int kms_valid,
3265                      obd_enqueue_update_f upcall, void *cookie,
3266                      struct ldlm_enqueue_info *einfo,
3267                      struct lustre_handle *lockh,
3268                      struct ptlrpc_request_set *rqset, int async)
3269 {
3270         struct obd_device *obd = exp->exp_obd;
3271         struct ptlrpc_request *req = NULL;
3272         int intent = *flags & LDLM_FL_HAS_INTENT;
3273         ldlm_mode_t mode;
3274         int rc;
3275         ENTRY;
3276
3277         /* Filesystem lock extents are extended to page boundaries so that
3278          * dealing with the page cache is a little smoother.  */
3279         policy->l_extent.start -= policy->l_extent.start & ~CFS_PAGE_MASK;
3280         policy->l_extent.end |= ~CFS_PAGE_MASK;
3281
3282         /*
3283          * kms is not valid when either object is completely fresh (so that no
3284          * locks are cached), or object was evicted. In the latter case cached
3285          * lock cannot be used, because it would prime inode state with
3286          * potentially stale LVB.
3287          */
3288         if (!kms_valid)
3289                 goto no_match;
3290
3291         /* Next, search for already existing extent locks that will cover us */
3292         /* If we're trying to read, we also search for an existing PW lock.  The
3293          * VFS and page cache already protect us locally, so lots of readers/
3294          * writers can share a single PW lock.
3295          *
3296          * There are problems with conversion deadlocks, so instead of
3297          * converting a read lock to a write lock, we'll just enqueue a new
3298          * one.
3299          *
3300          * At some point we should cancel the read lock instead of making them
3301          * send us a blocking callback, but there are problems with canceling
3302          * locks out from other users right now, too. */
3303         mode = einfo->ei_mode;
3304         if (einfo->ei_mode == LCK_PR)
3305                 mode |= LCK_PW;
3306         mode = ldlm_lock_match(obd->obd_namespace,
3307                                *flags | LDLM_FL_LVB_READY, res_id,
3308                                einfo->ei_type, policy, mode, lockh, 0);
3309         if (mode) {
3310                 struct ldlm_lock *matched = ldlm_handle2lock(lockh);
3311
3312                 if (matched->l_ast_data == NULL ||
3313                     matched->l_ast_data == einfo->ei_cbdata) {
3314                         /* addref the lock only if not async requests and PW
3315                          * lock is matched whereas we asked for PR. */
3316                         if (!rqset && einfo->ei_mode != mode)
3317                                 ldlm_lock_addref(lockh, LCK_PR);
3318                         osc_set_lock_data_with_check(matched, einfo, *flags);
3319                         if (intent) {
3320                                 /* I would like to be able to ASSERT here that
3321                                  * rss <= kms, but I can't, for reasons which
3322                                  * are explained in lov_enqueue() */
3323                         }
3324
3325                         /* We already have a lock, and it's referenced */
3326                         (*upcall)(cookie, ELDLM_OK);
3327
3328                         /* For async requests, decref the lock. */
3329                         if (einfo->ei_mode != mode)
3330                                 ldlm_lock_decref(lockh, LCK_PW);
3331                         else if (rqset)
3332                                 ldlm_lock_decref(lockh, einfo->ei_mode);
3333                         LDLM_LOCK_PUT(matched);
3334                         RETURN(ELDLM_OK);
3335                 } else
3336                         ldlm_lock_decref(lockh, mode);
3337                 LDLM_LOCK_PUT(matched);
3338         }
3339
3340  no_match:
3341         if (intent) {
3342                 CFS_LIST_HEAD(cancels);
3343                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
3344                                            &RQF_LDLM_ENQUEUE_LVB);
3345                 if (req == NULL)
3346                         RETURN(-ENOMEM);
3347
3348                 rc = ldlm_prep_enqueue_req(exp, req, &cancels, 0);
3349                 if (rc)
3350                         RETURN(rc);
3351
3352                 req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
3353                                      sizeof *lvb);
3354                 ptlrpc_request_set_replen(req);
3355         }
3356
3357         /* users of osc_enqueue() can pass this flag for ldlm_lock_match() */
3358         *flags &= ~LDLM_FL_BLOCK_GRANTED;
3359
3360         rc = ldlm_cli_enqueue(exp, &req, einfo, res_id, policy, flags, lvb,
3361                               sizeof(*lvb), lockh, async);
3362         if (rqset) {
3363                 if (!rc) {
3364                         struct osc_enqueue_args *aa;
3365                         CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
3366                         aa = ptlrpc_req_async_args(req);
3367                         aa->oa_ei = einfo;
3368                         aa->oa_exp = exp;
3369                         aa->oa_flags  = flags;
3370                         aa->oa_upcall = upcall;
3371                         aa->oa_cookie = cookie;
3372                         aa->oa_lvb    = lvb;
3373                         aa->oa_lockh  = lockh;
3374
3375                         req->rq_interpret_reply =
3376                                 (ptlrpc_interpterer_t)osc_enqueue_interpret;
3377                         if (rqset == PTLRPCD_SET)
3378                                 ptlrpcd_add_req(req, PSCOPE_OTHER);
3379                         else
3380                                 ptlrpc_set_add_req(rqset, req);
3381                 } else if (intent) {
3382                         ptlrpc_req_finished(req);
3383                 }
3384                 RETURN(rc);
3385         }
3386
3387         rc = osc_enqueue_fini(req, lvb, upcall, cookie, flags, rc);
3388         if (intent)
3389                 ptlrpc_req_finished(req);
3390
3391         RETURN(rc);
3392 }
3393
3394 static int osc_enqueue(struct obd_export *exp, struct obd_info *oinfo,
3395                        struct ldlm_enqueue_info *einfo,
3396                        struct ptlrpc_request_set *rqset)
3397 {
3398         struct ldlm_res_id res_id;
3399         int rc;
3400         ENTRY;
3401
3402         osc_build_res_name(oinfo->oi_md->lsm_object_id,
3403                            oinfo->oi_md->lsm_object_seq, &res_id);
3404
3405         rc = osc_enqueue_base(exp, &res_id, &oinfo->oi_flags, &oinfo->oi_policy,
3406                               &oinfo->oi_md->lsm_oinfo[0]->loi_lvb,
3407                               oinfo->oi_md->lsm_oinfo[0]->loi_kms_valid,
3408                               oinfo->oi_cb_up, oinfo, einfo, oinfo->oi_lockh,
3409                               rqset, rqset != NULL);
3410         RETURN(rc);
3411 }
3412
3413 int osc_match_base(struct obd_export *exp, struct ldlm_res_id *res_id,
3414                    __u32 type, ldlm_policy_data_t *policy, __u32 mode,
3415                    int *flags, void *data, struct lustre_handle *lockh,
3416                    int unref)
3417 {
3418         struct obd_device *obd = exp->exp_obd;
3419         int lflags = *flags;
3420         ldlm_mode_t rc;
3421         ENTRY;
3422
3423         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_MATCH))
3424                 RETURN(-EIO);
3425
3426         /* Filesystem lock extents are extended to page boundaries so that
3427          * dealing with the page cache is a little smoother */
3428         policy->l_extent.start -= policy->l_extent.start & ~CFS_PAGE_MASK;
3429         policy->l_extent.end |= ~CFS_PAGE_MASK;
3430
3431         /* Next, search for already existing extent locks that will cover us */
3432         /* If we're trying to read, we also search for an existing PW lock.  The
3433          * VFS and page cache already protect us locally, so lots of readers/
3434          * writers can share a single PW lock. */
3435         rc = mode;
3436         if (mode == LCK_PR)
3437                 rc |= LCK_PW;
3438         rc = ldlm_lock_match(obd->obd_namespace, lflags,
3439                              res_id, type, policy, rc, lockh, unref);
3440         if (rc) {
3441                 if (data != NULL)
3442                         osc_set_data_with_check(lockh, data, lflags);
3443                 if (!(lflags & LDLM_FL_TEST_LOCK) && mode != rc) {
3444                         ldlm_lock_addref(lockh, LCK_PR);
3445                         ldlm_lock_decref(lockh, LCK_PW);
3446                 }
3447                 RETURN(rc);
3448         }
3449         RETURN(rc);
3450 }
3451
3452 int osc_cancel_base(struct lustre_handle *lockh, __u32 mode)
3453 {
3454         ENTRY;
3455
3456         if (unlikely(mode == LCK_GROUP))
3457                 ldlm_lock_decref_and_cancel(lockh, mode);
3458         else
3459                 ldlm_lock_decref(lockh, mode);
3460
3461         RETURN(0);
3462 }
3463
3464 static int osc_cancel(struct obd_export *exp, struct lov_stripe_md *md,
3465                       __u32 mode, struct lustre_handle *lockh)
3466 {
3467         ENTRY;
3468         RETURN(osc_cancel_base(lockh, mode));
3469 }
3470
3471 static int osc_cancel_unused(struct obd_export *exp,
3472                              struct lov_stripe_md *lsm,
3473                              ldlm_cancel_flags_t flags,
3474                              void *opaque)
3475 {
3476         struct obd_device *obd = class_exp2obd(exp);
3477         struct ldlm_res_id res_id, *resp = NULL;
3478
3479         if (lsm != NULL) {
3480                 resp = osc_build_res_name(lsm->lsm_object_id,
3481                                           lsm->lsm_object_seq, &res_id);
3482         }
3483
3484         return ldlm_cli_cancel_unused(obd->obd_namespace, resp, flags, opaque);
3485 }
3486
3487 static int osc_statfs_interpret(const struct lu_env *env,
3488                                 struct ptlrpc_request *req,
3489                                 struct osc_async_args *aa, int rc)
3490 {
3491         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
3492         struct obd_statfs *msfs;
3493         __u64 used;
3494         ENTRY;
3495
3496         if (rc == -EBADR)
3497                 /* The request has in fact never been sent
3498                  * due to issues at a higher level (LOV).
3499                  * Exit immediately since the caller is
3500                  * aware of the problem and takes care
3501                  * of the clean up */
3502                  RETURN(rc);
3503
3504         if ((rc == -ENOTCONN || rc == -EAGAIN) &&
3505             (aa->aa_oi->oi_flags & OBD_STATFS_NODELAY))
3506                 GOTO(out, rc = 0);
3507
3508         if (rc != 0)
3509                 GOTO(out, rc);
3510
3511         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
3512         if (msfs == NULL) {
3513                 GOTO(out, rc = -EPROTO);
3514         }
3515
3516         /* Reinitialize the RDONLY and DEGRADED flags at the client
3517          * on each statfs, so they don't stay set permanently. */
3518         cfs_spin_lock(&cli->cl_oscc.oscc_lock);
3519
3520         if (unlikely(msfs->os_state & OS_STATE_DEGRADED))
3521                 cli->cl_oscc.oscc_flags |= OSCC_FLAG_DEGRADED;
3522         else if (unlikely(cli->cl_oscc.oscc_flags & OSCC_FLAG_DEGRADED))
3523                 cli->cl_oscc.oscc_flags &= ~OSCC_FLAG_DEGRADED;
3524
3525         if (unlikely(msfs->os_state & OS_STATE_READONLY))
3526                 cli->cl_oscc.oscc_flags |= OSCC_FLAG_RDONLY;
3527         else if (unlikely(cli->cl_oscc.oscc_flags & OSCC_FLAG_RDONLY))
3528                 cli->cl_oscc.oscc_flags &= ~OSCC_FLAG_RDONLY;
3529
3530         /* Add a bit of hysteresis so this flag isn't continually flapping,
3531          * and ensure that new files don't get extremely fragmented due to
3532          * only a small amount of available space in the filesystem.
3533          * We want to set the NOSPC flag when there is less than ~0.1% free
3534          * and clear it when there is at least ~0.2% free space, so:
3535          *                   avail < ~0.1% max          max = avail + used
3536          *            1025 * avail < avail + used       used = blocks - free
3537          *            1024 * avail < used
3538          *            1024 * avail < blocks - free
3539          *                   avail < ((blocks - free) >> 10)
3540          *
3541          * On very large disk, say 16TB 0.1% will be 16 GB. We don't want to
3542          * lose that amount of space so in those cases we report no space left
3543          * if their is less than 1 GB left.                             */
3544         used = min_t(__u64,(msfs->os_blocks - msfs->os_bfree) >> 10, 1 << 30);
3545         if (unlikely(((cli->cl_oscc.oscc_flags & OSCC_FLAG_NOSPC) == 0) &&
3546                      ((msfs->os_ffree < 32) || (msfs->os_bavail < used))))
3547                 cli->cl_oscc.oscc_flags |= OSCC_FLAG_NOSPC;
3548         else if (unlikely(((cli->cl_oscc.oscc_flags & OSCC_FLAG_NOSPC) != 0) &&
3549                 (msfs->os_ffree > 64) && (msfs->os_bavail > (used << 1))))
3550                         cli->cl_oscc.oscc_flags &= ~OSCC_FLAG_NOSPC;
3551
3552         cfs_spin_unlock(&cli->cl_oscc.oscc_lock);
3553
3554         *aa->aa_oi->oi_osfs = *msfs;
3555 out:
3556         rc = aa->aa_oi->oi_cb_up(aa->aa_oi, rc);
3557         RETURN(rc);
3558 }
3559
3560 static int osc_statfs_async(struct obd_device *obd, struct obd_info *oinfo,
3561                             __u64 max_age, struct ptlrpc_request_set *rqset)
3562 {
3563         struct ptlrpc_request *req;
3564         struct osc_async_args *aa;
3565         int                    rc;
3566         ENTRY;
3567
3568         /* We could possibly pass max_age in the request (as an absolute
3569          * timestamp or a "seconds.usec ago") so the target can avoid doing
3570          * extra calls into the filesystem if that isn't necessary (e.g.
3571          * during mount that would help a bit).  Having relative timestamps
3572          * is not so great if request processing is slow, while absolute
3573          * timestamps are not ideal because they need time synchronization. */
3574         req = ptlrpc_request_alloc(obd->u.cli.cl_import, &RQF_OST_STATFS);
3575         if (req == NULL)
3576                 RETURN(-ENOMEM);
3577
3578         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
3579         if (rc) {
3580                 ptlrpc_request_free(req);
3581                 RETURN(rc);
3582         }
3583         ptlrpc_request_set_replen(req);
3584         req->rq_request_portal = OST_CREATE_PORTAL;
3585         ptlrpc_at_set_req_timeout(req);
3586
3587         if (oinfo->oi_flags & OBD_STATFS_NODELAY) {
3588                 /* procfs requests not want stat in wait for avoid deadlock */
3589                 req->rq_no_resend = 1;
3590                 req->rq_no_delay = 1;
3591         }
3592
3593         req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_statfs_interpret;
3594         CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
3595         aa = ptlrpc_req_async_args(req);
3596         aa->aa_oi = oinfo;
3597
3598         ptlrpc_set_add_req(rqset, req);
3599         RETURN(0);
3600 }
3601
3602 static int osc_statfs(struct obd_device *obd, struct obd_statfs *osfs,
3603                       __u64 max_age, __u32 flags)
3604 {
3605         struct obd_statfs     *msfs;
3606         struct ptlrpc_request *req;
3607         struct obd_import     *imp = NULL;
3608         int rc;
3609         ENTRY;
3610
3611         /*Since the request might also come from lprocfs, so we need
3612          *sync this with client_disconnect_export Bug15684*/
3613         cfs_down_read(&obd->u.cli.cl_sem);
3614         if (obd->u.cli.cl_import)
3615                 imp = class_import_get(obd->u.cli.cl_import);
3616         cfs_up_read(&obd->u.cli.cl_sem);
3617         if (!imp)
3618                 RETURN(-ENODEV);
3619
3620         /* We could possibly pass max_age in the request (as an absolute
3621          * timestamp or a "seconds.usec ago") so the target can avoid doing
3622          * extra calls into the filesystem if that isn't necessary (e.g.
3623          * during mount that would help a bit).  Having relative timestamps
3624          * is not so great if request processing is slow, while absolute
3625          * timestamps are not ideal because they need time synchronization. */
3626         req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
3627
3628         class_import_put(imp);
3629
3630         if (req == NULL)
3631                 RETURN(-ENOMEM);
3632
3633         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
3634         if (rc) {
3635                 ptlrpc_request_free(req);
3636                 RETURN(rc);
3637         }
3638         ptlrpc_request_set_replen(req);
3639         req->rq_request_portal = OST_CREATE_PORTAL;
3640         ptlrpc_at_set_req_timeout(req);
3641
3642         if (flags & OBD_STATFS_NODELAY) {
3643                 /* procfs requests not want stat in wait for avoid deadlock */
3644                 req->rq_no_resend = 1;
3645                 req->rq_no_delay = 1;
3646         }
3647
3648         rc = ptlrpc_queue_wait(req);
3649         if (rc)
3650                 GOTO(out, rc);
3651
3652         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
3653         if (msfs == NULL) {
3654                 GOTO(out, rc = -EPROTO);
3655         }
3656
3657         *osfs = *msfs;
3658
3659         EXIT;
3660  out:
3661         ptlrpc_req_finished(req);
3662         return rc;
3663 }
3664
3665 /* Retrieve object striping information.
3666  *
3667  * @lmmu is a pointer to an in-core struct with lmm_ost_count indicating
3668  * the maximum number of OST indices which will fit in the user buffer.
3669  * lmm_magic must be LOV_MAGIC (we only use 1 slot here).
3670  */
3671 static int osc_getstripe(struct lov_stripe_md *lsm, struct lov_user_md *lump)
3672 {
3673         /* we use lov_user_md_v3 because it is larger than lov_user_md_v1 */
3674         struct lov_user_md_v3 lum, *lumk;
3675         struct lov_user_ost_data_v1 *lmm_objects;
3676         int rc = 0, lum_size;
3677         ENTRY;
3678
3679         if (!lsm)
3680                 RETURN(-ENODATA);
3681
3682         /* we only need the header part from user space to get lmm_magic and
3683          * lmm_stripe_count, (the header part is common to v1 and v3) */
3684         lum_size = sizeof(struct lov_user_md_v1);
3685         if (cfs_copy_from_user(&lum, lump, lum_size))
3686                 RETURN(-EFAULT);
3687
3688         if ((lum.lmm_magic != LOV_USER_MAGIC_V1) &&
3689             (lum.lmm_magic != LOV_USER_MAGIC_V3))
3690                 RETURN(-EINVAL);
3691
3692         /* lov_user_md_vX and lov_mds_md_vX must have the same size */
3693         LASSERT(sizeof(struct lov_user_md_v1) == sizeof(struct lov_mds_md_v1));
3694         LASSERT(sizeof(struct lov_user_md_v3) == sizeof(struct lov_mds_md_v3));
3695         LASSERT(sizeof(lum.lmm_objects[0]) == sizeof(lumk->lmm_objects[0]));
3696
3697         /* we can use lov_mds_md_size() to compute lum_size
3698          * because lov_user_md_vX and lov_mds_md_vX have the same size */
3699         if (lum.lmm_stripe_count > 0) {
3700                 lum_size = lov_mds_md_size(lum.lmm_stripe_count, lum.lmm_magic);
3701                 OBD_ALLOC(lumk, lum_size);
3702                 if (!lumk)
3703                         RETURN(-ENOMEM);
3704
3705                 if (lum.lmm_magic == LOV_USER_MAGIC_V1)
3706                         lmm_objects = &(((struct lov_user_md_v1 *)lumk)->lmm_objects[0]);
3707                 else
3708                         lmm_objects = &(lumk->lmm_objects[0]);
3709                 lmm_objects->l_object_id = lsm->lsm_object_id;
3710         } else {
3711                 lum_size = lov_mds_md_size(0, lum.lmm_magic);
3712                 lumk = &lum;
3713         }
3714
3715         lumk->lmm_object_id = lsm->lsm_object_id;
3716         lumk->lmm_object_seq = lsm->lsm_object_seq;
3717         lumk->lmm_stripe_count = 1;
3718
3719         if (cfs_copy_to_user(lump, lumk, lum_size))
3720                 rc = -EFAULT;
3721
3722         if (lumk != &lum)
3723                 OBD_FREE(lumk, lum_size);
3724
3725         RETURN(rc);
3726 }
3727
3728
3729 static int osc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
3730                          void *karg, void *uarg)
3731 {
3732         struct obd_device *obd = exp->exp_obd;
3733         struct obd_ioctl_data *data = karg;
3734         int err = 0;
3735         ENTRY;
3736
3737         if (!cfs_try_module_get(THIS_MODULE)) {
3738                 CERROR("Can't get module. Is it alive?");
3739                 return -EINVAL;
3740         }
3741         switch (cmd) {
3742         case OBD_IOC_LOV_GET_CONFIG: {
3743                 char *buf;
3744                 struct lov_desc *desc;
3745                 struct obd_uuid uuid;
3746
3747                 buf = NULL;
3748                 len = 0;
3749                 if (obd_ioctl_getdata(&buf, &len, (void *)uarg))
3750                         GOTO(out, err = -EINVAL);
3751
3752                 data = (struct obd_ioctl_data *)buf;
3753
3754                 if (sizeof(*desc) > data->ioc_inllen1) {
3755                         obd_ioctl_freedata(buf, len);
3756                         GOTO(out, err = -EINVAL);
3757                 }
3758
3759                 if (data->ioc_inllen2 < sizeof(uuid)) {
3760                         obd_ioctl_freedata(buf, len);
3761                         GOTO(out, err = -EINVAL);
3762                 }
3763
3764                 desc = (struct lov_desc *)data->ioc_inlbuf1;
3765                 desc->ld_tgt_count = 1;
3766                 desc->ld_active_tgt_count = 1;
3767                 desc->ld_default_stripe_count = 1;
3768                 desc->ld_default_stripe_size = 0;
3769                 desc->ld_default_stripe_offset = 0;
3770                 desc->ld_pattern = 0;
3771                 memcpy(&desc->ld_uuid, &obd->obd_uuid, sizeof(uuid));
3772
3773                 memcpy(data->ioc_inlbuf2, &obd->obd_uuid, sizeof(uuid));
3774
3775                 err = cfs_copy_to_user((void *)uarg, buf, len);
3776                 if (err)
3777                         err = -EFAULT;
3778                 obd_ioctl_freedata(buf, len);
3779                 GOTO(out, err);
3780         }
3781         case LL_IOC_LOV_SETSTRIPE:
3782                 err = obd_alloc_memmd(exp, karg);
3783                 if (err > 0)
3784                         err = 0;
3785                 GOTO(out, err);
3786         case LL_IOC_LOV_GETSTRIPE:
3787                 err = osc_getstripe(karg, uarg);
3788                 GOTO(out, err);
3789         case OBD_IOC_CLIENT_RECOVER:
3790                 err = ptlrpc_recover_import(obd->u.cli.cl_import,
3791                                             data->ioc_inlbuf1);
3792                 if (err > 0)
3793                         err = 0;
3794                 GOTO(out, err);
3795         case IOC_OSC_SET_ACTIVE:
3796                 err = ptlrpc_set_import_active(obd->u.cli.cl_import,
3797                                                data->ioc_offset);
3798                 GOTO(out, err);
3799         case OBD_IOC_POLL_QUOTACHECK:
3800                 err = lquota_poll_check(quota_interface, exp,
3801                                         (struct if_quotacheck *)karg);
3802                 GOTO(out, err);
3803         case OBD_IOC_PING_TARGET:
3804                 err = ptlrpc_obd_ping(obd);
3805                 GOTO(out, err);
3806         default:
3807                 CDEBUG(D_INODE, "unrecognised ioctl %#x by %s\n",
3808                        cmd, cfs_curproc_comm());
3809                 GOTO(out, err = -ENOTTY);
3810         }
3811 out:
3812         cfs_module_put(THIS_MODULE);
3813         return err;
3814 }
3815
3816 static int osc_get_info(struct obd_export *exp, obd_count keylen,
3817                         void *key, __u32 *vallen, void *val,
3818                         struct lov_stripe_md *lsm)
3819 {
3820         ENTRY;
3821         if (!vallen || !val)
3822                 RETURN(-EFAULT);
3823
3824         if (KEY_IS(KEY_LOCK_TO_STRIPE)) {
3825                 __u32 *stripe = val;
3826                 *vallen = sizeof(*stripe);
3827                 *stripe = 0;
3828                 RETURN(0);
3829         } else if (KEY_IS(KEY_LAST_ID)) {
3830                 struct ptlrpc_request *req;
3831                 obd_id                *reply;
3832                 char                  *tmp;
3833                 int                    rc;
3834
3835                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
3836                                            &RQF_OST_GET_INFO_LAST_ID);
3837                 if (req == NULL)
3838                         RETURN(-ENOMEM);
3839
3840                 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
3841                                      RCL_CLIENT, keylen);
3842                 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GET_INFO);
3843                 if (rc) {
3844                         ptlrpc_request_free(req);
3845                         RETURN(rc);
3846                 }
3847
3848                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
3849                 memcpy(tmp, key, keylen);
3850
3851                 req->rq_no_delay = req->rq_no_resend = 1;
3852                 ptlrpc_request_set_replen(req);
3853                 rc = ptlrpc_queue_wait(req);
3854                 if (rc)
3855                         GOTO(out, rc);
3856
3857                 reply = req_capsule_server_get(&req->rq_pill, &RMF_OBD_ID);
3858                 if (reply == NULL)
3859                         GOTO(out, rc = -EPROTO);
3860
3861                 *((obd_id *)val) = *reply;
3862         out:
3863                 ptlrpc_req_finished(req);
3864                 RETURN(rc);
3865         } else if (KEY_IS(KEY_FIEMAP)) {
3866                 struct ptlrpc_request *req;
3867                 struct ll_user_fiemap *reply;
3868                 char *tmp;
3869                 int rc;
3870
3871                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
3872                                            &RQF_OST_GET_INFO_FIEMAP);
3873                 if (req == NULL)
3874                         RETURN(-ENOMEM);
3875
3876                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_KEY,
3877                                      RCL_CLIENT, keylen);
3878                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_VAL,
3879                                      RCL_CLIENT, *vallen);
3880                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_VAL,
3881                                      RCL_SERVER, *vallen);
3882
3883                 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GET_INFO);
3884                 if (rc) {
3885                         ptlrpc_request_free(req);
3886                         RETURN(rc);
3887                 }
3888
3889                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_FIEMAP_KEY);
3890                 memcpy(tmp, key, keylen);
3891                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_FIEMAP_VAL);
3892                 memcpy(tmp, val, *vallen);
3893
3894                 ptlrpc_request_set_replen(req);
3895                 rc = ptlrpc_queue_wait(req);
3896                 if (rc)
3897                         GOTO(out1, rc);
3898
3899                 reply = req_capsule_server_get(&req->rq_pill, &RMF_FIEMAP_VAL);
3900                 if (reply == NULL)
3901                         GOTO(out1, rc = -EPROTO);
3902
3903                 memcpy(val, reply, *vallen);
3904         out1:
3905                 ptlrpc_req_finished(req);
3906
3907                 RETURN(rc);
3908         }
3909
3910         RETURN(-EINVAL);
3911 }
3912
3913 static int osc_setinfo_mds_connect_import(struct obd_import *imp)
3914 {
3915         struct llog_ctxt *ctxt;
3916         int rc = 0;
3917         ENTRY;
3918
3919         ctxt = llog_get_context(imp->imp_obd, LLOG_MDS_OST_ORIG_CTXT);
3920         if (ctxt) {
3921                 rc = llog_initiator_connect(ctxt);
3922                 llog_ctxt_put(ctxt);
3923         } else {
3924                 /* XXX return an error? skip setting below flags? */
3925         }
3926
3927         cfs_spin_lock(&imp->imp_lock);
3928         imp->imp_server_timeout = 1;
3929         imp->imp_pingable = 1;
3930         cfs_spin_unlock(&imp->imp_lock);
3931         CDEBUG(D_RPCTRACE, "pinging OST %s\n", obd2cli_tgt(imp->imp_obd));
3932
3933         RETURN(rc);
3934 }
3935
3936 static int osc_setinfo_mds_conn_interpret(const struct lu_env *env,
3937                                           struct ptlrpc_request *req,
3938                                           void *aa, int rc)
3939 {
3940         ENTRY;
3941         if (rc != 0)
3942                 RETURN(rc);
3943
3944         RETURN(osc_setinfo_mds_connect_import(req->rq_import));
3945 }
3946
3947 static int osc_set_info_async(struct obd_export *exp, obd_count keylen,
3948                               void *key, obd_count vallen, void *val,
3949                               struct ptlrpc_request_set *set)
3950 {
3951         struct ptlrpc_request *req;
3952         struct obd_device     *obd = exp->exp_obd;
3953         struct obd_import     *imp = class_exp2cliimp(exp);
3954         char                  *tmp;
3955         int                    rc;
3956         ENTRY;
3957
3958         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_SHUTDOWN, 10);
3959
3960         if (KEY_IS(KEY_NEXT_ID)) {
3961                 obd_id new_val;
3962                 struct osc_creator *oscc = &obd->u.cli.cl_oscc;
3963
3964                 if (vallen != sizeof(obd_id))
3965                         RETURN(-ERANGE);
3966                 if (val == NULL)
3967                         RETURN(-EINVAL);
3968
3969                 if (vallen != sizeof(obd_id))
3970                         RETURN(-EINVAL);
3971
3972                 /* avoid race between allocate new object and set next id
3973                  * from ll_sync thread */
3974                 cfs_spin_lock(&oscc->oscc_lock);
3975                 new_val = *((obd_id*)val) + 1;
3976                 if (new_val > oscc->oscc_next_id)
3977                         oscc->oscc_next_id = new_val;
3978                 cfs_spin_unlock(&oscc->oscc_lock);
3979                 CDEBUG(D_HA, "%s: set oscc_next_id = "LPU64"\n",
3980                        exp->exp_obd->obd_name,
3981                        obd->u.cli.cl_oscc.oscc_next_id);
3982
3983                 RETURN(0);
3984         }
3985
3986         if (KEY_IS(KEY_CHECKSUM)) {
3987                 if (vallen != sizeof(int))
3988                         RETURN(-EINVAL);
3989                 exp->exp_obd->u.cli.cl_checksum = (*(int *)val) ? 1 : 0;
3990                 RETURN(0);
3991         }
3992
3993         if (KEY_IS(KEY_SPTLRPC_CONF)) {
3994                 sptlrpc_conf_client_adapt(obd);
3995                 RETURN(0);
3996         }
3997
3998         if (KEY_IS(KEY_FLUSH_CTX)) {
3999                 sptlrpc_import_flush_my_ctx(imp);
4000                 RETURN(0);
4001         }
4002
4003         if (!set && !KEY_IS(KEY_GRANT_SHRINK))
4004                 RETURN(-EINVAL);
4005
4006         /* We pass all other commands directly to OST. Since nobody calls osc
4007            methods directly and everybody is supposed to go through LOV, we
4008            assume lov checked invalid values for us.
4009            The only recognised values so far are evict_by_nid and mds_conn.
4010            Even if something bad goes through, we'd get a -EINVAL from OST
4011            anyway. */
4012
4013         if (KEY_IS(KEY_GRANT_SHRINK))
4014                 req = ptlrpc_request_alloc(imp, &RQF_OST_SET_GRANT_INFO);
4015         else
4016                 req = ptlrpc_request_alloc(imp, &RQF_OBD_SET_INFO);
4017
4018         if (req == NULL)
4019                 RETURN(-ENOMEM);
4020
4021         req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
4022                              RCL_CLIENT, keylen);
4023         req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_VAL,
4024                              RCL_CLIENT, vallen);
4025         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SET_INFO);
4026         if (rc) {
4027                 ptlrpc_request_free(req);
4028                 RETURN(rc);
4029         }
4030
4031         tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
4032         memcpy(tmp, key, keylen);
4033         tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_VAL);
4034         memcpy(tmp, val, vallen);
4035
4036         if (KEY_IS(KEY_MDS_CONN)) {
4037                 struct osc_creator *oscc = &obd->u.cli.cl_oscc;
4038
4039                 oscc->oscc_oa.o_seq = (*(__u32 *)val);
4040                 oscc->oscc_oa.o_valid |= OBD_MD_FLGROUP;
4041                 LASSERT_SEQ_IS_MDT(oscc->oscc_oa.o_seq);
4042                 req->rq_no_delay = req->rq_no_resend = 1;
4043                 req->rq_interpret_reply = osc_setinfo_mds_conn_interpret;
4044         } else if (KEY_IS(KEY_GRANT_SHRINK)) {
4045                 struct osc_grant_args *aa;
4046                 struct obdo *oa;
4047
4048                 CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
4049                 aa = ptlrpc_req_async_args(req);
4050                 OBD_ALLOC_PTR(oa);
4051                 if (!oa) {
4052                         ptlrpc_req_finished(req);
4053                         RETURN(-ENOMEM);
4054                 }
4055                 *oa = ((struct ost_body *)val)->oa;
4056                 aa->aa_oa = oa;
4057                 req->rq_interpret_reply = osc_shrink_grant_interpret;
4058         }
4059
4060         ptlrpc_request_set_replen(req);
4061         if (!KEY_IS(KEY_GRANT_SHRINK)) {
4062                 LASSERT(set != NULL);
4063                 ptlrpc_set_add_req(set, req);
4064                 ptlrpc_check_set(NULL, set);
4065         } else
4066                 ptlrpcd_add_req(req, PSCOPE_OTHER);
4067
4068         RETURN(0);
4069 }
4070
4071
4072 static struct llog_operations osc_size_repl_logops = {
4073         lop_cancel: llog_obd_repl_cancel
4074 };
4075
4076 static struct llog_operations osc_mds_ost_orig_logops;
4077
4078 static int __osc_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
4079                            struct obd_device *tgt, struct llog_catid *catid)
4080 {
4081         int rc;
4082         ENTRY;
4083
4084         rc = llog_setup(obd, &obd->obd_olg, LLOG_MDS_OST_ORIG_CTXT, tgt, 1,
4085                         &catid->lci_logid, &osc_mds_ost_orig_logops);
4086         if (rc) {
4087                 CERROR("failed LLOG_MDS_OST_ORIG_CTXT\n");
4088                 GOTO(out, rc);
4089         }
4090
4091         rc = llog_setup(obd, &obd->obd_olg, LLOG_SIZE_REPL_CTXT, tgt, 1,
4092                         NULL, &osc_size_repl_logops);
4093         if (rc) {
4094                 struct llog_ctxt *ctxt =
4095                         llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
4096                 if (ctxt)
4097                         llog_cleanup(ctxt);
4098                 CERROR("failed LLOG_SIZE_REPL_CTXT\n");
4099         }
4100         GOTO(out, rc);
4101 out:
4102         if (rc) {
4103                 CERROR("osc '%s' tgt '%s' catid %p rc=%d\n",
4104                        obd->obd_name, tgt->obd_name, catid, rc);
4105                 CERROR("logid "LPX64":0x%x\n",
4106                        catid->lci_logid.lgl_oid, catid->lci_logid.lgl_ogen);
4107         }
4108         return rc;
4109 }
4110
4111 static int osc_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
4112                          struct obd_device *disk_obd, int *index)
4113 {
4114         struct llog_catid catid;
4115         static char name[32] = CATLIST;
4116         int rc;
4117         ENTRY;
4118
4119         LASSERT(olg == &obd->obd_olg);
4120
4121         cfs_mutex_down(&olg->olg_cat_processing);
4122         rc = llog_get_cat_list(disk_obd, name, *index, 1, &catid);
4123         if (rc) {
4124                 CERROR("rc: %d\n", rc);
4125                 GOTO(out, rc);
4126         }
4127
4128         CDEBUG(D_INFO, "%s: Init llog for %d - catid "LPX64"/"LPX64":%x\n",
4129                obd->obd_name, *index, catid.lci_logid.lgl_oid,
4130                catid.lci_logid.lgl_oseq, catid.lci_logid.lgl_ogen);
4131
4132         rc = __osc_llog_init(obd, olg, disk_obd, &catid);
4133         if (rc) {
4134                 CERROR("rc: %d\n", rc);
4135                 GOTO(out, rc);
4136         }
4137
4138         rc = llog_put_cat_list(disk_obd, name, *index, 1, &catid);
4139         if (rc) {
4140                 CERROR("rc: %d\n", rc);
4141                 GOTO(out, rc);
4142         }
4143
4144  out:
4145         cfs_mutex_up(&olg->olg_cat_processing);
4146
4147         return rc;
4148 }
4149
4150 static int osc_llog_finish(struct obd_device *obd, int count)
4151 {
4152         struct llog_ctxt *ctxt;
4153         int rc = 0, rc2 = 0;
4154         ENTRY;
4155
4156         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
4157         if (ctxt)
4158                 rc = llog_cleanup(ctxt);
4159
4160         ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
4161         if (ctxt)
4162                 rc2 = llog_cleanup(ctxt);
4163         if (!rc)
4164                 rc = rc2;
4165
4166         RETURN(rc);
4167 }
4168
4169 static int osc_reconnect(const struct lu_env *env,
4170                          struct obd_export *exp, struct obd_device *obd,
4171                          struct obd_uuid *cluuid,
4172                          struct obd_connect_data *data,
4173                          void *localdata)
4174 {
4175         struct client_obd *cli = &obd->u.cli;
4176
4177         if (data != NULL && (data->ocd_connect_flags & OBD_CONNECT_GRANT)) {
4178                 long lost_grant;
4179
4180                 client_obd_list_lock(&cli->cl_loi_list_lock);
4181                 data->ocd_grant = (cli->cl_avail_grant + cli->cl_dirty) ?:
4182                                 2 * cli->cl_max_pages_per_rpc << CFS_PAGE_SHIFT;
4183                 lost_grant = cli->cl_lost_grant;
4184                 cli->cl_lost_grant = 0;
4185                 client_obd_list_unlock(&cli->cl_loi_list_lock);
4186
4187                 CDEBUG(D_CACHE, "request ocd_grant: %d cl_avail_grant: %ld "
4188                        "cl_dirty: %ld cl_lost_grant: %ld\n", data->ocd_grant,
4189                        cli->cl_avail_grant, cli->cl_dirty, lost_grant);
4190                 CDEBUG(D_RPCTRACE, "ocd_connect_flags: "LPX64" ocd_version: %d"
4191                        " ocd_grant: %d\n", data->ocd_connect_flags,
4192                        data->ocd_version, data->ocd_grant);
4193         }
4194
4195         RETURN(0);
4196 }
4197
4198 static int osc_disconnect(struct obd_export *exp)
4199 {
4200         struct obd_device *obd = class_exp2obd(exp);
4201         struct llog_ctxt  *ctxt;
4202         int rc;
4203
4204         ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
4205         if (ctxt) {
4206                 if (obd->u.cli.cl_conn_count == 1) {
4207                         /* Flush any remaining cancel messages out to the
4208                          * target */
4209                         llog_sync(ctxt, exp);
4210                 }
4211                 llog_ctxt_put(ctxt);
4212         } else {
4213                 CDEBUG(D_HA, "No LLOG_SIZE_REPL_CTXT found in obd %p\n",
4214                        obd);
4215         }
4216
4217         rc = client_disconnect_export(exp);
4218         /**
4219          * Initially we put del_shrink_grant before disconnect_export, but it
4220          * causes the following problem if setup (connect) and cleanup
4221          * (disconnect) are tangled together.
4222          *      connect p1                     disconnect p2
4223          *   ptlrpc_connect_import
4224          *     ...............               class_manual_cleanup
4225          *                                     osc_disconnect
4226          *                                     del_shrink_grant
4227          *   ptlrpc_connect_interrupt
4228          *     init_grant_shrink
4229          *   add this client to shrink list
4230          *                                      cleanup_osc
4231          * Bang! pinger trigger the shrink.
4232          * So the osc should be disconnected from the shrink list, after we
4233          * are sure the import has been destroyed. BUG18662
4234          */
4235         if (obd->u.cli.cl_import == NULL)
4236                 osc_del_shrink_grant(&obd->u.cli);
4237         return rc;
4238 }
4239
4240 static int osc_import_event(struct obd_device *obd,
4241                             struct obd_import *imp,
4242                             enum obd_import_event event)
4243 {
4244         struct client_obd *cli;
4245         int rc = 0;
4246
4247         ENTRY;
4248         LASSERT(imp->imp_obd == obd);
4249
4250         switch (event) {
4251         case IMP_EVENT_DISCON: {
4252                 /* Only do this on the MDS OSC's */
4253                 if (imp->imp_server_timeout) {
4254                         struct osc_creator *oscc = &obd->u.cli.cl_oscc;
4255
4256                         cfs_spin_lock(&oscc->oscc_lock);
4257                         oscc->oscc_flags |= OSCC_FLAG_RECOVERING;
4258                         cfs_spin_unlock(&oscc->oscc_lock);
4259                 }
4260                 cli = &obd->u.cli;
4261                 client_obd_list_lock(&cli->cl_loi_list_lock);
4262                 cli->cl_avail_grant = 0;
4263                 cli->cl_lost_grant = 0;
4264                 client_obd_list_unlock(&cli->cl_loi_list_lock);
4265                 break;
4266         }
4267         case IMP_EVENT_INACTIVE: {
4268                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_INACTIVE, NULL);
4269                 break;
4270         }
4271         case IMP_EVENT_INVALIDATE: {
4272                 struct ldlm_namespace *ns = obd->obd_namespace;
4273                 struct lu_env         *env;
4274                 int                    refcheck;
4275
4276                 env = cl_env_get(&refcheck);
4277                 if (!IS_ERR(env)) {
4278                         /* Reset grants */
4279                         cli = &obd->u.cli;
4280                         client_obd_list_lock(&cli->cl_loi_list_lock);
4281                         /* all pages go to failing rpcs due to the invalid
4282                          * import */
4283                         osc_check_rpcs(env, cli);
4284                         client_obd_list_unlock(&cli->cl_loi_list_lock);
4285
4286                         ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
4287                         cl_env_put(env, &refcheck);
4288                 } else
4289                         rc = PTR_ERR(env);
4290                 break;
4291         }
4292         case IMP_EVENT_ACTIVE: {
4293                 /* Only do this on the MDS OSC's */
4294                 if (imp->imp_server_timeout) {
4295                         struct osc_creator *oscc = &obd->u.cli.cl_oscc;
4296
4297                         cfs_spin_lock(&oscc->oscc_lock);
4298                         oscc->oscc_flags &= ~OSCC_FLAG_NOSPC;
4299                         cfs_spin_unlock(&oscc->oscc_lock);
4300                 }
4301                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE, NULL);
4302                 break;
4303         }
4304         case IMP_EVENT_OCD: {
4305                 struct obd_connect_data *ocd = &imp->imp_connect_data;
4306
4307                 if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT)
4308                         osc_init_grant(&obd->u.cli, ocd);
4309
4310                 /* See bug 7198 */
4311                 if (ocd->ocd_connect_flags & OBD_CONNECT_REQPORTAL)
4312                         imp->imp_client->cli_request_portal =OST_REQUEST_PORTAL;
4313
4314                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_OCD, NULL);
4315                 break;
4316         }
4317         default:
4318                 CERROR("Unknown import event %d\n", event);
4319                 LBUG();
4320         }
4321         RETURN(rc);
4322 }
4323
4324 int osc_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
4325 {
4326         int rc;
4327         ENTRY;
4328
4329         ENTRY;
4330         rc = ptlrpcd_addref();
4331         if (rc)
4332                 RETURN(rc);
4333
4334         rc = client_obd_setup(obd, lcfg);
4335         if (rc) {
4336                 ptlrpcd_decref();
4337         } else {
4338                 struct lprocfs_static_vars lvars = { 0 };
4339                 struct client_obd *cli = &obd->u.cli;
4340
4341                 cli->cl_grant_shrink_interval = GRANT_SHRINK_INTERVAL;
4342                 lprocfs_osc_init_vars(&lvars);
4343                 if (lprocfs_obd_setup(obd, lvars.obd_vars) == 0) {
4344                         lproc_osc_attach_seqstat(obd);
4345                         sptlrpc_lprocfs_cliobd_attach(obd);
4346                         ptlrpc_lprocfs_register_obd(obd);
4347                 }
4348
4349                 oscc_init(obd);
4350                 /* We need to allocate a few requests more, because
4351                    brw_interpret tries to create new requests before freeing
4352                    previous ones. Ideally we want to have 2x max_rpcs_in_flight
4353                    reserved, but I afraid that might be too much wasted RAM
4354                    in fact, so 2 is just my guess and still should work. */
4355                 cli->cl_import->imp_rq_pool =
4356                         ptlrpc_init_rq_pool(cli->cl_max_rpcs_in_flight + 2,
4357                                             OST_MAXREQSIZE,
4358                                             ptlrpc_add_rqs_to_pool);
4359
4360                 CFS_INIT_LIST_HEAD(&cli->cl_grant_shrink_list);
4361                 cfs_sema_init(&cli->cl_grant_sem, 1);
4362         }
4363
4364         RETURN(rc);
4365 }
4366
4367 static int osc_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
4368 {
4369         int rc = 0;
4370         ENTRY;
4371
4372         switch (stage) {
4373         case OBD_CLEANUP_EARLY: {
4374                 struct obd_import *imp;
4375                 imp = obd->u.cli.cl_import;
4376                 CDEBUG(D_HA, "Deactivating import %s\n", obd->obd_name);
4377                 /* ptlrpc_abort_inflight to stop an mds_lov_synchronize */
4378                 ptlrpc_deactivate_import(imp);
4379                 cfs_spin_lock(&imp->imp_lock);
4380                 imp->imp_pingable = 0;
4381                 cfs_spin_unlock(&imp->imp_lock);
4382                 break;
4383         }
4384         case OBD_CLEANUP_EXPORTS: {
4385                 /* If we set up but never connected, the
4386                    client import will not have been cleaned. */
4387                 if (obd->u.cli.cl_import) {
4388                         struct obd_import *imp;
4389                         cfs_down_write(&obd->u.cli.cl_sem);
4390                         imp = obd->u.cli.cl_import;
4391                         CDEBUG(D_CONFIG, "%s: client import never connected\n",
4392                                obd->obd_name);
4393                         ptlrpc_invalidate_import(imp);
4394                         if (imp->imp_rq_pool) {
4395                                 ptlrpc_free_rq_pool(imp->imp_rq_pool);
4396                                 imp->imp_rq_pool = NULL;
4397                         }
4398                         class_destroy_import(imp);
4399                         cfs_up_write(&obd->u.cli.cl_sem);
4400                         obd->u.cli.cl_import = NULL;
4401                 }
4402                 rc = obd_llog_finish(obd, 0);
4403                 if (rc != 0)
4404                         CERROR("failed to cleanup llogging subsystems\n");
4405                 break;
4406                 }
4407         }
4408         RETURN(rc);
4409 }
4410
4411 int osc_cleanup(struct obd_device *obd)
4412 {
4413         int rc;
4414
4415         ENTRY;
4416         ptlrpc_lprocfs_unregister_obd(obd);
4417         lprocfs_obd_cleanup(obd);
4418
4419         /* free memory of osc quota cache */
4420         lquota_cleanup(quota_interface, obd);
4421
4422         rc = client_obd_cleanup(obd);
4423
4424         ptlrpcd_decref();
4425         RETURN(rc);
4426 }
4427
4428 int osc_process_config_base(struct obd_device *obd, struct lustre_cfg *lcfg)
4429 {
4430         struct lprocfs_static_vars lvars = { 0 };
4431         int rc = 0;
4432
4433         lprocfs_osc_init_vars(&lvars);
4434
4435         switch (lcfg->lcfg_command) {
4436         default:
4437                 rc = class_process_proc_param(PARAM_OSC, lvars.obd_vars,
4438                                               lcfg, obd);
4439                 if (rc > 0)
4440                         rc = 0;
4441                 break;
4442         }
4443
4444         return(rc);
4445 }
4446
4447 static int osc_process_config(struct obd_device *obd, obd_count len, void *buf)
4448 {
4449         return osc_process_config_base(obd, buf);
4450 }
4451
4452 struct obd_ops osc_obd_ops = {
4453         .o_owner                = THIS_MODULE,
4454         .o_setup                = osc_setup,
4455         .o_precleanup           = osc_precleanup,
4456         .o_cleanup              = osc_cleanup,
4457         .o_add_conn             = client_import_add_conn,
4458         .o_del_conn             = client_import_del_conn,
4459         .o_connect              = client_connect_import,
4460         .o_reconnect            = osc_reconnect,
4461         .o_disconnect           = osc_disconnect,
4462         .o_statfs               = osc_statfs,
4463         .o_statfs_async         = osc_statfs_async,
4464         .o_packmd               = osc_packmd,
4465         .o_unpackmd             = osc_unpackmd,
4466         .o_precreate            = osc_precreate,
4467         .o_create               = osc_create,
4468         .o_create_async         = osc_create_async,
4469         .o_destroy              = osc_destroy,
4470         .o_getattr              = osc_getattr,
4471         .o_getattr_async        = osc_getattr_async,
4472         .o_setattr              = osc_setattr,
4473         .o_setattr_async        = osc_setattr_async,
4474         .o_brw                  = osc_brw,
4475         .o_punch                = osc_punch,
4476         .o_sync                 = osc_sync,
4477         .o_enqueue              = osc_enqueue,
4478         .o_change_cbdata        = osc_change_cbdata,
4479         .o_find_cbdata          = osc_find_cbdata,
4480         .o_cancel               = osc_cancel,
4481         .o_cancel_unused        = osc_cancel_unused,
4482         .o_iocontrol            = osc_iocontrol,
4483         .o_get_info             = osc_get_info,
4484         .o_set_info_async       = osc_set_info_async,
4485         .o_import_event         = osc_import_event,
4486         .o_llog_init            = osc_llog_init,
4487         .o_llog_finish          = osc_llog_finish,
4488         .o_process_config       = osc_process_config,
4489 };
4490
4491 extern struct lu_kmem_descr osc_caches[];
4492 extern cfs_spinlock_t       osc_ast_guard;
4493 extern cfs_lock_class_key_t osc_ast_guard_class;
4494
4495 int __init osc_init(void)
4496 {
4497         struct lprocfs_static_vars lvars = { 0 };
4498         int rc;
4499         ENTRY;
4500
4501         /* print an address of _any_ initialized kernel symbol from this
4502          * module, to allow debugging with gdb that doesn't support data
4503          * symbols from modules.*/
4504         CDEBUG(D_CONSOLE, "Lustre OSC module (%p).\n", &osc_caches);
4505
4506         rc = lu_kmem_init(osc_caches);
4507
4508         lprocfs_osc_init_vars(&lvars);
4509
4510         cfs_request_module("lquota");
4511         quota_interface = PORTAL_SYMBOL_GET(osc_quota_interface);
4512         lquota_init(quota_interface);
4513         init_obd_quota_ops(quota_interface, &osc_obd_ops);
4514
4515         rc = class_register_type(&osc_obd_ops, NULL, lvars.module_vars,
4516                                  LUSTRE_OSC_NAME, &osc_device_type);
4517         if (rc) {
4518                 if (quota_interface)
4519                         PORTAL_SYMBOL_PUT(osc_quota_interface);
4520                 lu_kmem_fini(osc_caches);
4521                 RETURN(rc);
4522         }
4523
4524         cfs_spin_lock_init(&osc_ast_guard);
4525         cfs_lockdep_set_class(&osc_ast_guard, &osc_ast_guard_class);
4526
4527         osc_mds_ost_orig_logops = llog_lvfs_ops;
4528         osc_mds_ost_orig_logops.lop_setup = llog_obd_origin_setup;
4529         osc_mds_ost_orig_logops.lop_cleanup = llog_obd_origin_cleanup;
4530         osc_mds_ost_orig_logops.lop_add = llog_obd_origin_add;
4531         osc_mds_ost_orig_logops.lop_connect = llog_origin_connect;
4532
4533         RETURN(rc);
4534 }
4535
4536 #ifdef __KERNEL__
4537 static void /*__exit*/ osc_exit(void)
4538 {
4539         lu_device_type_fini(&osc_device_type);
4540
4541         lquota_exit(quota_interface);
4542         if (quota_interface)
4543                 PORTAL_SYMBOL_PUT(osc_quota_interface);
4544
4545         class_unregister_type(LUSTRE_OSC_NAME);
4546         lu_kmem_fini(osc_caches);
4547 }
4548
4549 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
4550 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC)");
4551 MODULE_LICENSE("GPL");
4552
4553 cfs_module(osc, LUSTRE_VERSION_STRING, osc_init, osc_exit);
4554 #endif