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