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