lustre/osc/osc_request.c

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