lustre/osc/osc_request.c

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