lustre/osc/osc_request.c

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