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