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