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         cli->cl_next_shrink_grant =
 804                 cfs_time_shift(cli->cl_grant_shrink_interval);
 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 /* Shrink the current grant, either from some large amount to enough for a
 968  * full set of in-flight RPCs, or if we have already shrunk to that limit
 969  * then to enough for a single RPC.  This avoids keeping more grant than
 970  * needed, and avoids shrinking the grant piecemeal. */
 971 static int osc_shrink_grant(struct client_obd *cli)
 972 {
 973         long target = (cli->cl_max_rpcs_in_flight + 1) *
 974                       cli->cl_max_pages_per_rpc;
 975
 976         client_obd_list_lock(&cli->cl_loi_list_lock);
 977         if (cli->cl_avail_grant <= target)
 978                 target = cli->cl_max_pages_per_rpc;
 979         client_obd_list_unlock(&cli->cl_loi_list_lock);
 980
 981         return osc_shrink_grant_to_target(cli, target);
 982 }
 983
 984 int osc_shrink_grant_to_target(struct client_obd *cli, long target)
 985 {
 986         int    rc = 0;
 987         struct ost_body     *body;
 988         ENTRY;
 989
 990         client_obd_list_lock(&cli->cl_loi_list_lock);
 991         /* Don't shrink if we are already above or below the desired limit
 992          * We don't want to shrink below a single RPC, as that will negatively
 993          * impact block allocation and long-term performance. */
 994         if (target < cli->cl_max_pages_per_rpc)
 995                 target = cli->cl_max_pages_per_rpc;
 996
 997         if (target >= cli->cl_avail_grant) {
 998                 client_obd_list_unlock(&cli->cl_loi_list_lock);
 999                 RETURN(0);
1000         }
1001         client_obd_list_unlock(&cli->cl_loi_list_lock);
1002
1003         OBD_ALLOC_PTR(body);
1004         if (!body)
1005                 RETURN(-ENOMEM);
1006
1007         osc_announce_cached(cli, &body->oa, 0);
1008
1009         client_obd_list_lock(&cli->cl_loi_list_lock);
1010         body->oa.o_grant = cli->cl_avail_grant - target;
1011         cli->cl_avail_grant = target;
1012         client_obd_list_unlock(&cli->cl_loi_list_lock);
1013         body->oa.o_flags |= OBD_FL_SHRINK_GRANT;
1014         osc_update_next_shrink(cli);
1015
1016         rc = osc_set_info_async(cli->cl_import->imp_obd->obd_self_export,
1017                                 sizeof(KEY_GRANT_SHRINK), KEY_GRANT_SHRINK,
1018                                 sizeof(*body), body, NULL);
1019         if (rc != 0)
1020                 __osc_update_grant(cli, body->oa.o_grant);
1021         OBD_FREE_PTR(body);
1022         RETURN(rc);
1023 }
1024
1025 #define GRANT_SHRINK_LIMIT PTLRPC_MAX_BRW_SIZE
1026 static int osc_should_shrink_grant(struct client_obd *client)
1027 {
1028         cfs_time_t time = cfs_time_current();
1029         cfs_time_t next_shrink = client->cl_next_shrink_grant;
1030         if (cfs_time_aftereq(time, next_shrink - 5 * CFS_TICK)) {
1031                 if (client->cl_import->imp_state == LUSTRE_IMP_FULL &&
1032                     client->cl_avail_grant > GRANT_SHRINK_LIMIT)
1033                         return 1;
1034                 else
1035                         osc_update_next_shrink(client);
1036         }
1037         return 0;
1038 }
1039
1040 static int osc_grant_shrink_grant_cb(struct timeout_item *item, void *data)
1041 {
1042         struct client_obd *client;
1043
1044         list_for_each_entry(client, &item->ti_obd_list, cl_grant_shrink_list) {
1045                 if (osc_should_shrink_grant(client))
1046                         osc_shrink_grant(client);
1047         }
1048         return 0;
1049 }
1050
1051 static int osc_add_shrink_grant(struct client_obd *client)
1052 {
1053         int rc;
1054
1055         rc = ptlrpc_add_timeout_client(client->cl_grant_shrink_interval,
1056                                        TIMEOUT_GRANT,
1057                                        osc_grant_shrink_grant_cb, NULL,
1058                                        &client->cl_grant_shrink_list);
1059         if (rc) {
1060                 CERROR("add grant client %s error %d\n",
1061                         client->cl_import->imp_obd->obd_name, rc);
1062                 return rc;
1063         }
1064         CDEBUG(D_CACHE, "add grant client %s \n",
1065                client->cl_import->imp_obd->obd_name);
1066         osc_update_next_shrink(client);
1067         return 0;
1068 }
1069
1070 static int osc_del_shrink_grant(struct client_obd *client)
1071 {
1072         return ptlrpc_del_timeout_client(&client->cl_grant_shrink_list,
1073                                          TIMEOUT_GRANT);
1074 }
1075
1076 static void osc_init_grant(struct client_obd *cli, struct obd_connect_data *ocd)
1077 {
1078         client_obd_list_lock(&cli->cl_loi_list_lock);
1079         cli->cl_avail_grant = ocd->ocd_grant;
1080         client_obd_list_unlock(&cli->cl_loi_list_lock);
1081
1082         if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT_SHRINK &&
1083             list_empty(&cli->cl_grant_shrink_list))
1084                 osc_add_shrink_grant(cli);
1085
1086         CDEBUG(D_CACHE, "setting cl_avail_grant: %ld cl_lost_grant: %ld \n",
1087                cli->cl_avail_grant, cli->cl_lost_grant);
1088         LASSERT(cli->cl_avail_grant >= 0);
1089 }
1090
1091 /* We assume that the reason this OSC got a short read is because it read
1092  * beyond the end of a stripe file; i.e. lustre is reading a sparse file
1093  * via the LOV, and it _knows_ it's reading inside the file, it's just that
1094  * this stripe never got written at or beyond this stripe offset yet. */
1095 static void handle_short_read(int nob_read, obd_count page_count,
1096                               struct brw_page **pga)
1097 {
1098         char *ptr;
1099         int i = 0;
1100
1101         /* skip bytes read OK */
1102         while (nob_read > 0) {
1103                 LASSERT (page_count > 0);
1104
1105                 if (pga[i]->count > nob_read) {
1106                         /* EOF inside this page */
1107                         ptr = cfs_kmap(pga[i]->pg) +
1108                                 (pga[i]->off & ~CFS_PAGE_MASK);
1109                         memset(ptr + nob_read, 0, pga[i]->count - nob_read);
1110                         cfs_kunmap(pga[i]->pg);
1111                         page_count--;
1112                         i++;
1113                         break;
1114                 }
1115
1116                 nob_read -= pga[i]->count;
1117                 page_count--;
1118                 i++;
1119         }
1120
1121         /* zero remaining pages */
1122         while (page_count-- > 0) {
1123                 ptr = cfs_kmap(pga[i]->pg) + (pga[i]->off & ~CFS_PAGE_MASK);
1124                 memset(ptr, 0, pga[i]->count);
1125                 cfs_kunmap(pga[i]->pg);
1126                 i++;
1127         }
1128 }
1129
1130 static int check_write_rcs(struct ptlrpc_request *req,
1131                            int requested_nob, int niocount,
1132                            obd_count page_count, struct brw_page **pga)
1133 {
1134         int    *remote_rcs, i;
1135
1136         /* return error if any niobuf was in error */
1137         remote_rcs = lustre_swab_repbuf(req, REQ_REC_OFF + 1,
1138                                         sizeof(*remote_rcs) * niocount, NULL);
1139         if (remote_rcs == NULL) {
1140                 CDEBUG(D_INFO, "Missing/short RC vector on BRW_WRITE reply\n");
1141                 return(-EPROTO);
1142         }
1143         if (lustre_msg_swabbed(req->rq_repmsg))
1144                 for (i = 0; i < niocount; i++)
1145                         __swab32s(&remote_rcs[i]);
1146
1147         for (i = 0; i < niocount; i++) {
1148                 if (remote_rcs[i] < 0)
1149                         return(remote_rcs[i]);
1150
1151                 if (remote_rcs[i] != 0) {
1152                         CDEBUG(D_INFO, "rc[%d] invalid (%d) req %p\n",
1153                                 i, remote_rcs[i], req);
1154                         return(-EPROTO);
1155                 }
1156         }
1157
1158         if (req->rq_bulk->bd_nob_transferred != requested_nob) {
1159                 CERROR("Unexpected # bytes transferred: %d (requested %d)\n",
1160                        req->rq_bulk->bd_nob_transferred, requested_nob);
1161                 return(-EPROTO);
1162         }
1163
1164         return (0);
1165 }
1166
1167 static inline int can_merge_pages(struct brw_page *p1, struct brw_page *p2)
1168 {
1169         if (p1->flag != p2->flag) {
1170                 unsigned mask = ~(OBD_BRW_FROM_GRANT|
1171                                   OBD_BRW_NOCACHE|OBD_BRW_SYNC);
1172
1173                 /* warn if we try to combine flags that we don't know to be
1174                  * safe to combine */
1175                 if ((p1->flag & mask) != (p2->flag & mask))
1176                         CERROR("is it ok to have flags 0x%x and 0x%x in the "
1177                                "same brw?\n", p1->flag, p2->flag);
1178                 return 0;
1179         }
1180
1181         return (p1->off + p1->count == p2->off);
1182 }
1183
1184 static obd_count osc_checksum_bulk(int nob, obd_count pg_count,
1185                                    struct brw_page **pga, int opc,
1186                                    cksum_type_t cksum_type)
1187 {
1188         __u32 cksum;
1189         int i = 0;
1190
1191         LASSERT (pg_count > 0);
1192         cksum = init_checksum(cksum_type);
1193         while (nob > 0 && pg_count > 0) {
1194                 unsigned char *ptr = cfs_kmap(pga[i]->pg);
1195                 int off = pga[i]->off & ~CFS_PAGE_MASK;
1196                 int count = pga[i]->count > nob ? nob : pga[i]->count;
1197
1198                 /* corrupt the data before we compute the checksum, to
1199                  * simulate an OST->client data error */
1200                 if (i == 0 && opc == OST_READ &&
1201                     OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_RECEIVE))
1202                         memcpy(ptr + off, "bad1", min(4, nob));
1203                 cksum = compute_checksum(cksum, ptr + off, count, cksum_type);
1204                 cfs_kunmap(pga[i]->pg);
1205                 LL_CDEBUG_PAGE(D_PAGE, pga[i]->pg, "off %d checksum %x\n",
1206                                off, cksum);
1207
1208                 nob -= pga[i]->count;
1209                 pg_count--;
1210                 i++;
1211         }
1212         /* For sending we only compute the wrong checksum instead
1213          * of corrupting the data so it is still correct on a redo */
1214         if (opc == OST_WRITE && OBD_FAIL_CHECK(OBD_FAIL_OSC_CHECKSUM_SEND))
1215                 cksum++;
1216
1217         return cksum;
1218 }
1219
1220 static int osc_brw_prep_request(int cmd, struct client_obd *cli,struct obdo *oa,
1221                                 struct lov_stripe_md *lsm, obd_count page_count,
1222                                 struct brw_page **pga,
1223                                 struct ptlrpc_request **reqp,
1224                                 struct obd_capa *ocapa, int reserve)
1225 {
1226         struct ptlrpc_request   *req;
1227         struct ptlrpc_bulk_desc *desc;
1228         struct ost_body         *body;
1229         struct obd_ioobj        *ioobj;
1230         struct niobuf_remote    *niobuf;
1231         int niocount, i, requested_nob, opc, rc;
1232         struct osc_brw_async_args *aa;
1233         struct req_capsule      *pill;
1234         struct brw_page *pg_prev;
1235
1236         ENTRY;
1237         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_PREP_REQ))
1238                 RETURN(-ENOMEM); /* Recoverable */
1239         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_PREP_REQ2))
1240                 RETURN(-EINVAL); /* Fatal */
1241
1242         if ((cmd & OBD_BRW_WRITE) != 0) {
1243                 opc = OST_WRITE;
1244                 req = ptlrpc_request_alloc_pool(cli->cl_import,
1245                                                 cli->cl_import->imp_rq_pool,
1246                                                 &RQF_OST_BRW);
1247         } else {
1248                 opc = OST_READ;
1249                 req = ptlrpc_request_alloc(cli->cl_import, &RQF_OST_BRW);
1250         }
1251         if (req == NULL)
1252                 RETURN(-ENOMEM);
1253
1254         for (niocount = i = 1; i < page_count; i++) {
1255                 if (!can_merge_pages(pga[i - 1], pga[i]))
1256                         niocount++;
1257         }
1258
1259         pill = &req->rq_pill;
1260         req_capsule_set_size(pill, &RMF_NIOBUF_REMOTE, RCL_CLIENT,
1261                              niocount * sizeof(*niobuf));
1262         osc_set_capa_size(req, &RMF_CAPA1, ocapa);
1263
1264         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, opc);
1265         if (rc) {
1266                 ptlrpc_request_free(req);
1267                 RETURN(rc);
1268         }
1269         req->rq_request_portal = OST_IO_PORTAL; /* bug 7198 */
1270         ptlrpc_at_set_req_timeout(req);
1271
1272         if (opc == OST_WRITE)
1273                 desc = ptlrpc_prep_bulk_imp(req, page_count,
1274                                             BULK_GET_SOURCE, OST_BULK_PORTAL);
1275         else
1276                 desc = ptlrpc_prep_bulk_imp(req, page_count,
1277                                             BULK_PUT_SINK, OST_BULK_PORTAL);
1278
1279         if (desc == NULL)
1280                 GOTO(out, rc = -ENOMEM);
1281         /* NB request now owns desc and will free it when it gets freed */
1282
1283         body = req_capsule_client_get(pill, &RMF_OST_BODY);
1284         ioobj = req_capsule_client_get(pill, &RMF_OBD_IOOBJ);
1285         niobuf = req_capsule_client_get(pill, &RMF_NIOBUF_REMOTE);
1286         LASSERT(body && ioobj && niobuf);
1287
1288         body->oa = *oa;
1289
1290         obdo_to_ioobj(oa, ioobj);
1291         ioobj->ioo_bufcnt = niocount;
1292         osc_pack_capa(req, body, ocapa);
1293         LASSERT (page_count > 0);
1294         pg_prev = pga[0];
1295         for (requested_nob = i = 0; i < page_count; i++, niobuf++) {
1296                 struct brw_page *pg = pga[i];
1297
1298                 LASSERT(pg->count > 0);
1299                 LASSERTF((pg->off & ~CFS_PAGE_MASK) + pg->count <= CFS_PAGE_SIZE,
1300                          "i: %d pg: %p off: "LPU64", count: %u\n", i, pg,
1301                          pg->off, pg->count);
1302 #ifdef __linux__
1303                 LASSERTF(i == 0 || pg->off > pg_prev->off,
1304                          "i %d p_c %u pg %p [pri %lu ind %lu] off "LPU64
1305                          " prev_pg %p [pri %lu ind %lu] off "LPU64"\n",
1306                          i, page_count,
1307                          pg->pg, page_private(pg->pg), pg->pg->index, pg->off,
1308                          pg_prev->pg, page_private(pg_prev->pg),
1309                          pg_prev->pg->index, pg_prev->off);
1310 #else
1311                 LASSERTF(i == 0 || pg->off > pg_prev->off,
1312                          "i %d p_c %u\n", i, page_count);
1313 #endif
1314                 LASSERT((pga[0]->flag & OBD_BRW_SRVLOCK) ==
1315                         (pg->flag & OBD_BRW_SRVLOCK));
1316
1317                 ptlrpc_prep_bulk_page(desc, pg->pg, pg->off & ~CFS_PAGE_MASK,
1318                                       pg->count);
1319                 requested_nob += pg->count;
1320
1321                 if (i > 0 && can_merge_pages(pg_prev, pg)) {
1322                         niobuf--;
1323                         niobuf->len += pg->count;
1324                 } else {
1325                         niobuf->offset = pg->off;
1326                         niobuf->len    = pg->count;
1327                         niobuf->flags  = pg->flag;
1328                 }
1329                 pg_prev = pg;
1330         }
1331
1332         LASSERTF((void *)(niobuf - niocount) ==
1333                 lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 2,
1334                                niocount * sizeof(*niobuf)),
1335                 "want %p - real %p\n", lustre_msg_buf(req->rq_reqmsg,
1336                 REQ_REC_OFF + 2, niocount * sizeof(*niobuf)),
1337                 (void *)(niobuf - niocount));
1338
1339         osc_announce_cached(cli, &body->oa, opc == OST_WRITE ? requested_nob:0);
1340         if (osc_should_shrink_grant(cli))
1341                 osc_shrink_grant_local(cli, &body->oa);
1342
1343         /* size[REQ_REC_OFF] still sizeof (*body) */
1344         if (opc == OST_WRITE) {
1345                 if (unlikely(cli->cl_checksum) &&
1346                     !sptlrpc_flavor_has_bulk(&req->rq_flvr)) {
1347                         /* store cl_cksum_type in a local variable since
1348                          * it can be changed via lprocfs */
1349                         cksum_type_t cksum_type = cli->cl_cksum_type;
1350
1351                         if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0)
1352                                 oa->o_flags = body->oa.o_flags = 0;
1353                         body->oa.o_flags |= cksum_type_pack(cksum_type);
1354                         body->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1355                         body->oa.o_cksum = osc_checksum_bulk(requested_nob,
1356                                                              page_count, pga,
1357                                                              OST_WRITE,
1358                                                              cksum_type);
1359                         CDEBUG(D_PAGE, "checksum at write origin: %x\n",
1360                                body->oa.o_cksum);
1361                         /* save this in 'oa', too, for later checking */
1362                         oa->o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1363                         oa->o_flags |= cksum_type_pack(cksum_type);
1364                 } else {
1365                         /* clear out the checksum flag, in case this is a
1366                          * resend but cl_checksum is no longer set. b=11238 */
1367                         oa->o_valid &= ~OBD_MD_FLCKSUM;
1368                 }
1369                 oa->o_cksum = body->oa.o_cksum;
1370                 /* 1 RC per niobuf */
1371                 req_capsule_set_size(pill, &RMF_NIOBUF_REMOTE, RCL_SERVER,
1372                                      sizeof(__u32) * niocount);
1373         } else {
1374                 if (unlikely(cli->cl_checksum) &&
1375                     !sptlrpc_flavor_has_bulk(&req->rq_flvr)) {
1376                         if ((body->oa.o_valid & OBD_MD_FLFLAGS) == 0)
1377                                 body->oa.o_flags = 0;
1378                         body->oa.o_flags |= cksum_type_pack(cli->cl_cksum_type);
1379                         body->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1380                 }
1381                 req_capsule_set_size(pill, &RMF_NIOBUF_REMOTE, RCL_SERVER, 0);
1382                 /* 1 RC for the whole I/O */
1383         }
1384         ptlrpc_request_set_replen(req);
1385
1386         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
1387         aa = ptlrpc_req_async_args(req);
1388         aa->aa_oa = oa;
1389         aa->aa_requested_nob = requested_nob;
1390         aa->aa_nio_count = niocount;
1391         aa->aa_page_count = page_count;
1392         aa->aa_resends = 0;
1393         aa->aa_ppga = pga;
1394         aa->aa_cli = cli;
1395         CFS_INIT_LIST_HEAD(&aa->aa_oaps);
1396         if (ocapa && reserve)
1397                 aa->aa_ocapa = capa_get(ocapa);
1398
1399         *reqp = req;
1400         RETURN(0);
1401
1402  out:
1403         ptlrpc_req_finished(req);
1404         RETURN(rc);
1405 }
1406
1407 static int check_write_checksum(struct obdo *oa, const lnet_process_id_t *peer,
1408                                 __u32 client_cksum, __u32 server_cksum, int nob,
1409                                 obd_count page_count, struct brw_page **pga,
1410                                 cksum_type_t client_cksum_type)
1411 {
1412         __u32 new_cksum;
1413         char *msg;
1414         cksum_type_t cksum_type;
1415
1416         if (server_cksum == client_cksum) {
1417                 CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
1418                 return 0;
1419         }
1420
1421         if (oa->o_valid & OBD_MD_FLFLAGS)
1422                 cksum_type = cksum_type_unpack(oa->o_flags);
1423         else
1424                 cksum_type = OBD_CKSUM_CRC32;
1425
1426         new_cksum = osc_checksum_bulk(nob, page_count, pga, OST_WRITE,
1427                                       cksum_type);
1428
1429         if (cksum_type != client_cksum_type)
1430                 msg = "the server did not use the checksum type specified in "
1431                       "the original request - likely a protocol problem";
1432         else if (new_cksum == server_cksum)
1433                 msg = "changed on the client after we checksummed it - "
1434                       "likely false positive due to mmap IO (bug 11742)";
1435         else if (new_cksum == client_cksum)
1436                 msg = "changed in transit before arrival at OST";
1437         else
1438                 msg = "changed in transit AND doesn't match the original - "
1439                       "likely false positive due to mmap IO (bug 11742)";
1440
1441         LCONSOLE_ERROR_MSG(0x132, "BAD WRITE CHECKSUM: %s: from %s inum "
1442                            LPU64"/"LPU64" object "LPU64"/"LPU64" extent "
1443                            "["LPU64"-"LPU64"]\n",
1444                            msg, libcfs_nid2str(peer->nid),
1445                            oa->o_valid & OBD_MD_FLFID ? oa->o_fid : (__u64)0,
1446                            oa->o_valid & OBD_MD_FLFID ? oa->o_generation :
1447                                                         (__u64)0,
1448                            oa->o_id,
1449                            oa->o_valid & OBD_MD_FLGROUP ? oa->o_gr : (__u64)0,
1450                            pga[0]->off,
1451                            pga[page_count-1]->off + pga[page_count-1]->count - 1);
1452         CERROR("original client csum %x (type %x), server csum %x (type %x), "
1453                "client csum now %x\n", client_cksum, client_cksum_type,
1454                server_cksum, cksum_type, new_cksum);
1455         return 1;
1456 }
1457
1458 /* Note rc enters this function as number of bytes transferred */
1459 static int osc_brw_fini_request(struct ptlrpc_request *req, int rc)
1460 {
1461         struct osc_brw_async_args *aa = (void *)&req->rq_async_args;
1462         const lnet_process_id_t *peer =
1463                         &req->rq_import->imp_connection->c_peer;
1464         struct client_obd *cli = aa->aa_cli;
1465         struct ost_body *body;
1466         __u32 client_cksum = 0;
1467         ENTRY;
1468
1469         if (rc < 0 && rc != -EDQUOT)
1470                 RETURN(rc);
1471
1472         LASSERTF(req->rq_repmsg != NULL, "rc = %d\n", rc);
1473         body = lustre_swab_repbuf(req, REPLY_REC_OFF, sizeof(*body),
1474                                   lustre_swab_ost_body);
1475         if (body == NULL) {
1476                 CDEBUG(D_INFO, "Can't unpack body\n");
1477                 RETURN(-EPROTO);
1478         }
1479
1480         /* set/clear over quota flag for a uid/gid */
1481         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE &&
1482             body->oa.o_valid & (OBD_MD_FLUSRQUOTA | OBD_MD_FLGRPQUOTA)) {
1483                 unsigned int qid[MAXQUOTAS] = { body->oa.o_uid, body->oa.o_gid };
1484
1485                 lquota_setdq(quota_interface, cli, qid, body->oa.o_valid,
1486                              body->oa.o_flags);
1487         }
1488
1489         if (rc < 0)
1490                 RETURN(rc);
1491
1492         if (aa->aa_oa->o_valid & OBD_MD_FLCKSUM)
1493                 client_cksum = aa->aa_oa->o_cksum; /* save for later */
1494
1495         osc_update_grant(cli, body);
1496
1497         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE) {
1498                 if (rc > 0) {
1499                         CERROR("Unexpected +ve rc %d\n", rc);
1500                         RETURN(-EPROTO);
1501                 }
1502                 LASSERT(req->rq_bulk->bd_nob == aa->aa_requested_nob);
1503
1504                 if (sptlrpc_cli_unwrap_bulk_write(req, req->rq_bulk))
1505                         RETURN(-EAGAIN);
1506
1507                 if ((aa->aa_oa->o_valid & OBD_MD_FLCKSUM) && client_cksum &&
1508                     check_write_checksum(&body->oa, peer, client_cksum,
1509                                          body->oa.o_cksum, aa->aa_requested_nob,
1510                                          aa->aa_page_count, aa->aa_ppga,
1511                                          cksum_type_unpack(aa->aa_oa->o_flags)))
1512                         RETURN(-EAGAIN);
1513
1514                 rc = check_write_rcs(req, aa->aa_requested_nob,aa->aa_nio_count,
1515                                      aa->aa_page_count, aa->aa_ppga);
1516                 GOTO(out, rc);
1517         }
1518
1519         /* The rest of this function executes only for OST_READs */
1520
1521         /* if unwrap_bulk failed, return -EAGAIN to retry */
1522         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk, rc);
1523         if (rc < 0)
1524                 GOTO(out, rc = -EAGAIN);
1525
1526         if (rc > aa->aa_requested_nob) {
1527                 CERROR("Unexpected rc %d (%d requested)\n", rc,
1528                        aa->aa_requested_nob);
1529                 RETURN(-EPROTO);
1530         }
1531
1532         if (rc != req->rq_bulk->bd_nob_transferred) {
1533                 CERROR ("Unexpected rc %d (%d transferred)\n",
1534                         rc, req->rq_bulk->bd_nob_transferred);
1535                 return (-EPROTO);
1536         }
1537
1538         if (rc < aa->aa_requested_nob)
1539                 handle_short_read(rc, aa->aa_page_count, aa->aa_ppga);
1540
1541         if (body->oa.o_valid & OBD_MD_FLCKSUM) {
1542                 static int cksum_counter;
1543                 __u32      server_cksum = body->oa.o_cksum;
1544                 char      *via;
1545                 char      *router;
1546                 cksum_type_t cksum_type;
1547
1548                 if (body->oa.o_valid & OBD_MD_FLFLAGS)
1549                         cksum_type = cksum_type_unpack(body->oa.o_flags);
1550                 else
1551                         cksum_type = OBD_CKSUM_CRC32;
1552                 client_cksum = osc_checksum_bulk(rc, aa->aa_page_count,
1553                                                  aa->aa_ppga, OST_READ,
1554                                                  cksum_type);
1555
1556                 if (peer->nid == req->rq_bulk->bd_sender) {
1557                         via = router = "";
1558                 } else {
1559                         via = " via ";
1560                         router = libcfs_nid2str(req->rq_bulk->bd_sender);
1561                 }
1562
1563                 if (server_cksum == ~0 && rc > 0) {
1564                         CERROR("Protocol error: server %s set the 'checksum' "
1565                                "bit, but didn't send a checksum.  Not fatal, "
1566                                "but please notify on http://bugzilla.lustre.org/\n",
1567                                libcfs_nid2str(peer->nid));
1568                 } else if (server_cksum != client_cksum) {
1569                         LCONSOLE_ERROR_MSG(0x133, "%s: BAD READ CHECKSUM: from "
1570                                            "%s%s%s inum "LPU64"/"LPU64" object "
1571                                            LPU64"/"LPU64" extent "
1572                                            "["LPU64"-"LPU64"]\n",
1573                                            req->rq_import->imp_obd->obd_name,
1574                                            libcfs_nid2str(peer->nid),
1575                                            via, router,
1576                                            body->oa.o_valid & OBD_MD_FLFID ?
1577                                                 body->oa.o_fid : (__u64)0,
1578                                            body->oa.o_valid & OBD_MD_FLFID ?
1579                                                 body->oa.o_generation :(__u64)0,
1580                                            body->oa.o_id,
1581                                            body->oa.o_valid & OBD_MD_FLGROUP ?
1582                                                 body->oa.o_gr : (__u64)0,
1583                                            aa->aa_ppga[0]->off,
1584                                            aa->aa_ppga[aa->aa_page_count-1]->off +
1585                                            aa->aa_ppga[aa->aa_page_count-1]->count -
1586                                                                         1);
1587                         CERROR("client %x, server %x, cksum_type %x\n",
1588                                client_cksum, server_cksum, cksum_type);
1589                         cksum_counter = 0;
1590                         aa->aa_oa->o_cksum = client_cksum;
1591                         rc = -EAGAIN;
1592                 } else {
1593                         cksum_counter++;
1594                         CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
1595                         rc = 0;
1596                 }
1597         } else if (unlikely(client_cksum)) {
1598                 static int cksum_missed;
1599
1600                 cksum_missed++;
1601                 if ((cksum_missed & (-cksum_missed)) == cksum_missed)
1602                         CERROR("Checksum %u requested from %s but not sent\n",
1603                                cksum_missed, libcfs_nid2str(peer->nid));
1604         } else {
1605                 rc = 0;
1606         }
1607 out:
1608         if (rc >= 0)
1609                 *aa->aa_oa = body->oa;
1610
1611         RETURN(rc);
1612 }
1613
1614 static int osc_brw_internal(int cmd, struct obd_export *exp, struct obdo *oa,
1615                             struct lov_stripe_md *lsm,
1616                             obd_count page_count, struct brw_page **pga,
1617                             struct obd_capa *ocapa)
1618 {
1619         struct ptlrpc_request *req;
1620         int                    rc;
1621         cfs_waitq_t            waitq;
1622         int                    resends = 0;
1623         struct l_wait_info     lwi;
1624
1625         ENTRY;
1626
1627         cfs_waitq_init(&waitq);
1628
1629 restart_bulk:
1630         rc = osc_brw_prep_request(cmd, &exp->exp_obd->u.cli, oa, lsm,
1631                                   page_count, pga, &req, ocapa, 0);
1632         if (rc != 0)
1633                 return (rc);
1634
1635         rc = ptlrpc_queue_wait(req);
1636
1637         if (rc == -ETIMEDOUT && req->rq_resend) {
1638                 DEBUG_REQ(D_HA, req,  "BULK TIMEOUT");
1639                 ptlrpc_req_finished(req);
1640                 goto restart_bulk;
1641         }
1642
1643         rc = osc_brw_fini_request(req, rc);
1644
1645         ptlrpc_req_finished(req);
1646         if (osc_recoverable_error(rc)) {
1647                 resends++;
1648                 if (!osc_should_resend(resends, &exp->exp_obd->u.cli)) {
1649                         CERROR("too many resend retries, returning error\n");
1650                         RETURN(-EIO);
1651                 }
1652
1653                 lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(resends), NULL, NULL, NULL);
1654                 l_wait_event(waitq, 0, &lwi);
1655
1656                 goto restart_bulk;
1657         }
1658
1659         RETURN (rc);
1660 }
1661
1662 int osc_brw_redo_request(struct ptlrpc_request *request,
1663                          struct osc_brw_async_args *aa)
1664 {
1665         struct ptlrpc_request *new_req;
1666         struct ptlrpc_request_set *set = request->rq_set;
1667         struct osc_brw_async_args *new_aa;
1668         struct osc_async_page *oap;
1669         int rc = 0;
1670         ENTRY;
1671
1672         if (!osc_should_resend(aa->aa_resends, aa->aa_cli)) {
1673                 CERROR("too many resend retries, returning error\n");
1674                 RETURN(-EIO);
1675         }
1676
1677         DEBUG_REQ(D_ERROR, request, "redo for recoverable error");
1678
1679         rc = osc_brw_prep_request(lustre_msg_get_opc(request->rq_reqmsg) ==
1680                                         OST_WRITE ? OBD_BRW_WRITE :OBD_BRW_READ,
1681                                   aa->aa_cli, aa->aa_oa,
1682                                   NULL /* lsm unused by osc currently */,
1683                                   aa->aa_page_count, aa->aa_ppga,
1684                                   &new_req, aa->aa_ocapa, 0);
1685         if (rc)
1686                 RETURN(rc);
1687
1688         client_obd_list_lock(&aa->aa_cli->cl_loi_list_lock);
1689
1690         list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
1691                 if (oap->oap_request != NULL) {
1692                         LASSERTF(request == oap->oap_request,
1693                                  "request %p != oap_request %p\n",
1694                                  request, oap->oap_request);
1695                         if (oap->oap_interrupted) {
1696                                 client_obd_list_unlock(&aa->aa_cli->cl_loi_list_lock);
1697                                 ptlrpc_req_finished(new_req);
1698                                 RETURN(-EINTR);
1699                         }
1700                 }
1701         }
1702         /* New request takes over pga and oaps from old request.
1703          * Note that copying a list_head doesn't work, need to move it... */
1704         aa->aa_resends++;
1705         new_req->rq_interpret_reply = request->rq_interpret_reply;
1706         new_req->rq_async_args = request->rq_async_args;
1707         new_req->rq_sent = cfs_time_current_sec() + aa->aa_resends;
1708
1709         new_aa = ptlrpc_req_async_args(new_req);
1710
1711         CFS_INIT_LIST_HEAD(&new_aa->aa_oaps);
1712         list_splice(&aa->aa_oaps, &new_aa->aa_oaps);
1713         CFS_INIT_LIST_HEAD(&aa->aa_oaps);
1714
1715         list_for_each_entry(oap, &new_aa->aa_oaps, oap_rpc_item) {
1716                 if (oap->oap_request) {
1717                         ptlrpc_req_finished(oap->oap_request);
1718                         oap->oap_request = ptlrpc_request_addref(new_req);
1719                 }
1720         }
1721
1722         new_aa->aa_ocapa = aa->aa_ocapa;
1723         aa->aa_ocapa = NULL;
1724
1725         /* use ptlrpc_set_add_req is safe because interpret functions work
1726          * in check_set context. only one way exist with access to request
1727          * from different thread got -EINTR - this way protected with
1728          * cl_loi_list_lock */
1729         ptlrpc_set_add_req(set, new_req);
1730
1731         client_obd_list_unlock(&aa->aa_cli->cl_loi_list_lock);
1732
1733         DEBUG_REQ(D_INFO, new_req, "new request");
1734         RETURN(0);
1735 }
1736
1737 /*
1738  * ugh, we want disk allocation on the target to happen in offset order.  we'll
1739  * follow sedgewicks advice and stick to the dead simple shellsort -- it'll do
1740  * fine for our small page arrays and doesn't require allocation.  its an
1741  * insertion sort that swaps elements that are strides apart, shrinking the
1742  * stride down until its '1' and the array is sorted.
1743  */
1744 static void sort_brw_pages(struct brw_page **array, int num)
1745 {
1746         int stride, i, j;
1747         struct brw_page *tmp;
1748
1749         if (num == 1)
1750                 return;
1751         for (stride = 1; stride < num ; stride = (stride * 3) + 1)
1752                 ;
1753
1754         do {
1755                 stride /= 3;
1756                 for (i = stride ; i < num ; i++) {
1757                         tmp = array[i];
1758                         j = i;
1759                         while (j >= stride && array[j - stride]->off > tmp->off) {
1760                                 array[j] = array[j - stride];
1761                                 j -= stride;
1762                         }
1763                         array[j] = tmp;
1764                 }
1765         } while (stride > 1);
1766 }
1767
1768 static obd_count max_unfragmented_pages(struct brw_page **pg, obd_count pages)
1769 {
1770         int count = 1;
1771         int offset;
1772         int i = 0;
1773
1774         LASSERT (pages > 0);
1775         offset = pg[i]->off & ~CFS_PAGE_MASK;
1776
1777         for (;;) {
1778                 pages--;
1779                 if (pages == 0)         /* that's all */
1780                         return count;
1781
1782                 if (offset + pg[i]->count < CFS_PAGE_SIZE)
1783                         return count;   /* doesn't end on page boundary */
1784
1785                 i++;
1786                 offset = pg[i]->off & ~CFS_PAGE_MASK;
1787                 if (offset != 0)        /* doesn't start on page boundary */
1788                         return count;
1789
1790                 count++;
1791         }
1792 }
1793
1794 static struct brw_page **osc_build_ppga(struct brw_page *pga, obd_count count)
1795 {
1796         struct brw_page **ppga;
1797         int i;
1798
1799         OBD_ALLOC(ppga, sizeof(*ppga) * count);
1800         if (ppga == NULL)
1801                 return NULL;
1802
1803         for (i = 0; i < count; i++)
1804                 ppga[i] = pga + i;
1805         return ppga;
1806 }
1807
1808 static void osc_release_ppga(struct brw_page **ppga, obd_count count)
1809 {
1810         LASSERT(ppga != NULL);
1811         OBD_FREE(ppga, sizeof(*ppga) * count);
1812 }
1813
1814 static int osc_brw(int cmd, struct obd_export *exp, struct obd_info *oinfo,
1815                    obd_count page_count, struct brw_page *pga,
1816                    struct obd_trans_info *oti)
1817 {
1818         struct obdo *saved_oa = NULL;
1819         struct brw_page **ppga, **orig;
1820         struct obd_import *imp = class_exp2cliimp(exp);
1821         struct client_obd *cli;
1822         int rc, page_count_orig;
1823         ENTRY;
1824
1825         LASSERT((imp != NULL) && (imp->imp_obd != NULL));
1826         cli = &imp->imp_obd->u.cli;
1827
1828         if (cmd & OBD_BRW_CHECK) {
1829                 /* The caller just wants to know if there's a chance that this
1830                  * I/O can succeed */
1831
1832                 if (imp->imp_invalid)
1833                         RETURN(-EIO);
1834                 RETURN(0);
1835         }
1836
1837         /* test_brw with a failed create can trip this, maybe others. */
1838         LASSERT(cli->cl_max_pages_per_rpc);
1839
1840         rc = 0;
1841
1842         orig = ppga = osc_build_ppga(pga, page_count);
1843         if (ppga == NULL)
1844                 RETURN(-ENOMEM);
1845         page_count_orig = page_count;
1846
1847         sort_brw_pages(ppga, page_count);
1848         while (page_count) {
1849                 obd_count pages_per_brw;
1850
1851                 if (page_count > cli->cl_max_pages_per_rpc)
1852                         pages_per_brw = cli->cl_max_pages_per_rpc;
1853                 else
1854                         pages_per_brw = page_count;
1855
1856                 pages_per_brw = max_unfragmented_pages(ppga, pages_per_brw);
1857
1858                 if (saved_oa != NULL) {
1859                         /* restore previously saved oa */
1860                         *oinfo->oi_oa = *saved_oa;
1861                 } else if (page_count > pages_per_brw) {
1862                         /* save a copy of oa (brw will clobber it) */
1863                         OBDO_ALLOC(saved_oa);
1864                         if (saved_oa == NULL)
1865                                 GOTO(out, rc = -ENOMEM);
1866                         *saved_oa = *oinfo->oi_oa;
1867                 }
1868
1869                 rc = osc_brw_internal(cmd, exp, oinfo->oi_oa, oinfo->oi_md,
1870                                       pages_per_brw, ppga, oinfo->oi_capa);
1871
1872                 if (rc != 0)
1873                         break;
1874
1875                 page_count -= pages_per_brw;
1876                 ppga += pages_per_brw;
1877         }
1878
1879 out:
1880         osc_release_ppga(orig, page_count_orig);
1881
1882         if (saved_oa != NULL)
1883                 OBDO_FREE(saved_oa);
1884
1885         RETURN(rc);
1886 }
1887
1888 /* The companion to osc_enter_cache(), called when @oap is no longer part of
1889  * the dirty accounting.  Writeback completes or truncate happens before
1890  * writing starts.  Must be called with the loi lock held. */
1891 static void osc_exit_cache(struct client_obd *cli, struct osc_async_page *oap,
1892                            int sent)
1893 {
1894         osc_release_write_grant(cli, &oap->oap_brw_page, sent);
1895 }
1896
1897
1898 /* This maintains the lists of pending pages to read/write for a given object
1899  * (lop).  This is used by osc_check_rpcs->osc_next_loi() and loi_list_maint()
1900  * to quickly find objects that are ready to send an RPC. */
1901 static int lop_makes_rpc(struct client_obd *cli, struct loi_oap_pages *lop,
1902                          int cmd)
1903 {
1904         int optimal;
1905         ENTRY;
1906
1907         if (lop->lop_num_pending == 0)
1908                 RETURN(0);
1909
1910         /* if we have an invalid import we want to drain the queued pages
1911          * by forcing them through rpcs that immediately fail and complete
1912          * the pages.  recovery relies on this to empty the queued pages
1913          * before canceling the locks and evicting down the llite pages */
1914         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
1915                 RETURN(1);
1916
1917         /* stream rpcs in queue order as long as as there is an urgent page
1918          * queued.  this is our cheap solution for good batching in the case
1919          * where writepage marks some random page in the middle of the file
1920          * as urgent because of, say, memory pressure */
1921         if (!list_empty(&lop->lop_urgent)) {
1922                 CDEBUG(D_CACHE, "urgent request forcing RPC\n");
1923                 RETURN(1);
1924         }
1925         /* fire off rpcs when we have 'optimal' rpcs as tuned for the wire. */
1926         optimal = cli->cl_max_pages_per_rpc;
1927         if (cmd & OBD_BRW_WRITE) {
1928                 /* trigger a write rpc stream as long as there are dirtiers
1929                  * waiting for space.  as they're waiting, they're not going to
1930                  * create more pages to coallesce with what's waiting.. */
1931                 if (!list_empty(&cli->cl_cache_waiters)) {
1932                         CDEBUG(D_CACHE, "cache waiters forcing RPC\n");
1933                         RETURN(1);
1934                 }
1935                 /* +16 to avoid triggering rpcs that would want to include pages
1936                  * that are being queued but which can't be made ready until
1937                  * the queuer finishes with the page. this is a wart for
1938                  * llite::commit_write() */
1939                 optimal += 16;
1940         }
1941         if (lop->lop_num_pending >= optimal)
1942                 RETURN(1);
1943
1944         RETURN(0);
1945 }
1946
1947 static int lop_makes_hprpc(struct loi_oap_pages *lop)
1948 {
1949         struct osc_async_page *oap;
1950         ENTRY;
1951
1952         if (list_empty(&lop->lop_urgent))
1953                 RETURN(0);
1954
1955         oap = list_entry(lop->lop_urgent.next,
1956                          struct osc_async_page, oap_urgent_item);
1957
1958         if (oap->oap_async_flags & ASYNC_HP) {
1959                 CDEBUG(D_CACHE, "hp request forcing RPC\n");
1960                 RETURN(1);
1961         }
1962
1963         RETURN(0);
1964 }
1965
1966 static void on_list(struct list_head *item, struct list_head *list,
1967                     int should_be_on)
1968 {
1969         if (list_empty(item) && should_be_on)
1970                 list_add_tail(item, list);
1971         else if (!list_empty(item) && !should_be_on)
1972                 list_del_init(item);
1973 }
1974
1975 /* maintain the loi's cli list membership invariants so that osc_send_oap_rpc
1976  * can find pages to build into rpcs quickly */
1977 void loi_list_maint(struct client_obd *cli, struct lov_oinfo *loi)
1978 {
1979         if (lop_makes_hprpc(&loi->loi_write_lop) ||
1980             lop_makes_hprpc(&loi->loi_read_lop)) {
1981                 /* HP rpc */
1982                 on_list(&loi->loi_ready_item, &cli->cl_loi_ready_list, 0);
1983                 on_list(&loi->loi_hp_ready_item, &cli->cl_loi_hp_ready_list, 1);
1984         } else {
1985                 on_list(&loi->loi_hp_ready_item, &cli->cl_loi_hp_ready_list, 0);
1986                 on_list(&loi->loi_ready_item, &cli->cl_loi_ready_list,
1987                         lop_makes_rpc(cli, &loi->loi_write_lop, OBD_BRW_WRITE)||
1988                         lop_makes_rpc(cli, &loi->loi_read_lop, OBD_BRW_READ));
1989         }
1990
1991         on_list(&loi->loi_write_item, &cli->cl_loi_write_list,
1992                 loi->loi_write_lop.lop_num_pending);
1993
1994         on_list(&loi->loi_read_item, &cli->cl_loi_read_list,
1995                 loi->loi_read_lop.lop_num_pending);
1996 }
1997
1998 static void lop_update_pending(struct client_obd *cli,
1999                                struct loi_oap_pages *lop, int cmd, int delta)
2000 {
2001         lop->lop_num_pending += delta;
2002         if (cmd & OBD_BRW_WRITE)
2003                 cli->cl_pending_w_pages += delta;
2004         else
2005                 cli->cl_pending_r_pages += delta;
2006 }
2007
2008 /**
2009  * this is called when a sync waiter receives an interruption.  Its job is to
2010  * get the caller woken as soon as possible.  If its page hasn't been put in an
2011  * rpc yet it can dequeue immediately.  Otherwise it has to mark the rpc as
2012  * desiring interruption which will forcefully complete the rpc once the rpc
2013  * has timed out.
2014  */
2015 int osc_oap_interrupted(const struct lu_env *env, struct osc_async_page *oap)
2016 {
2017         struct loi_oap_pages *lop;
2018         struct lov_oinfo *loi;
2019         int rc = -EBUSY;
2020         ENTRY;
2021
2022         LASSERT(!oap->oap_interrupted);
2023         oap->oap_interrupted = 1;
2024
2025         /* ok, it's been put in an rpc. only one oap gets a request reference */
2026         if (oap->oap_request != NULL) {
2027                 ptlrpc_mark_interrupted(oap->oap_request);
2028                 ptlrpcd_wake(oap->oap_request);
2029                 ptlrpc_req_finished(oap->oap_request);
2030                 oap->oap_request = NULL;
2031         }
2032
2033         /*
2034          * page completion may be called only if ->cpo_prep() method was
2035          * executed by osc_io_submit(), that also adds page the to pending list
2036          */
2037         if (!list_empty(&oap->oap_pending_item)) {
2038                 list_del_init(&oap->oap_pending_item);
2039                 list_del_init(&oap->oap_urgent_item);
2040
2041                 loi = oap->oap_loi;
2042                 lop = (oap->oap_cmd & OBD_BRW_WRITE) ?
2043                         &loi->loi_write_lop : &loi->loi_read_lop;
2044                 lop_update_pending(oap->oap_cli, lop, oap->oap_cmd, -1);
2045                 loi_list_maint(oap->oap_cli, oap->oap_loi);
2046                 rc = oap->oap_caller_ops->ap_completion(env,
2047                                           oap->oap_caller_data,
2048                                           oap->oap_cmd, NULL, -EINTR);
2049         }
2050
2051         RETURN(rc);
2052 }
2053
2054 /* this is trying to propogate async writeback errors back up to the
2055  * application.  As an async write fails we record the error code for later if
2056  * the app does an fsync.  As long as errors persist we force future rpcs to be
2057  * sync so that the app can get a sync error and break the cycle of queueing
2058  * pages for which writeback will fail. */
2059 static void osc_process_ar(struct osc_async_rc *ar, __u64 xid,
2060                            int rc)
2061 {
2062         if (rc) {
2063                 if (!ar->ar_rc)
2064                         ar->ar_rc = rc;
2065
2066                 ar->ar_force_sync = 1;
2067                 ar->ar_min_xid = ptlrpc_sample_next_xid();
2068                 return;
2069
2070         }
2071
2072         if (ar->ar_force_sync && (xid >= ar->ar_min_xid))
2073                 ar->ar_force_sync = 0;
2074 }
2075
2076 void osc_oap_to_pending(struct osc_async_page *oap)
2077 {
2078         struct loi_oap_pages *lop;
2079
2080         if (oap->oap_cmd & OBD_BRW_WRITE)
2081                 lop = &oap->oap_loi->loi_write_lop;
2082         else
2083                 lop = &oap->oap_loi->loi_read_lop;
2084
2085         if (oap->oap_async_flags & ASYNC_HP)
2086                 list_add(&oap->oap_urgent_item, &lop->lop_urgent);
2087         else if (oap->oap_async_flags & ASYNC_URGENT)
2088                 list_add_tail(&oap->oap_urgent_item, &lop->lop_urgent);
2089         list_add_tail(&oap->oap_pending_item, &lop->lop_pending);
2090         lop_update_pending(oap->oap_cli, lop, oap->oap_cmd, 1);
2091 }
2092
2093 /* this must be called holding the loi list lock to give coverage to exit_cache,
2094  * async_flag maintenance, and oap_request */
2095 static void osc_ap_completion(const struct lu_env *env,
2096                               struct client_obd *cli, struct obdo *oa,
2097                               struct osc_async_page *oap, int sent, int rc)
2098 {
2099         __u64 xid = 0;
2100
2101         ENTRY;
2102         if (oap->oap_request != NULL) {
2103                 xid = ptlrpc_req_xid(oap->oap_request);
2104                 ptlrpc_req_finished(oap->oap_request);
2105                 oap->oap_request = NULL;
2106         }
2107
2108         oap->oap_async_flags = 0;
2109         oap->oap_interrupted = 0;
2110
2111         if (oap->oap_cmd & OBD_BRW_WRITE) {
2112                 osc_process_ar(&cli->cl_ar, xid, rc);
2113                 osc_process_ar(&oap->oap_loi->loi_ar, xid, rc);
2114         }
2115
2116         if (rc == 0 && oa != NULL) {
2117                 if (oa->o_valid & OBD_MD_FLBLOCKS)
2118                         oap->oap_loi->loi_lvb.lvb_blocks = oa->o_blocks;
2119                 if (oa->o_valid & OBD_MD_FLMTIME)
2120                         oap->oap_loi->loi_lvb.lvb_mtime = oa->o_mtime;
2121                 if (oa->o_valid & OBD_MD_FLATIME)
2122                         oap->oap_loi->loi_lvb.lvb_atime = oa->o_atime;
2123                 if (oa->o_valid & OBD_MD_FLCTIME)
2124                         oap->oap_loi->loi_lvb.lvb_ctime = oa->o_ctime;
2125         }
2126
2127         rc = oap->oap_caller_ops->ap_completion(env, oap->oap_caller_data,
2128                                                 oap->oap_cmd, oa, rc);
2129
2130         /* ll_ap_completion (from llite) drops PG_locked. so, a new
2131          * I/O on the page could start, but OSC calls it under lock
2132          * and thus we can add oap back to pending safely */
2133         if (rc)
2134                 /* upper layer wants to leave the page on pending queue */
2135                 osc_oap_to_pending(oap);
2136         else
2137                 osc_exit_cache(cli, oap, sent);
2138         EXIT;
2139 }
2140
2141 static int brw_interpret(const struct lu_env *env,
2142                          struct ptlrpc_request *req, void *data, int rc)
2143 {
2144         struct osc_brw_async_args *aa = data;
2145         struct client_obd *cli;
2146         int async;
2147         ENTRY;
2148
2149         rc = osc_brw_fini_request(req, rc);
2150         CDEBUG(D_INODE, "request %p aa %p rc %d\n", req, aa, rc);
2151         if (osc_recoverable_error(rc)) {
2152                 rc = osc_brw_redo_request(req, aa);
2153                 if (rc == 0)
2154                         RETURN(0);
2155         }
2156
2157         if (aa->aa_ocapa) {
2158                 capa_put(aa->aa_ocapa);
2159                 aa->aa_ocapa = NULL;
2160         }
2161
2162         cli = aa->aa_cli;
2163
2164         client_obd_list_lock(&cli->cl_loi_list_lock);
2165
2166         /* We need to decrement before osc_ap_completion->osc_wake_cache_waiters
2167          * is called so we know whether to go to sync BRWs or wait for more
2168          * RPCs to complete */
2169         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE)
2170                 cli->cl_w_in_flight--;
2171         else
2172                 cli->cl_r_in_flight--;
2173
2174         async = list_empty(&aa->aa_oaps);
2175         if (!async) { /* from osc_send_oap_rpc() */
2176                 struct osc_async_page *oap, *tmp;
2177                 /* the caller may re-use the oap after the completion call so
2178                  * we need to clean it up a little */
2179                 list_for_each_entry_safe(oap, tmp, &aa->aa_oaps, oap_rpc_item) {
2180                         list_del_init(&oap->oap_rpc_item);
2181                         osc_ap_completion(env, cli, aa->aa_oa, oap, 1, rc);
2182                 }
2183                 OBDO_FREE(aa->aa_oa);
2184         } else { /* from async_internal() */
2185                 int i;
2186                 for (i = 0; i < aa->aa_page_count; i++)
2187                         osc_release_write_grant(aa->aa_cli, aa->aa_ppga[i], 1);
2188         }
2189         osc_wake_cache_waiters(cli);
2190         osc_check_rpcs(env, cli);
2191         client_obd_list_unlock(&cli->cl_loi_list_lock);
2192         if (!async)
2193                 cl_req_completion(env, aa->aa_clerq, rc);
2194         osc_release_ppga(aa->aa_ppga, aa->aa_page_count);
2195         RETURN(rc);
2196 }
2197
2198 static struct ptlrpc_request *osc_build_req(const struct lu_env *env,
2199                                             struct client_obd *cli,
2200                                             struct list_head *rpc_list,
2201                                             int page_count, int cmd)
2202 {
2203         struct ptlrpc_request *req;
2204         struct brw_page **pga = NULL;
2205         struct osc_brw_async_args *aa;
2206         struct obdo *oa = NULL;
2207         const struct obd_async_page_ops *ops = NULL;
2208         void *caller_data = NULL;
2209         struct osc_async_page *oap;
2210         struct osc_async_page *tmp;
2211         struct ost_body *body;
2212         struct cl_req *clerq = NULL;
2213         enum cl_req_type crt = (cmd & OBD_BRW_WRITE) ? CRT_WRITE : CRT_READ;
2214         struct ldlm_lock *lock = NULL;
2215         struct cl_req_attr crattr;
2216         int i, rc;
2217
2218         ENTRY;
2219         LASSERT(!list_empty(rpc_list));
2220
2221         memset(&crattr, 0, sizeof crattr);
2222         OBD_ALLOC(pga, sizeof(*pga) * page_count);
2223         if (pga == NULL)
2224                 GOTO(out, req = ERR_PTR(-ENOMEM));
2225
2226         OBDO_ALLOC(oa);
2227         if (oa == NULL)
2228                 GOTO(out, req = ERR_PTR(-ENOMEM));
2229
2230         i = 0;
2231         list_for_each_entry(oap, rpc_list, oap_rpc_item) {
2232                 struct cl_page *page = osc_oap2cl_page(oap);
2233                 if (ops == NULL) {
2234                         ops = oap->oap_caller_ops;
2235                         caller_data = oap->oap_caller_data;
2236
2237                         clerq = cl_req_alloc(env, page, crt,
2238                                              1 /* only 1-object rpcs for
2239                                                 * now */);
2240                         if (IS_ERR(clerq))
2241                                 GOTO(out, req = (void *)clerq);
2242                         lock = oap->oap_ldlm_lock;
2243                 }
2244                 pga[i] = &oap->oap_brw_page;
2245                 pga[i]->off = oap->oap_obj_off + oap->oap_page_off;
2246                 CDEBUG(0, "put page %p index %lu oap %p flg %x to pga\n",
2247                        pga[i]->pg, cfs_page_index(oap->oap_page), oap, pga[i]->flag);
2248                 i++;
2249                 cl_req_page_add(env, clerq, page);
2250         }
2251
2252         /* always get the data for the obdo for the rpc */
2253         LASSERT(ops != NULL);
2254         crattr.cra_oa = oa;
2255         crattr.cra_capa = NULL;
2256         cl_req_attr_set(env, clerq, &crattr, ~0ULL);
2257         if (lock) {
2258                 oa->o_handle = lock->l_remote_handle;
2259                 oa->o_valid |= OBD_MD_FLHANDLE;
2260         }
2261
2262         rc = cl_req_prep(env, clerq);
2263         if (rc != 0) {
2264                 CERROR("cl_req_prep failed: %d\n", rc);
2265                 GOTO(out, req = ERR_PTR(rc));
2266         }
2267
2268         sort_brw_pages(pga, page_count);
2269         rc = osc_brw_prep_request(cmd, cli, oa, NULL, page_count,
2270                                   pga, &req, crattr.cra_capa, 1);
2271         if (rc != 0) {
2272                 CERROR("prep_req failed: %d\n", rc);
2273                 GOTO(out, req = ERR_PTR(rc));
2274         }
2275
2276         /* Need to update the timestamps after the request is built in case
2277          * we race with setattr (locally or in queue at OST).  If OST gets
2278          * later setattr before earlier BRW (as determined by the request xid),
2279          * the OST will not use BRW timestamps.  Sadly, there is no obvious
2280          * way to do this in a single call.  bug 10150 */
2281         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
2282         cl_req_attr_set(env, clerq, &crattr,
2283                         OBD_MD_FLMTIME|OBD_MD_FLCTIME|OBD_MD_FLATIME);
2284
2285         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
2286         aa = ptlrpc_req_async_args(req);
2287         CFS_INIT_LIST_HEAD(&aa->aa_oaps);
2288         list_splice(rpc_list, &aa->aa_oaps);
2289         CFS_INIT_LIST_HEAD(rpc_list);
2290         aa->aa_clerq = clerq;
2291 out:
2292         capa_put(crattr.cra_capa);
2293         if (IS_ERR(req)) {
2294                 if (oa)
2295                         OBDO_FREE(oa);
2296                 if (pga)
2297                         OBD_FREE(pga, sizeof(*pga) * page_count);
2298                 /* this should happen rarely and is pretty bad, it makes the
2299                  * pending list not follow the dirty order */
2300                 client_obd_list_lock(&cli->cl_loi_list_lock);
2301                 list_for_each_entry_safe(oap, tmp, rpc_list, oap_rpc_item) {
2302                         list_del_init(&oap->oap_rpc_item);
2303
2304                         /* queued sync pages can be torn down while the pages
2305                          * were between the pending list and the rpc */
2306                         if (oap->oap_interrupted) {
2307                                 CDEBUG(D_INODE, "oap %p interrupted\n", oap);
2308                                 osc_ap_completion(env, cli, NULL, oap, 0,
2309                                                   oap->oap_count);
2310                                 continue;
2311                         }
2312                         osc_ap_completion(env, cli, NULL, oap, 0, PTR_ERR(req));
2313                 }
2314                 if (clerq && !IS_ERR(clerq))
2315                         cl_req_completion(env, clerq, PTR_ERR(req));
2316         }
2317         RETURN(req);
2318 }
2319
2320 /**
2321  * prepare pages for ASYNC io and put pages in send queue.
2322  *
2323  * \param cli -
2324  * \param loi -
2325  * \param cmd - OBD_BRW_* macroses
2326  * \param lop - pending pages
2327  *
2328  * \return zero if pages successfully add to send queue.
2329  * \return not zere if error occurring.
2330  */
2331 static int
2332 osc_send_oap_rpc(const struct lu_env *env, struct client_obd *cli,
2333                  struct lov_oinfo *loi,
2334                  int cmd, struct loi_oap_pages *lop)
2335 {
2336         struct ptlrpc_request *req;
2337         obd_count page_count = 0;
2338         struct osc_async_page *oap = NULL, *tmp;
2339         struct osc_brw_async_args *aa;
2340         const struct obd_async_page_ops *ops;
2341         CFS_LIST_HEAD(rpc_list);
2342         unsigned int ending_offset;
2343         unsigned  starting_offset = 0;
2344         int srvlock = 0;
2345         struct cl_object *clob = NULL;
2346         ENTRY;
2347
2348         /* If there are HP OAPs we need to handle at least 1 of them,
2349          * move it the beginning of the pending list for that. */
2350         if (!list_empty(&lop->lop_urgent)) {
2351                 oap = list_entry(lop->lop_urgent.next,
2352                                  struct osc_async_page, oap_urgent_item);
2353                 if (oap->oap_async_flags & ASYNC_HP)
2354                         list_move(&oap->oap_pending_item, &lop->lop_pending);
2355         }
2356
2357         /* first we find the pages we're allowed to work with */
2358         list_for_each_entry_safe(oap, tmp, &lop->lop_pending,
2359                                  oap_pending_item) {
2360                 ops = oap->oap_caller_ops;
2361
2362                 LASSERTF(oap->oap_magic == OAP_MAGIC, "Bad oap magic: oap %p, "
2363                          "magic 0x%x\n", oap, oap->oap_magic);
2364
2365                 if (clob == NULL) {
2366                         /* pin object in memory, so that completion call-backs
2367                          * can be safely called under client_obd_list lock. */
2368                         clob = osc_oap2cl_page(oap)->cp_obj;
2369                         cl_object_get(clob);
2370                 }
2371
2372                 if (page_count != 0 &&
2373                     srvlock != !!(oap->oap_brw_flags & OBD_BRW_SRVLOCK)) {
2374                         CDEBUG(D_PAGE, "SRVLOCK flag mismatch,"
2375                                " oap %p, page %p, srvlock %u\n",
2376                                oap, oap->oap_brw_page.pg, (unsigned)!srvlock);
2377                         break;
2378                 }
2379                 /* in llite being 'ready' equates to the page being locked
2380                  * until completion unlocks it.  commit_write submits a page
2381                  * as not ready because its unlock will happen unconditionally
2382                  * as the call returns.  if we race with commit_write giving
2383                  * us that page we dont' want to create a hole in the page
2384                  * stream, so we stop and leave the rpc to be fired by
2385                  * another dirtier or kupdated interval (the not ready page
2386                  * will still be on the dirty list).  we could call in
2387                  * at the end of ll_file_write to process the queue again. */
2388                 if (!(oap->oap_async_flags & ASYNC_READY)) {
2389                         int rc = ops->ap_make_ready(env, oap->oap_caller_data,
2390                                                     cmd);
2391                         if (rc < 0)
2392                                 CDEBUG(D_INODE, "oap %p page %p returned %d "
2393                                                 "instead of ready\n", oap,
2394                                                 oap->oap_page, rc);
2395                         switch (rc) {
2396                         case -EAGAIN:
2397                                 /* llite is telling us that the page is still
2398                                  * in commit_write and that we should try
2399                                  * and put it in an rpc again later.  we
2400                                  * break out of the loop so we don't create
2401                                  * a hole in the sequence of pages in the rpc
2402                                  * stream.*/
2403                                 oap = NULL;
2404                                 break;
2405                         case -EINTR:
2406                                 /* the io isn't needed.. tell the checks
2407                                  * below to complete the rpc with EINTR */
2408                                 oap->oap_async_flags |= ASYNC_COUNT_STABLE;
2409                                 oap->oap_count = -EINTR;
2410                                 break;
2411                         case 0:
2412                                 oap->oap_async_flags |= ASYNC_READY;
2413                                 break;
2414                         default:
2415                                 LASSERTF(0, "oap %p page %p returned %d "
2416                                             "from make_ready\n", oap,
2417                                             oap->oap_page, rc);
2418                                 break;
2419                         }
2420                 }
2421                 if (oap == NULL)
2422                         break;
2423                 /*
2424                  * Page submitted for IO has to be locked. Either by
2425                  * ->ap_make_ready() or by higher layers.
2426                  */
2427 #if defined(__KERNEL__) && defined(__linux__)
2428                 {
2429                         struct cl_page *page;
2430
2431                         page = osc_oap2cl_page(oap);
2432
2433                         if (page->cp_type == CPT_CACHEABLE &&
2434                             !(PageLocked(oap->oap_page) &&
2435                               (CheckWriteback(oap->oap_page, cmd)))) {
2436                                 CDEBUG(D_PAGE, "page %p lost wb %lx/%x\n",
2437                                        oap->oap_page,
2438                                        (long)oap->oap_page->flags,
2439                                        oap->oap_async_flags);
2440                                 LBUG();
2441                         }
2442                 }
2443 #endif
2444                 /* If there is a gap at the start of this page, it can't merge
2445                  * with any previous page, so we'll hand the network a
2446                  * "fragmented" page array that it can't transfer in 1 RDMA */
2447                 if (page_count != 0 && oap->oap_page_off != 0)
2448                         break;
2449
2450                 /* take the page out of our book-keeping */
2451                 list_del_init(&oap->oap_pending_item);
2452                 lop_update_pending(cli, lop, cmd, -1);
2453                 list_del_init(&oap->oap_urgent_item);
2454
2455                 if (page_count == 0)
2456                         starting_offset = (oap->oap_obj_off+oap->oap_page_off) &
2457                                           (PTLRPC_MAX_BRW_SIZE - 1);
2458
2459                 /* ask the caller for the size of the io as the rpc leaves. */
2460                 if (!(oap->oap_async_flags & ASYNC_COUNT_STABLE)) {
2461                         oap->oap_count =
2462                                 ops->ap_refresh_count(env, oap->oap_caller_data,
2463                                                       cmd);
2464                         LASSERT(oap->oap_page_off + oap->oap_count <= CFS_PAGE_SIZE);
2465                 }
2466                 if (oap->oap_count <= 0) {
2467                         CDEBUG(D_CACHE, "oap %p count %d, completing\n", oap,
2468                                oap->oap_count);
2469                         osc_ap_completion(env, cli, NULL,
2470                                           oap, 0, oap->oap_count);
2471                         continue;
2472                 }
2473
2474                 /* now put the page back in our accounting */
2475                 list_add_tail(&oap->oap_rpc_item, &rpc_list);
2476                 if (page_count == 0)
2477                         srvlock = !!(oap->oap_brw_flags & OBD_BRW_SRVLOCK);
2478                 if (++page_count >= cli->cl_max_pages_per_rpc)
2479                         break;
2480
2481                 /* End on a PTLRPC_MAX_BRW_SIZE boundary.  We want full-sized
2482                  * RPCs aligned on PTLRPC_MAX_BRW_SIZE boundaries to help reads
2483                  * have the same alignment as the initial writes that allocated
2484                  * extents on the server. */
2485                 ending_offset = (oap->oap_obj_off + oap->oap_page_off +
2486                                  oap->oap_count) & (PTLRPC_MAX_BRW_SIZE - 1);
2487                 if (ending_offset == 0)
2488                         break;
2489
2490                 /* If there is a gap at the end of this page, it can't merge
2491                  * with any subsequent pages, so we'll hand the network a
2492                  * "fragmented" page array that it can't transfer in 1 RDMA */
2493                 if (oap->oap_page_off + oap->oap_count < CFS_PAGE_SIZE)
2494                         break;
2495         }
2496
2497         osc_wake_cache_waiters(cli);
2498
2499         loi_list_maint(cli, loi);
2500
2501         client_obd_list_unlock(&cli->cl_loi_list_lock);
2502
2503         if (clob != NULL)
2504                 cl_object_put(env, clob);
2505
2506         if (page_count == 0) {
2507                 client_obd_list_lock(&cli->cl_loi_list_lock);
2508                 RETURN(0);
2509         }
2510
2511         req = osc_build_req(env, cli, &rpc_list, page_count, cmd);
2512         if (IS_ERR(req)) {
2513                 LASSERT(list_empty(&rpc_list));
2514                 loi_list_maint(cli, loi);
2515                 RETURN(PTR_ERR(req));
2516         }
2517
2518         aa = ptlrpc_req_async_args(req);
2519
2520         if (cmd == OBD_BRW_READ) {
2521                 lprocfs_oh_tally_log2(&cli->cl_read_page_hist, page_count);
2522                 lprocfs_oh_tally(&cli->cl_read_rpc_hist, cli->cl_r_in_flight);
2523                 lprocfs_oh_tally_log2(&cli->cl_read_offset_hist,
2524                                       (starting_offset >> CFS_PAGE_SHIFT) + 1);
2525         } else {
2526                 lprocfs_oh_tally_log2(&cli->cl_write_page_hist, page_count);
2527                 lprocfs_oh_tally(&cli->cl_write_rpc_hist,
2528                                  cli->cl_w_in_flight);
2529                 lprocfs_oh_tally_log2(&cli->cl_write_offset_hist,
2530                                       (starting_offset >> CFS_PAGE_SHIFT) + 1);
2531         }
2532         ptlrpc_lprocfs_brw(req, aa->aa_requested_nob);
2533
2534         client_obd_list_lock(&cli->cl_loi_list_lock);
2535
2536         if (cmd == OBD_BRW_READ)
2537                 cli->cl_r_in_flight++;
2538         else
2539                 cli->cl_w_in_flight++;
2540
2541         /* queued sync pages can be torn down while the pages
2542          * were between the pending list and the rpc */
2543         tmp = NULL;
2544         list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
2545                 /* only one oap gets a request reference */
2546                 if (tmp == NULL)
2547                         tmp = oap;
2548                 if (oap->oap_interrupted && !req->rq_intr) {
2549                         CDEBUG(D_INODE, "oap %p in req %p interrupted\n",
2550                                oap, req);
2551                         ptlrpc_mark_interrupted(req);
2552                 }
2553         }
2554         if (tmp != NULL)
2555                 tmp->oap_request = ptlrpc_request_addref(req);
2556
2557         DEBUG_REQ(D_INODE, req, "%d pages, aa %p. now %dr/%dw in flight",
2558                   page_count, aa, cli->cl_r_in_flight, cli->cl_w_in_flight);
2559
2560         req->rq_interpret_reply = brw_interpret;
2561         ptlrpcd_add_req(req, PSCOPE_BRW);
2562         RETURN(1);
2563 }
2564
2565 #define LOI_DEBUG(LOI, STR, args...)                                     \
2566         CDEBUG(D_INODE, "loi ready %d wr %d:%d rd %d:%d " STR,           \
2567                !list_empty(&(LOI)->loi_ready_item) ||                    \
2568                !list_empty(&(LOI)->loi_hp_ready_item),                   \
2569                (LOI)->loi_write_lop.lop_num_pending,                     \
2570                !list_empty(&(LOI)->loi_write_lop.lop_urgent),            \
2571                (LOI)->loi_read_lop.lop_num_pending,                      \
2572                !list_empty(&(LOI)->loi_read_lop.lop_urgent),             \
2573                args)                                                     \
2574
2575 /* This is called by osc_check_rpcs() to find which objects have pages that
2576  * we could be sending.  These lists are maintained by lop_makes_rpc(). */
2577 struct lov_oinfo *osc_next_loi(struct client_obd *cli)
2578 {
2579         ENTRY;
2580
2581         /* First return objects that have blocked locks so that they
2582          * will be flushed quickly and other clients can get the lock,
2583          * then objects which have pages ready to be stuffed into RPCs */
2584         if (!list_empty(&cli->cl_loi_hp_ready_list))
2585                 RETURN(list_entry(cli->cl_loi_hp_ready_list.next,
2586                                   struct lov_oinfo, loi_hp_ready_item));
2587         if (!list_empty(&cli->cl_loi_ready_list))
2588                 RETURN(list_entry(cli->cl_loi_ready_list.next,
2589                                   struct lov_oinfo, loi_ready_item));
2590
2591         /* then if we have cache waiters, return all objects with queued
2592          * writes.  This is especially important when many small files
2593          * have filled up the cache and not been fired into rpcs because
2594          * they don't pass the nr_pending/object threshhold */
2595         if (!list_empty(&cli->cl_cache_waiters) &&
2596             !list_empty(&cli->cl_loi_write_list))
2597                 RETURN(list_entry(cli->cl_loi_write_list.next,
2598                                   struct lov_oinfo, loi_write_item));
2599
2600         /* then return all queued objects when we have an invalid import
2601          * so that they get flushed */
2602         if (cli->cl_import == NULL || cli->cl_import->imp_invalid) {
2603                 if (!list_empty(&cli->cl_loi_write_list))
2604                         RETURN(list_entry(cli->cl_loi_write_list.next,
2605                                           struct lov_oinfo, loi_write_item));
2606                 if (!list_empty(&cli->cl_loi_read_list))
2607                         RETURN(list_entry(cli->cl_loi_read_list.next,
2608                                           struct lov_oinfo, loi_read_item));
2609         }
2610         RETURN(NULL);
2611 }
2612
2613 static int osc_max_rpc_in_flight(struct client_obd *cli, struct lov_oinfo *loi)
2614 {
2615         struct osc_async_page *oap;
2616         int hprpc = 0;
2617
2618         if (!list_empty(&loi->loi_write_lop.lop_urgent)) {
2619                 oap = list_entry(loi->loi_write_lop.lop_urgent.next,
2620                                  struct osc_async_page, oap_urgent_item);
2621                 hprpc = !!(oap->oap_async_flags & ASYNC_HP);
2622         }
2623
2624         if (!hprpc && !list_empty(&loi->loi_read_lop.lop_urgent)) {
2625                 oap = list_entry(loi->loi_write_lop.lop_urgent.next,
2626                                  struct osc_async_page, oap_urgent_item);
2627                 hprpc = !!(oap->oap_async_flags & ASYNC_HP);
2628         }
2629
2630         return rpcs_in_flight(cli) >= cli->cl_max_rpcs_in_flight + hprpc;
2631 }
2632
2633 /* called with the loi list lock held */
2634 void osc_check_rpcs(const struct lu_env *env, struct client_obd *cli)
2635 {
2636         struct lov_oinfo *loi;
2637         int rc = 0, race_counter = 0;
2638         ENTRY;
2639
2640         while ((loi = osc_next_loi(cli)) != NULL) {
2641                 LOI_DEBUG(loi, "%lu in flight\n", rpcs_in_flight(cli));
2642
2643                 if (osc_max_rpc_in_flight(cli, loi))
2644                         break;
2645
2646                 /* attempt some read/write balancing by alternating between
2647                  * reads and writes in an object.  The makes_rpc checks here
2648                  * would be redundant if we were getting read/write work items
2649                  * instead of objects.  we don't want send_oap_rpc to drain a
2650                  * partial read pending queue when we're given this object to
2651                  * do io on writes while there are cache waiters */
2652                 if (lop_makes_rpc(cli, &loi->loi_write_lop, OBD_BRW_WRITE)) {
2653                         rc = osc_send_oap_rpc(env, cli, loi, OBD_BRW_WRITE,
2654                                               &loi->loi_write_lop);
2655                         if (rc < 0)
2656                                 break;
2657                         if (rc > 0)
2658                                 race_counter = 0;
2659                         else
2660                                 race_counter++;
2661                 }
2662                 if (lop_makes_rpc(cli, &loi->loi_read_lop, OBD_BRW_READ)) {
2663                         rc = osc_send_oap_rpc(env, cli, loi, OBD_BRW_READ,
2664                                               &loi->loi_read_lop);
2665                         if (rc < 0)
2666                                 break;
2667                         if (rc > 0)
2668                                 race_counter = 0;
2669                         else
2670                                 race_counter++;
2671                 }
2672
2673                 /* attempt some inter-object balancing by issueing rpcs
2674                  * for each object in turn */
2675                 if (!list_empty(&loi->loi_hp_ready_item))
2676                         list_del_init(&loi->loi_hp_ready_item);
2677                 if (!list_empty(&loi->loi_ready_item))
2678                         list_del_init(&loi->loi_ready_item);
2679                 if (!list_empty(&loi->loi_write_item))
2680                         list_del_init(&loi->loi_write_item);
2681                 if (!list_empty(&loi->loi_read_item))
2682                         list_del_init(&loi->loi_read_item);
2683
2684                 loi_list_maint(cli, loi);
2685
2686                 /* send_oap_rpc fails with 0 when make_ready tells it to
2687                  * back off.  llite's make_ready does this when it tries
2688                  * to lock a page queued for write that is already locked.
2689                  * we want to try sending rpcs from many objects, but we
2690                  * don't want to spin failing with 0.  */
2691                 if (race_counter == 10)
2692                         break;
2693         }
2694         EXIT;
2695 }
2696
2697 /* we're trying to queue a page in the osc so we're subject to the
2698  * 'cl_dirty_max' limit on the number of pages that can be queued in the osc.
2699  * If the osc's queued pages are already at that limit, then we want to sleep
2700  * until there is space in the osc's queue for us.  We also may be waiting for
2701  * write credits from the OST if there are RPCs in flight that may return some
2702  * before we fall back to sync writes.
2703  *
2704  * We need this know our allocation was granted in the presence of signals */
2705 static int ocw_granted(struct client_obd *cli, struct osc_cache_waiter *ocw)
2706 {
2707         int rc;
2708         ENTRY;
2709         client_obd_list_lock(&cli->cl_loi_list_lock);
2710         rc = list_empty(&ocw->ocw_entry) || rpcs_in_flight(cli) == 0;
2711         client_obd_list_unlock(&cli->cl_loi_list_lock);
2712         RETURN(rc);
2713 };
2714
2715 /**
2716  * Non-blocking version of osc_enter_cache() that consumes grant only when it
2717  * is available.
2718  */
2719 int osc_enter_cache_try(const struct lu_env *env,
2720                         struct client_obd *cli, struct lov_oinfo *loi,
2721                         struct osc_async_page *oap, int transient)
2722 {
2723         int has_grant;
2724
2725         has_grant = cli->cl_avail_grant >= CFS_PAGE_SIZE;
2726         if (has_grant) {
2727                 osc_consume_write_grant(cli, &oap->oap_brw_page);
2728                 if (transient) {
2729                         cli->cl_dirty_transit += CFS_PAGE_SIZE;
2730                         atomic_inc(&obd_dirty_transit_pages);
2731                         oap->oap_brw_flags |= OBD_BRW_NOCACHE;
2732                 }
2733         }
2734         return has_grant;
2735 }
2736
2737 /* Caller must hold loi_list_lock - we drop/regain it if we need to wait for
2738  * grant or cache space. */
2739 static int osc_enter_cache(const struct lu_env *env,
2740                            struct client_obd *cli, struct lov_oinfo *loi,
2741                            struct osc_async_page *oap)
2742 {
2743         struct osc_cache_waiter ocw;
2744         struct l_wait_info lwi = { 0 };
2745
2746         ENTRY;
2747
2748         CDEBUG(D_CACHE, "dirty: %ld/%d dirty_max: %ld/%d dropped: %lu "
2749                "grant: %lu\n", cli->cl_dirty, atomic_read(&obd_dirty_pages),
2750                cli->cl_dirty_max, obd_max_dirty_pages,
2751                cli->cl_lost_grant, cli->cl_avail_grant);
2752
2753         /* force the caller to try sync io.  this can jump the list
2754          * of queued writes and create a discontiguous rpc stream */
2755         if (cli->cl_dirty_max < CFS_PAGE_SIZE || cli->cl_ar.ar_force_sync ||
2756             loi->loi_ar.ar_force_sync)
2757                 RETURN(-EDQUOT);
2758
2759         /* Hopefully normal case - cache space and write credits available */
2760         if (cli->cl_dirty + CFS_PAGE_SIZE <= cli->cl_dirty_max &&
2761             atomic_read(&obd_dirty_pages) + 1 <= obd_max_dirty_pages &&
2762             osc_enter_cache_try(env, cli, loi, oap, 0))
2763                 RETURN(0);
2764
2765         /* Make sure that there are write rpcs in flight to wait for.  This
2766          * is a little silly as this object may not have any pending but
2767          * other objects sure might. */
2768         if (cli->cl_w_in_flight) {
2769                 list_add_tail(&ocw.ocw_entry, &cli->cl_cache_waiters);
2770                 cfs_waitq_init(&ocw.ocw_waitq);
2771                 ocw.ocw_oap = oap;
2772                 ocw.ocw_rc = 0;
2773
2774                 loi_list_maint(cli, loi);
2775                 osc_check_rpcs(env, cli);
2776                 client_obd_list_unlock(&cli->cl_loi_list_lock);
2777
2778                 CDEBUG(D_CACHE, "sleeping for cache space\n");
2779                 l_wait_event(ocw.ocw_waitq, ocw_granted(cli, &ocw), &lwi);
2780
2781                 client_obd_list_lock(&cli->cl_loi_list_lock);
2782                 if (!list_empty(&ocw.ocw_entry)) {
2783                         list_del(&ocw.ocw_entry);
2784                         RETURN(-EINTR);
2785                 }
2786                 RETURN(ocw.ocw_rc);
2787         }
2788
2789         RETURN(-EDQUOT);
2790 }
2791
2792
2793 int osc_prep_async_page(struct obd_export *exp, struct lov_stripe_md *lsm,
2794                         struct lov_oinfo *loi, cfs_page_t *page,
2795                         obd_off offset, const struct obd_async_page_ops *ops,
2796                         void *data, void **res, int nocache,
2797                         struct lustre_handle *lockh)
2798 {
2799         struct osc_async_page *oap;
2800
2801         ENTRY;
2802
2803         if (!page)
2804                 return size_round(sizeof(*oap));
2805
2806         oap = *res;
2807         oap->oap_magic = OAP_MAGIC;
2808         oap->oap_cli = &exp->exp_obd->u.cli;
2809         oap->oap_loi = loi;
2810
2811         oap->oap_caller_ops = ops;
2812         oap->oap_caller_data = data;
2813
2814         oap->oap_page = page;
2815         oap->oap_obj_off = offset;
2816         if (!client_is_remote(exp) &&
2817             cfs_capable(CFS_CAP_SYS_RESOURCE))
2818                 oap->oap_brw_flags = OBD_BRW_NOQUOTA;
2819
2820         LASSERT(!(offset & ~CFS_PAGE_MASK));
2821
2822         CFS_INIT_LIST_HEAD(&oap->oap_pending_item);
2823         CFS_INIT_LIST_HEAD(&oap->oap_urgent_item);
2824         CFS_INIT_LIST_HEAD(&oap->oap_rpc_item);
2825         CFS_INIT_LIST_HEAD(&oap->oap_page_list);
2826
2827         spin_lock_init(&oap->oap_lock);
2828         CDEBUG(D_CACHE, "oap %p page %p obj off "LPU64"\n", oap, page, offset);
2829         RETURN(0);
2830 }
2831
2832 struct osc_async_page *oap_from_cookie(void *cookie)
2833 {
2834         struct osc_async_page *oap = cookie;
2835         if (oap->oap_magic != OAP_MAGIC)
2836                 return ERR_PTR(-EINVAL);
2837         return oap;
2838 };
2839
2840 int osc_queue_async_io(const struct lu_env *env,
2841                        struct obd_export *exp, struct lov_stripe_md *lsm,
2842                        struct lov_oinfo *loi, void *cookie,
2843                        int cmd, obd_off off, int count,
2844                        obd_flag brw_flags, enum async_flags async_flags)
2845 {
2846         struct client_obd *cli = &exp->exp_obd->u.cli;
2847         struct osc_async_page *oap;
2848         int rc = 0;
2849         ENTRY;
2850
2851         oap = oap_from_cookie(cookie);
2852         if (IS_ERR(oap))
2853                 RETURN(PTR_ERR(oap));
2854
2855         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
2856                 RETURN(-EIO);
2857
2858         if (!list_empty(&oap->oap_pending_item) ||
2859             !list_empty(&oap->oap_urgent_item) ||
2860             !list_empty(&oap->oap_rpc_item))
2861                 RETURN(-EBUSY);
2862
2863         /* check if the file's owner/group is over quota */
2864         if ((cmd & OBD_BRW_WRITE) && !(cmd & OBD_BRW_NOQUOTA)) {
2865                 struct cl_object *obj;
2866                 struct cl_attr    attr; /* XXX put attr into thread info */
2867                 unsigned int qid[MAXQUOTAS];
2868
2869                 obj = cl_object_top(osc_oap2cl_page(oap)->cp_obj);
2870
2871                 cl_object_attr_lock(obj);
2872                 rc = cl_object_attr_get(env, obj, &attr);
2873                 cl_object_attr_unlock(obj);
2874
2875                 qid[USRQUOTA] = attr.cat_uid;
2876                 qid[GRPQUOTA] = attr.cat_gid;
2877                 if (rc == 0 &&
2878                     lquota_chkdq(quota_interface, cli, qid) == NO_QUOTA)
2879                         rc = -EDQUOT;
2880                 if (rc)
2881                         RETURN(rc);
2882         }
2883
2884         if (loi == NULL)
2885                 loi = lsm->lsm_oinfo[0];
2886
2887         client_obd_list_lock(&cli->cl_loi_list_lock);
2888
2889         LASSERT(off + count <= CFS_PAGE_SIZE);
2890         oap->oap_cmd = cmd;
2891         oap->oap_page_off = off;
2892         oap->oap_count = count;
2893         oap->oap_brw_flags = brw_flags;
2894         oap->oap_async_flags = async_flags;
2895
2896         if (cmd & OBD_BRW_WRITE) {
2897                 rc = osc_enter_cache(env, cli, loi, oap);
2898                 if (rc) {
2899                         client_obd_list_unlock(&cli->cl_loi_list_lock);
2900                         RETURN(rc);
2901                 }
2902         }
2903
2904         osc_oap_to_pending(oap);
2905         loi_list_maint(cli, loi);
2906
2907         LOI_DEBUG(loi, "oap %p page %p added for cmd %d\n", oap, oap->oap_page,
2908                   cmd);
2909
2910         osc_check_rpcs(env, cli);
2911         client_obd_list_unlock(&cli->cl_loi_list_lock);
2912
2913         RETURN(0);
2914 }
2915
2916 /* aka (~was & now & flag), but this is more clear :) */
2917 #define SETTING(was, now, flag) (!(was & flag) && (now & flag))
2918
2919 int osc_set_async_flags_base(struct client_obd *cli,
2920                              struct lov_oinfo *loi, struct osc_async_page *oap,
2921                              obd_flag async_flags)
2922 {
2923         struct loi_oap_pages *lop;
2924         ENTRY;
2925
2926         LASSERT(!list_empty(&oap->oap_pending_item));
2927
2928         if (oap->oap_cmd & OBD_BRW_WRITE) {
2929                 lop = &loi->loi_write_lop;
2930         } else {
2931                 lop = &loi->loi_read_lop;
2932         }
2933
2934         if ((oap->oap_async_flags & async_flags) == async_flags)
2935                 RETURN(0);
2936
2937         if (SETTING(oap->oap_async_flags, async_flags, ASYNC_READY))
2938                 oap->oap_async_flags |= ASYNC_READY;
2939
2940         if (SETTING(oap->oap_async_flags, async_flags, ASYNC_URGENT) &&
2941             list_empty(&oap->oap_rpc_item)) {
2942                 if (oap->oap_async_flags & ASYNC_HP)
2943                         list_add(&oap->oap_urgent_item, &lop->lop_urgent);
2944                 else
2945                         list_add_tail(&oap->oap_urgent_item, &lop->lop_urgent);
2946                 oap->oap_async_flags |= ASYNC_URGENT;
2947                 loi_list_maint(cli, loi);
2948         }
2949
2950         LOI_DEBUG(loi, "oap %p page %p has flags %x\n", oap, oap->oap_page,
2951                         oap->oap_async_flags);
2952         RETURN(0);
2953 }
2954
2955 int osc_teardown_async_page(struct obd_export *exp,
2956                             struct lov_stripe_md *lsm,
2957                             struct lov_oinfo *loi, void *cookie)
2958 {
2959         struct client_obd *cli = &exp->exp_obd->u.cli;
2960         struct loi_oap_pages *lop;
2961         struct osc_async_page *oap;
2962         int rc = 0;
2963         ENTRY;
2964
2965         oap = oap_from_cookie(cookie);
2966         if (IS_ERR(oap))
2967                 RETURN(PTR_ERR(oap));
2968
2969         if (loi == NULL)
2970                 loi = lsm->lsm_oinfo[0];
2971
2972         if (oap->oap_cmd & OBD_BRW_WRITE) {
2973                 lop = &loi->loi_write_lop;
2974         } else {
2975                 lop = &loi->loi_read_lop;
2976         }
2977
2978         client_obd_list_lock(&cli->cl_loi_list_lock);
2979
2980         if (!list_empty(&oap->oap_rpc_item))
2981                 GOTO(out, rc = -EBUSY);
2982
2983         osc_exit_cache(cli, oap, 0);
2984         osc_wake_cache_waiters(cli);
2985
2986         if (!list_empty(&oap->oap_urgent_item)) {
2987                 list_del_init(&oap->oap_urgent_item);
2988                 oap->oap_async_flags &= ~(ASYNC_URGENT | ASYNC_HP);
2989         }
2990         if (!list_empty(&oap->oap_pending_item)) {
2991                 list_del_init(&oap->oap_pending_item);
2992                 lop_update_pending(cli, lop, oap->oap_cmd, -1);
2993         }
2994         loi_list_maint(cli, loi);
2995         LOI_DEBUG(loi, "oap %p page %p torn down\n", oap, oap->oap_page);
2996 out:
2997         client_obd_list_unlock(&cli->cl_loi_list_lock);
2998         RETURN(rc);
2999 }
3000
3001 static void osc_set_lock_data_with_check(struct ldlm_lock *lock,
3002                                          struct ldlm_enqueue_info *einfo,
3003                                          int flags)
3004 {
3005         void *data = einfo->ei_cbdata;
3006
3007         LASSERT(lock != NULL);
3008         LASSERT(lock->l_blocking_ast == einfo->ei_cb_bl);
3009         LASSERT(lock->l_resource->lr_type == einfo->ei_type);
3010         LASSERT(lock->l_completion_ast == einfo->ei_cb_cp);
3011         LASSERT(lock->l_glimpse_ast == einfo->ei_cb_gl);
3012
3013         lock_res_and_lock(lock);
3014         spin_lock(&osc_ast_guard);
3015         LASSERT(lock->l_ast_data == NULL || lock->l_ast_data == data);
3016         lock->l_ast_data = data;
3017         spin_unlock(&osc_ast_guard);
3018         unlock_res_and_lock(lock);
3019 }
3020
3021 static void osc_set_data_with_check(struct lustre_handle *lockh,
3022                                     struct ldlm_enqueue_info *einfo,
3023                                     int flags)
3024 {
3025         struct ldlm_lock *lock = ldlm_handle2lock(lockh);
3026
3027         if (lock != NULL) {
3028                 osc_set_lock_data_with_check(lock, einfo, flags);
3029                 LDLM_LOCK_PUT(lock);
3030         } else
3031                 CERROR("lockh %p, data %p - client evicted?\n",
3032                        lockh, einfo->ei_cbdata);
3033 }
3034
3035 static int osc_change_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
3036                              ldlm_iterator_t replace, void *data)
3037 {
3038         struct ldlm_res_id res_id;
3039         struct obd_device *obd = class_exp2obd(exp);
3040
3041         osc_build_res_name(lsm->lsm_object_id, lsm->lsm_object_gr, &res_id);
3042         ldlm_resource_iterate(obd->obd_namespace, &res_id, replace, data);
3043         return 0;
3044 }
3045
3046 static int osc_enqueue_fini(struct ptlrpc_request *req, struct ost_lvb *lvb,
3047                             obd_enqueue_update_f upcall, void *cookie,
3048                             int *flags, int rc)
3049 {
3050         int intent = *flags & LDLM_FL_HAS_INTENT;
3051         ENTRY;
3052
3053         if (intent) {
3054                 /* The request was created before ldlm_cli_enqueue call. */
3055                 if (rc == ELDLM_LOCK_ABORTED) {
3056                         struct ldlm_reply *rep;
3057                         rep = req_capsule_server_get(&req->rq_pill,
3058                                                      &RMF_DLM_REP);
3059
3060                         LASSERT(rep != NULL);
3061                         if (rep->lock_policy_res1)
3062                                 rc = rep->lock_policy_res1;
3063                 }
3064         }
3065
3066         if ((intent && rc == ELDLM_LOCK_ABORTED) || !rc) {
3067                 *flags |= LDLM_FL_LVB_READY;
3068                 CDEBUG(D_INODE,"got kms "LPU64" blocks "LPU64" mtime "LPU64"\n",
3069                        lvb->lvb_size, lvb->lvb_blocks, lvb->lvb_mtime);
3070         }
3071
3072         /* Call the update callback. */
3073         rc = (*upcall)(cookie, rc);
3074         RETURN(rc);
3075 }
3076
3077 static int osc_enqueue_interpret(const struct lu_env *env,
3078                                  struct ptlrpc_request *req,
3079                                  struct osc_enqueue_args *aa, int rc)
3080 {
3081         struct ldlm_lock *lock;
3082         struct lustre_handle handle;
3083         __u32 mode;
3084
3085         /* Make a local copy of a lock handle and a mode, because aa->oa_*
3086          * might be freed anytime after lock upcall has been called. */
3087         lustre_handle_copy(&handle, aa->oa_lockh);
3088         mode = aa->oa_ei->ei_mode;
3089
3090         /* ldlm_cli_enqueue is holding a reference on the lock, so it must
3091          * be valid. */
3092         lock = ldlm_handle2lock(&handle);
3093
3094         /* Take an additional reference so that a blocking AST that
3095          * ldlm_cli_enqueue_fini() might post for a failed lock, is guaranteed
3096          * to arrive after an upcall has been executed by
3097          * osc_enqueue_fini(). */
3098         ldlm_lock_addref(&handle, mode);
3099
3100         /* Complete obtaining the lock procedure. */
3101         rc = ldlm_cli_enqueue_fini(aa->oa_exp, req, aa->oa_ei->ei_type, 1,
3102                                    mode, aa->oa_flags, aa->oa_lvb,
3103                                    sizeof(*aa->oa_lvb), lustre_swab_ost_lvb,
3104                                    &handle, rc);
3105         /* Complete osc stuff. */
3106         rc = osc_enqueue_fini(req, aa->oa_lvb,
3107                               aa->oa_upcall, aa->oa_cookie, aa->oa_flags, rc);
3108
3109         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_CANCEL_RACE, 10);
3110
3111         /* Release the lock for async request. */
3112         if (lustre_handle_is_used(&handle) && rc == ELDLM_OK)
3113                 /*
3114                  * Releases a reference taken by ldlm_cli_enqueue(), if it is
3115                  * not already released by
3116                  * ldlm_cli_enqueue_fini()->failed_lock_cleanup()
3117                  */
3118                 ldlm_lock_decref(&handle, mode);
3119
3120         LASSERTF(lock != NULL, "lockh %p, req %p, aa %p - client evicted?\n",
3121                  aa->oa_lockh, req, aa);
3122         ldlm_lock_decref(&handle, mode);
3123         LDLM_LOCK_PUT(lock);
3124         return rc;
3125 }
3126
3127 void osc_update_enqueue(struct lustre_handle *lov_lockhp,
3128                         struct lov_oinfo *loi, int flags,
3129                         struct ost_lvb *lvb, __u32 mode, int rc)
3130 {
3131         if (rc == ELDLM_OK) {
3132                 struct ldlm_lock *lock = ldlm_handle2lock(lov_lockhp);
3133                 __u64 tmp;
3134
3135                 LASSERT(lock != NULL);
3136                 loi->loi_lvb = *lvb;
3137                 tmp = loi->loi_lvb.lvb_size;
3138                 /* Extend KMS up to the end of this lock and no further
3139                  * A lock on [x,y] means a KMS of up to y + 1 bytes! */
3140                 if (tmp > lock->l_policy_data.l_extent.end)
3141                         tmp = lock->l_policy_data.l_extent.end + 1;
3142                 if (tmp >= loi->loi_kms) {
3143                         LDLM_DEBUG(lock, "lock acquired, setting rss="LPU64
3144                                    ", kms="LPU64, loi->loi_lvb.lvb_size, tmp);
3145                         loi_kms_set(loi, tmp);
3146                 } else {
3147                         LDLM_DEBUG(lock, "lock acquired, setting rss="
3148                                    LPU64"; leaving kms="LPU64", end="LPU64,
3149                                    loi->loi_lvb.lvb_size, loi->loi_kms,
3150                                    lock->l_policy_data.l_extent.end);
3151                 }
3152                 ldlm_lock_allow_match(lock);
3153                 LDLM_LOCK_PUT(lock);
3154         } else if (rc == ELDLM_LOCK_ABORTED && (flags & LDLM_FL_HAS_INTENT)) {
3155                 loi->loi_lvb = *lvb;
3156                 CDEBUG(D_INODE, "glimpsed, setting rss="LPU64"; leaving"
3157                        " kms="LPU64"\n", loi->loi_lvb.lvb_size, loi->loi_kms);
3158                 rc = ELDLM_OK;
3159         }
3160 }
3161 EXPORT_SYMBOL(osc_update_enqueue);
3162
3163 struct ptlrpc_request_set *PTLRPCD_SET = (void *)1;
3164
3165 /* When enqueuing asynchronously, locks are not ordered, we can obtain a lock
3166  * from the 2nd OSC before a lock from the 1st one. This does not deadlock with
3167  * other synchronous requests, however keeping some locks and trying to obtain
3168  * others may take a considerable amount of time in a case of ost failure; and
3169  * when other sync requests do not get released lock from a client, the client
3170  * is excluded from the cluster -- such scenarious make the life difficult, so
3171  * release locks just after they are obtained. */
3172 int osc_enqueue_base(struct obd_export *exp, struct ldlm_res_id *res_id,
3173                      int *flags, ldlm_policy_data_t *policy,
3174                      struct ost_lvb *lvb, int kms_valid,
3175                      obd_enqueue_update_f upcall, void *cookie,
3176                      struct ldlm_enqueue_info *einfo,
3177                      struct lustre_handle *lockh,
3178                      struct ptlrpc_request_set *rqset, int async)
3179 {
3180         struct obd_device *obd = exp->exp_obd;
3181         struct ptlrpc_request *req = NULL;
3182         int intent = *flags & LDLM_FL_HAS_INTENT;