lustre/osc/osc_request.c

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