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, PSCOPE_OTHER);
 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, PSCOPE_OTHER);
 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, PSCOPE_OTHER);
 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, PSCOPE_OTHER);
 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         if (oa->o_valid & OBD_MD_FLFLAGS)
1492                 cksum_type = cksum_type_unpack(oa->o_flags);
1493         else
1494                 cksum_type = OBD_CKSUM_CRC32;
1495
1496         new_cksum = osc_checksum_bulk(nob, page_count, pga, OST_WRITE,
1497                                       cksum_type);
1498
1499         if (cksum_type != client_cksum_type)
1500                 msg = "the server did not use the checksum type specified in "
1501                       "the original request - likely a protocol problem";
1502         else if (new_cksum == server_cksum)
1503                 msg = "changed on the client after we checksummed it - "
1504                       "likely false positive due to mmap IO (bug 11742)";
1505         else if (new_cksum == client_cksum)
1506                 msg = "changed in transit before arrival at OST";
1507         else
1508                 msg = "changed in transit AND doesn't match the original - "
1509                       "likely false positive due to mmap IO (bug 11742)";
1510
1511         LCONSOLE_ERROR_MSG(0x132, "BAD WRITE CHECKSUM: %s: from %s inode "DFID
1512                            " object "LPU64"/"LPU64" extent ["LPU64"-"LPU64"]\n",
1513                            msg, libcfs_nid2str(peer->nid),
1514                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_seq : (__u64)0,
1515                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_oid : 0,
1516                            oa->o_valid & OBD_MD_FLFID ? oa->o_parent_ver : 0,
1517                            oa->o_id,
1518                            oa->o_valid & OBD_MD_FLGROUP ? oa->o_seq : (__u64)0,
1519                            pga[0]->off,
1520                            pga[page_count-1]->off + pga[page_count-1]->count - 1);
1521         CERROR("original client csum %x (type %x), server csum %x (type %x), "
1522                "client csum now %x\n", client_cksum, client_cksum_type,
1523                server_cksum, cksum_type, new_cksum);
1524         return 1;
1525 }
1526
1527 /* Note rc enters this function as number of bytes transferred */
1528 static int osc_brw_fini_request(struct ptlrpc_request *req, int rc)
1529 {
1530         struct osc_brw_async_args *aa = (void *)&req->rq_async_args;
1531         const lnet_process_id_t *peer =
1532                         &req->rq_import->imp_connection->c_peer;
1533         struct client_obd *cli = aa->aa_cli;
1534         struct ost_body *body;
1535         __u32 client_cksum = 0;
1536         ENTRY;
1537
1538         if (rc < 0 && rc != -EDQUOT) {
1539                 DEBUG_REQ(D_INFO, req, "Failed request with rc = %d\n", rc);
1540                 RETURN(rc);
1541         }
1542
1543         LASSERTF(req->rq_repmsg != NULL, "rc = %d\n", rc);
1544         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
1545         if (body == NULL) {
1546                 DEBUG_REQ(D_INFO, req, "Can't unpack body\n");
1547                 RETURN(-EPROTO);
1548         }
1549
1550 #ifdef HAVE_QUOTA_SUPPORT
1551         /* set/clear over quota flag for a uid/gid */
1552         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE &&
1553             body->oa.o_valid & (OBD_MD_FLUSRQUOTA | OBD_MD_FLGRPQUOTA)) {
1554                 unsigned int qid[MAXQUOTAS] = { body->oa.o_uid, body->oa.o_gid };
1555
1556                 CDEBUG(D_QUOTA, "setdq for [%u %u] with valid "LPX64", flags %x\n",
1557                        body->oa.o_uid, body->oa.o_gid, body->oa.o_valid,
1558                        body->oa.o_flags);
1559                 lquota_setdq(quota_interface, cli, qid, body->oa.o_valid,
1560                              body->oa.o_flags);
1561         }
1562 #endif
1563
1564         osc_update_grant(cli, body);
1565
1566         if (rc < 0)
1567                 RETURN(rc);
1568
1569         if (aa->aa_oa->o_valid & OBD_MD_FLCKSUM)
1570                 client_cksum = aa->aa_oa->o_cksum; /* save for later */
1571
1572         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE) {
1573                 if (rc > 0) {
1574                         CERROR("Unexpected +ve rc %d\n", rc);
1575                         RETURN(-EPROTO);
1576                 }
1577                 LASSERT(req->rq_bulk->bd_nob == aa->aa_requested_nob);
1578
1579                 if (sptlrpc_cli_unwrap_bulk_write(req, req->rq_bulk))
1580                         RETURN(-EAGAIN);
1581
1582                 if ((aa->aa_oa->o_valid & OBD_MD_FLCKSUM) && client_cksum &&
1583                     check_write_checksum(&body->oa, peer, client_cksum,
1584                                          body->oa.o_cksum, aa->aa_requested_nob,
1585                                          aa->aa_page_count, aa->aa_ppga,
1586                                          cksum_type_unpack(aa->aa_oa->o_flags)))
1587                         RETURN(-EAGAIN);
1588
1589                 rc = check_write_rcs(req, aa->aa_requested_nob,aa->aa_nio_count,
1590                                      aa->aa_page_count, aa->aa_ppga);
1591                 GOTO(out, rc);
1592         }
1593
1594         /* The rest of this function executes only for OST_READs */
1595
1596         /* if unwrap_bulk failed, return -EAGAIN to retry */
1597         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk, rc);
1598         if (rc < 0)
1599                 GOTO(out, rc = -EAGAIN);
1600
1601         if (rc > aa->aa_requested_nob) {
1602                 CERROR("Unexpected rc %d (%d requested)\n", rc,
1603                        aa->aa_requested_nob);
1604                 RETURN(-EPROTO);
1605         }
1606
1607         if (rc != req->rq_bulk->bd_nob_transferred) {
1608                 CERROR ("Unexpected rc %d (%d transferred)\n",
1609                         rc, req->rq_bulk->bd_nob_transferred);
1610                 return (-EPROTO);
1611         }
1612
1613         if (rc < aa->aa_requested_nob)
1614                 handle_short_read(rc, aa->aa_page_count, aa->aa_ppga);
1615
1616         if (body->oa.o_valid & OBD_MD_FLCKSUM) {
1617                 static int cksum_counter;
1618                 __u32      server_cksum = body->oa.o_cksum;
1619                 char      *via;
1620                 char      *router;
1621                 cksum_type_t cksum_type;
1622
1623                 if (body->oa.o_valid & OBD_MD_FLFLAGS)
1624                         cksum_type = cksum_type_unpack(body->oa.o_flags);
1625                 else
1626                         cksum_type = OBD_CKSUM_CRC32;
1627                 client_cksum = osc_checksum_bulk(rc, aa->aa_page_count,
1628                                                  aa->aa_ppga, OST_READ,
1629                                                  cksum_type);
1630
1631                 if (peer->nid == req->rq_bulk->bd_sender) {
1632                         via = router = "";
1633                 } else {
1634                         via = " via ";
1635                         router = libcfs_nid2str(req->rq_bulk->bd_sender);
1636                 }
1637
1638                 if (server_cksum == ~0 && rc > 0) {
1639                         CERROR("Protocol error: server %s set the 'checksum' "
1640                                "bit, but didn't send a checksum.  Not fatal, "
1641                                "but please notify on http://bugs.whamcloud.com/\n",
1642                                libcfs_nid2str(peer->nid));
1643                 } else if (server_cksum != client_cksum) {
1644                         LCONSOLE_ERROR_MSG(0x133, "%s: BAD READ CHECKSUM: from "
1645                                            "%s%s%s inode "DFID" object "
1646                                            LPU64"/"LPU64" extent "
1647                                            "["LPU64"-"LPU64"]\n",
1648                                            req->rq_import->imp_obd->obd_name,
1649                                            libcfs_nid2str(peer->nid),
1650                                            via, router,
1651                                            body->oa.o_valid & OBD_MD_FLFID ?
1652                                                 body->oa.o_parent_seq : (__u64)0,
1653                                            body->oa.o_valid & OBD_MD_FLFID ?
1654                                                 body->oa.o_parent_oid : 0,
1655                                            body->oa.o_valid & OBD_MD_FLFID ?
1656                                                 body->oa.o_parent_ver : 0,
1657                                            body->oa.o_id,
1658                                            body->oa.o_valid & OBD_MD_FLGROUP ?
1659                                                 body->oa.o_seq : (__u64)0,
1660                                            aa->aa_ppga[0]->off,
1661                                            aa->aa_ppga[aa->aa_page_count-1]->off +
1662                                            aa->aa_ppga[aa->aa_page_count-1]->count -
1663                                                                         1);
1664                         CERROR("client %x, server %x, cksum_type %x\n",
1665                                client_cksum, server_cksum, cksum_type);
1666                         cksum_counter = 0;
1667                         aa->aa_oa->o_cksum = client_cksum;
1668                         rc = -EAGAIN;
1669                 } else {
1670                         cksum_counter++;
1671                         CDEBUG(D_PAGE, "checksum %x confirmed\n", client_cksum);
1672                         rc = 0;
1673                 }
1674         } else if (unlikely(client_cksum)) {
1675                 static int cksum_missed;
1676
1677                 cksum_missed++;
1678                 if ((cksum_missed & (-cksum_missed)) == cksum_missed)
1679                         CERROR("Checksum %u requested from %s but not sent\n",
1680                                cksum_missed, libcfs_nid2str(peer->nid));
1681         } else {
1682                 rc = 0;
1683         }
1684 out:
1685         if (rc >= 0)
1686                 lustre_get_wire_obdo(aa->aa_oa, &body->oa);
1687
1688         RETURN(rc);
1689 }
1690
1691 static int osc_brw_internal(int cmd, struct obd_export *exp, struct obdo *oa,
1692                             struct lov_stripe_md *lsm,
1693                             obd_count page_count, struct brw_page **pga,
1694                             struct obd_capa *ocapa)
1695 {
1696         struct ptlrpc_request *req;
1697         int                    rc;
1698         cfs_waitq_t            waitq;
1699         int                    resends = 0;
1700         struct l_wait_info     lwi;
1701
1702         ENTRY;
1703
1704         cfs_waitq_init(&waitq);
1705
1706 restart_bulk:
1707         rc = osc_brw_prep_request(cmd, &exp->exp_obd->u.cli, oa, lsm,
1708                                   page_count, pga, &req, ocapa, 0, resends);
1709         if (rc != 0)
1710                 return (rc);
1711
1712         rc = ptlrpc_queue_wait(req);
1713
1714         if (rc == -ETIMEDOUT && req->rq_resend) {
1715                 DEBUG_REQ(D_HA, req,  "BULK TIMEOUT");
1716                 ptlrpc_req_finished(req);
1717                 goto restart_bulk;
1718         }
1719
1720         rc = osc_brw_fini_request(req, rc);
1721
1722         ptlrpc_req_finished(req);
1723         if (osc_recoverable_error(rc)) {
1724                 resends++;
1725                 if (!client_should_resend(resends, &exp->exp_obd->u.cli)) {
1726                         CERROR("too many resend retries, returning error\n");
1727                         RETURN(-EIO);
1728                 }
1729
1730                 lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(resends), NULL, NULL, NULL);
1731                 l_wait_event(waitq, 0, &lwi);
1732
1733                 goto restart_bulk;
1734         }
1735
1736         RETURN (rc);
1737 }
1738
1739 int osc_brw_redo_request(struct ptlrpc_request *request,
1740                          struct osc_brw_async_args *aa)
1741 {
1742         struct ptlrpc_request *new_req;
1743         struct ptlrpc_request_set *set = request->rq_set;
1744         struct osc_brw_async_args *new_aa;
1745         struct osc_async_page *oap;
1746         int rc = 0;
1747         ENTRY;
1748
1749         if (!client_should_resend(aa->aa_resends, aa->aa_cli)) {
1750                 CERROR("too many resent retries, returning error\n");
1751                 RETURN(-EIO);
1752         }
1753
1754         DEBUG_REQ(D_ERROR, request, "redo for recoverable error");
1755
1756         rc = osc_brw_prep_request(lustre_msg_get_opc(request->rq_reqmsg) ==
1757                                         OST_WRITE ? OBD_BRW_WRITE :OBD_BRW_READ,
1758                                   aa->aa_cli, aa->aa_oa,
1759                                   NULL /* lsm unused by osc currently */,
1760                                   aa->aa_page_count, aa->aa_ppga,
1761                                   &new_req, aa->aa_ocapa, 0, 1);
1762         if (rc)
1763                 RETURN(rc);
1764
1765         client_obd_list_lock(&aa->aa_cli->cl_loi_list_lock);
1766
1767         cfs_list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
1768                 if (oap->oap_request != NULL) {
1769                         LASSERTF(request == oap->oap_request,
1770                                  "request %p != oap_request %p\n",
1771                                  request, oap->oap_request);
1772                         if (oap->oap_interrupted) {
1773                                 client_obd_list_unlock(&aa->aa_cli->cl_loi_list_lock);
1774                                 ptlrpc_req_finished(new_req);
1775                                 RETURN(-EINTR);
1776                         }
1777                 }
1778         }
1779         /* New request takes over pga and oaps from old request.
1780          * Note that copying a list_head doesn't work, need to move it... */
1781         aa->aa_resends++;
1782         new_req->rq_interpret_reply = request->rq_interpret_reply;
1783         new_req->rq_async_args = request->rq_async_args;
1784         new_req->rq_sent = cfs_time_current_sec() + aa->aa_resends;
1785
1786         new_aa = ptlrpc_req_async_args(new_req);
1787
1788         CFS_INIT_LIST_HEAD(&new_aa->aa_oaps);
1789         cfs_list_splice(&aa->aa_oaps, &new_aa->aa_oaps);
1790         CFS_INIT_LIST_HEAD(&aa->aa_oaps);
1791
1792         cfs_list_for_each_entry(oap, &new_aa->aa_oaps, oap_rpc_item) {
1793                 if (oap->oap_request) {
1794                         ptlrpc_req_finished(oap->oap_request);
1795                         oap->oap_request = ptlrpc_request_addref(new_req);
1796                 }
1797         }
1798
1799         new_aa->aa_ocapa = aa->aa_ocapa;
1800         aa->aa_ocapa = NULL;
1801
1802         /* use ptlrpc_set_add_req is safe because interpret functions work
1803          * in check_set context. only one way exist with access to request
1804          * from different thread got -EINTR - this way protected with
1805          * cl_loi_list_lock */
1806         ptlrpc_set_add_req(set, new_req);
1807
1808         client_obd_list_unlock(&aa->aa_cli->cl_loi_list_lock);
1809
1810         DEBUG_REQ(D_INFO, new_req, "new request");
1811         RETURN(0);
1812 }
1813
1814 /*
1815  * ugh, we want disk allocation on the target to happen in offset order.  we'll
1816  * follow sedgewicks advice and stick to the dead simple shellsort -- it'll do
1817  * fine for our small page arrays and doesn't require allocation.  its an
1818  * insertion sort that swaps elements that are strides apart, shrinking the
1819  * stride down until its '1' and the array is sorted.
1820  */
1821 static void sort_brw_pages(struct brw_page **array, int num)
1822 {
1823         int stride, i, j;
1824         struct brw_page *tmp;
1825
1826         if (num == 1)
1827                 return;
1828         for (stride = 1; stride < num ; stride = (stride * 3) + 1)
1829                 ;
1830
1831         do {
1832                 stride /= 3;
1833                 for (i = stride ; i < num ; i++) {
1834                         tmp = array[i];
1835                         j = i;
1836                         while (j >= stride && array[j - stride]->off > tmp->off) {
1837                                 array[j] = array[j - stride];
1838                                 j -= stride;
1839                         }
1840                         array[j] = tmp;
1841                 }
1842         } while (stride > 1);
1843 }
1844
1845 static obd_count max_unfragmented_pages(struct brw_page **pg, obd_count pages)
1846 {
1847         int count = 1;
1848         int offset;
1849         int i = 0;
1850
1851         LASSERT (pages > 0);
1852         offset = pg[i]->off & ~CFS_PAGE_MASK;
1853
1854         for (;;) {
1855                 pages--;
1856                 if (pages == 0)         /* that's all */
1857                         return count;
1858
1859                 if (offset + pg[i]->count < CFS_PAGE_SIZE)
1860                         return count;   /* doesn't end on page boundary */
1861
1862                 i++;
1863                 offset = pg[i]->off & ~CFS_PAGE_MASK;
1864                 if (offset != 0)        /* doesn't start on page boundary */
1865                         return count;
1866
1867                 count++;
1868         }
1869 }
1870
1871 static struct brw_page **osc_build_ppga(struct brw_page *pga, obd_count count)
1872 {
1873         struct brw_page **ppga;
1874         int i;
1875
1876         OBD_ALLOC(ppga, sizeof(*ppga) * count);
1877         if (ppga == NULL)
1878                 return NULL;
1879
1880         for (i = 0; i < count; i++)
1881                 ppga[i] = pga + i;
1882         return ppga;
1883 }
1884
1885 static void osc_release_ppga(struct brw_page **ppga, obd_count count)
1886 {
1887         LASSERT(ppga != NULL);
1888         OBD_FREE(ppga, sizeof(*ppga) * count);
1889 }
1890
1891 static int osc_brw(int cmd, struct obd_export *exp, struct obd_info *oinfo,
1892                    obd_count page_count, struct brw_page *pga,
1893                    struct obd_trans_info *oti)
1894 {
1895         struct obdo *saved_oa = NULL;
1896         struct brw_page **ppga, **orig;
1897         struct obd_import *imp = class_exp2cliimp(exp);
1898         struct client_obd *cli;
1899         int rc, page_count_orig;
1900         ENTRY;
1901
1902         LASSERT((imp != NULL) && (imp->imp_obd != NULL));
1903         cli = &imp->imp_obd->u.cli;
1904
1905         if (cmd & OBD_BRW_CHECK) {
1906                 /* The caller just wants to know if there's a chance that this
1907                  * I/O can succeed */
1908
1909                 if (imp->imp_invalid)
1910                         RETURN(-EIO);
1911                 RETURN(0);
1912         }
1913
1914         /* test_brw with a failed create can trip this, maybe others. */
1915         LASSERT(cli->cl_max_pages_per_rpc);
1916
1917         rc = 0;
1918
1919         orig = ppga = osc_build_ppga(pga, page_count);
1920         if (ppga == NULL)
1921                 RETURN(-ENOMEM);
1922         page_count_orig = page_count;
1923
1924         sort_brw_pages(ppga, page_count);
1925         while (page_count) {
1926                 obd_count pages_per_brw;
1927
1928                 if (page_count > cli->cl_max_pages_per_rpc)
1929                         pages_per_brw = cli->cl_max_pages_per_rpc;
1930                 else
1931                         pages_per_brw = page_count;
1932
1933                 pages_per_brw = max_unfragmented_pages(ppga, pages_per_brw);
1934
1935                 if (saved_oa != NULL) {
1936                         /* restore previously saved oa */
1937                         *oinfo->oi_oa = *saved_oa;
1938                 } else if (page_count > pages_per_brw) {
1939                         /* save a copy of oa (brw will clobber it) */
1940                         OBDO_ALLOC(saved_oa);
1941                         if (saved_oa == NULL)
1942                                 GOTO(out, rc = -ENOMEM);
1943                         *saved_oa = *oinfo->oi_oa;
1944                 }
1945
1946                 rc = osc_brw_internal(cmd, exp, oinfo->oi_oa, oinfo->oi_md,
1947                                       pages_per_brw, ppga, oinfo->oi_capa);
1948
1949                 if (rc != 0)
1950                         break;
1951
1952                 page_count -= pages_per_brw;
1953                 ppga += pages_per_brw;
1954         }
1955
1956 out:
1957         osc_release_ppga(orig, page_count_orig);
1958
1959         if (saved_oa != NULL)
1960                 OBDO_FREE(saved_oa);
1961
1962         RETURN(rc);
1963 }
1964
1965 /* The companion to osc_enter_cache(), called when @oap is no longer part of
1966  * the dirty accounting.  Writeback completes or truncate happens before
1967  * writing starts.  Must be called with the loi lock held. */
1968 static void osc_exit_cache(struct client_obd *cli, struct osc_async_page *oap,
1969                            int sent)
1970 {
1971         osc_release_write_grant(cli, &oap->oap_brw_page, sent);
1972 }
1973
1974
1975 /* This maintains the lists of pending pages to read/write for a given object
1976  * (lop).  This is used by osc_check_rpcs->osc_next_loi() and loi_list_maint()
1977  * to quickly find objects that are ready to send an RPC. */
1978 static int lop_makes_rpc(struct client_obd *cli, struct loi_oap_pages *lop,
1979                          int cmd)
1980 {
1981         int optimal;
1982         ENTRY;
1983
1984         if (lop->lop_num_pending == 0)
1985                 RETURN(0);
1986
1987         /* if we have an invalid import we want to drain the queued pages
1988          * by forcing them through rpcs that immediately fail and complete
1989          * the pages.  recovery relies on this to empty the queued pages
1990          * before canceling the locks and evicting down the llite pages */
1991         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
1992                 RETURN(1);
1993
1994         /* stream rpcs in queue order as long as as there is an urgent page
1995          * queued.  this is our cheap solution for good batching in the case
1996          * where writepage marks some random page in the middle of the file
1997          * as urgent because of, say, memory pressure */
1998         if (!cfs_list_empty(&lop->lop_urgent)) {
1999                 CDEBUG(D_CACHE, "urgent request forcing RPC\n");
2000                 RETURN(1);
2001         }
2002         /* fire off rpcs when we have 'optimal' rpcs as tuned for the wire. */
2003         optimal = cli->cl_max_pages_per_rpc;
2004         if (cmd & OBD_BRW_WRITE) {
2005                 /* trigger a write rpc stream as long as there are dirtiers
2006                  * waiting for space.  as they're waiting, they're not going to
2007                  * create more pages to coalesce with what's waiting.. */
2008                 if (!cfs_list_empty(&cli->cl_cache_waiters)) {
2009                         CDEBUG(D_CACHE, "cache waiters forcing RPC\n");
2010                         RETURN(1);
2011                 }
2012                 /* +16 to avoid triggering rpcs that would want to include pages
2013                  * that are being queued but which can't be made ready until
2014                  * the queuer finishes with the page. this is a wart for
2015                  * llite::commit_write() */
2016                 optimal += 16;
2017         }
2018         if (lop->lop_num_pending >= optimal)
2019                 RETURN(1);
2020
2021         RETURN(0);
2022 }
2023
2024 static int lop_makes_hprpc(struct loi_oap_pages *lop)
2025 {
2026         struct osc_async_page *oap;
2027         ENTRY;
2028
2029         if (cfs_list_empty(&lop->lop_urgent))
2030                 RETURN(0);
2031
2032         oap = cfs_list_entry(lop->lop_urgent.next,
2033                          struct osc_async_page, oap_urgent_item);
2034
2035         if (oap->oap_async_flags & ASYNC_HP) {
2036                 CDEBUG(D_CACHE, "hp request forcing RPC\n");
2037                 RETURN(1);
2038         }
2039
2040         RETURN(0);
2041 }
2042
2043 static void on_list(cfs_list_t *item, cfs_list_t *list,
2044                     int should_be_on)
2045 {
2046         if (cfs_list_empty(item) && should_be_on)
2047                 cfs_list_add_tail(item, list);
2048         else if (!cfs_list_empty(item) && !should_be_on)
2049                 cfs_list_del_init(item);
2050 }
2051
2052 /* maintain the loi's cli list membership invariants so that osc_send_oap_rpc
2053  * can find pages to build into rpcs quickly */
2054 void loi_list_maint(struct client_obd *cli, struct lov_oinfo *loi)
2055 {
2056         if (lop_makes_hprpc(&loi->loi_write_lop) ||
2057             lop_makes_hprpc(&loi->loi_read_lop)) {
2058                 /* HP rpc */
2059                 on_list(&loi->loi_ready_item, &cli->cl_loi_ready_list, 0);
2060                 on_list(&loi->loi_hp_ready_item, &cli->cl_loi_hp_ready_list, 1);
2061         } else {
2062                 on_list(&loi->loi_hp_ready_item, &cli->cl_loi_hp_ready_list, 0);
2063                 on_list(&loi->loi_ready_item, &cli->cl_loi_ready_list,
2064                         lop_makes_rpc(cli, &loi->loi_write_lop, OBD_BRW_WRITE)||
2065                         lop_makes_rpc(cli, &loi->loi_read_lop, OBD_BRW_READ));
2066         }
2067
2068         on_list(&loi->loi_write_item, &cli->cl_loi_write_list,
2069                 loi->loi_write_lop.lop_num_pending);
2070
2071         on_list(&loi->loi_read_item, &cli->cl_loi_read_list,
2072                 loi->loi_read_lop.lop_num_pending);
2073 }
2074
2075 static void lop_update_pending(struct client_obd *cli,
2076                                struct loi_oap_pages *lop, int cmd, int delta)
2077 {
2078         lop->lop_num_pending += delta;
2079         if (cmd & OBD_BRW_WRITE)
2080                 cli->cl_pending_w_pages += delta;
2081         else
2082                 cli->cl_pending_r_pages += delta;
2083 }
2084
2085 /**
2086  * this is called when a sync waiter receives an interruption.  Its job is to
2087  * get the caller woken as soon as possible.  If its page hasn't been put in an
2088  * rpc yet it can dequeue immediately.  Otherwise it has to mark the rpc as
2089  * desiring interruption which will forcefully complete the rpc once the rpc
2090  * has timed out.
2091  */
2092 int osc_oap_interrupted(const struct lu_env *env, struct osc_async_page *oap)
2093 {
2094         struct loi_oap_pages *lop;
2095         struct lov_oinfo *loi;
2096         int rc = -EBUSY;
2097         ENTRY;
2098
2099         LASSERT(!oap->oap_interrupted);
2100         oap->oap_interrupted = 1;
2101
2102         /* ok, it's been put in an rpc. only one oap gets a request reference */
2103         if (oap->oap_request != NULL) {
2104                 ptlrpc_mark_interrupted(oap->oap_request);
2105                 ptlrpcd_wake(oap->oap_request);
2106                 ptlrpc_req_finished(oap->oap_request);
2107                 oap->oap_request = NULL;
2108         }
2109
2110         /*
2111          * page completion may be called only if ->cpo_prep() method was
2112          * executed by osc_io_submit(), that also adds page the to pending list
2113          */
2114         if (!cfs_list_empty(&oap->oap_pending_item)) {
2115                 cfs_list_del_init(&oap->oap_pending_item);
2116                 cfs_list_del_init(&oap->oap_urgent_item);
2117
2118                 loi = oap->oap_loi;
2119                 lop = (oap->oap_cmd & OBD_BRW_WRITE) ?
2120                         &loi->loi_write_lop : &loi->loi_read_lop;
2121                 lop_update_pending(oap->oap_cli, lop, oap->oap_cmd, -1);
2122                 loi_list_maint(oap->oap_cli, oap->oap_loi);
2123                 rc = oap->oap_caller_ops->ap_completion(env,
2124                                           oap->oap_caller_data,
2125                                           oap->oap_cmd, NULL, -EINTR);
2126         }
2127
2128         RETURN(rc);
2129 }
2130
2131 /* this is trying to propogate async writeback errors back up to the
2132  * application.  As an async write fails we record the error code for later if
2133  * the app does an fsync.  As long as errors persist we force future rpcs to be
2134  * sync so that the app can get a sync error and break the cycle of queueing
2135  * pages for which writeback will fail. */
2136 static void osc_process_ar(struct osc_async_rc *ar, __u64 xid,
2137                            int rc)
2138 {
2139         if (rc) {
2140                 if (!ar->ar_rc)
2141                         ar->ar_rc = rc;
2142
2143                 ar->ar_force_sync = 1;
2144                 ar->ar_min_xid = ptlrpc_sample_next_xid();
2145                 return;
2146
2147         }
2148
2149         if (ar->ar_force_sync && (xid >= ar->ar_min_xid))
2150                 ar->ar_force_sync = 0;
2151 }
2152
2153 void osc_oap_to_pending(struct osc_async_page *oap)
2154 {
2155         struct loi_oap_pages *lop;
2156
2157         if (oap->oap_cmd & OBD_BRW_WRITE)
2158                 lop = &oap->oap_loi->loi_write_lop;
2159         else
2160                 lop = &oap->oap_loi->loi_read_lop;
2161
2162         if (oap->oap_async_flags & ASYNC_HP)
2163                 cfs_list_add(&oap->oap_urgent_item, &lop->lop_urgent);
2164         else if (oap->oap_async_flags & ASYNC_URGENT)
2165                 cfs_list_add_tail(&oap->oap_urgent_item, &lop->lop_urgent);
2166         cfs_list_add_tail(&oap->oap_pending_item, &lop->lop_pending);
2167         lop_update_pending(oap->oap_cli, lop, oap->oap_cmd, 1);
2168 }
2169
2170 /* this must be called holding the loi list lock to give coverage to exit_cache,
2171  * async_flag maintenance, and oap_request */
2172 static void osc_ap_completion(const struct lu_env *env,
2173                               struct client_obd *cli, struct obdo *oa,
2174                               struct osc_async_page *oap, int sent, int rc)
2175 {
2176         __u64 xid = 0;
2177
2178         ENTRY;
2179         if (oap->oap_request != NULL) {
2180                 xid = ptlrpc_req_xid(oap->oap_request);
2181                 ptlrpc_req_finished(oap->oap_request);
2182                 oap->oap_request = NULL;
2183         }
2184
2185         cfs_spin_lock(&oap->oap_lock);
2186         oap->oap_async_flags = 0;
2187         cfs_spin_unlock(&oap->oap_lock);
2188         oap->oap_interrupted = 0;
2189
2190         if (oap->oap_cmd & OBD_BRW_WRITE) {
2191                 osc_process_ar(&cli->cl_ar, xid, rc);
2192                 osc_process_ar(&oap->oap_loi->loi_ar, xid, rc);
2193         }
2194
2195         if (rc == 0 && oa != NULL) {
2196                 if (oa->o_valid & OBD_MD_FLBLOCKS)
2197                         oap->oap_loi->loi_lvb.lvb_blocks = oa->o_blocks;
2198                 if (oa->o_valid & OBD_MD_FLMTIME)
2199                         oap->oap_loi->loi_lvb.lvb_mtime = oa->o_mtime;
2200                 if (oa->o_valid & OBD_MD_FLATIME)
2201                         oap->oap_loi->loi_lvb.lvb_atime = oa->o_atime;
2202                 if (oa->o_valid & OBD_MD_FLCTIME)
2203                         oap->oap_loi->loi_lvb.lvb_ctime = oa->o_ctime;
2204         }
2205
2206         rc = oap->oap_caller_ops->ap_completion(env, oap->oap_caller_data,
2207                                                 oap->oap_cmd, oa, rc);
2208
2209         /* ll_ap_completion (from llite) drops PG_locked. so, a new
2210          * I/O on the page could start, but OSC calls it under lock
2211          * and thus we can add oap back to pending safely */
2212         if (rc)
2213                 /* upper layer wants to leave the page on pending queue */
2214                 osc_oap_to_pending(oap);
2215         else
2216                 osc_exit_cache(cli, oap, sent);
2217         EXIT;
2218 }
2219
2220 static int brw_interpret(const struct lu_env *env,
2221                          struct ptlrpc_request *req, void *data, int rc)
2222 {
2223         struct osc_brw_async_args *aa = data;
2224         struct client_obd *cli;
2225         int async;
2226         ENTRY;
2227
2228         rc = osc_brw_fini_request(req, rc);
2229         CDEBUG(D_INODE, "request %p aa %p rc %d\n", req, aa, rc);
2230         if (osc_recoverable_error(rc)) {
2231                 /* Only retry once for mmaped files since the mmaped page
2232                  * might be modified at anytime. We have to retry at least
2233                  * once in case there WAS really a corruption of the page
2234                  * on the network, that was not caused by mmap() modifying
2235                  * the page. Bug11742 */
2236                 if ((rc == -EAGAIN) && (aa->aa_resends > 0) &&
2237                     aa->aa_oa->o_valid & OBD_MD_FLFLAGS &&
2238                     aa->aa_oa->o_flags & OBD_FL_MMAP) {
2239                         rc = 0;
2240                 } else {
2241                         rc = osc_brw_redo_request(req, aa);
2242                         if (rc == 0)
2243                                 RETURN(0);
2244                 }
2245         }
2246
2247         if (aa->aa_ocapa) {
2248                 capa_put(aa->aa_ocapa);
2249                 aa->aa_ocapa = NULL;
2250         }
2251
2252         cli = aa->aa_cli;
2253
2254         client_obd_list_lock(&cli->cl_loi_list_lock);
2255
2256         /* We need to decrement before osc_ap_completion->osc_wake_cache_waiters
2257          * is called so we know whether to go to sync BRWs or wait for more
2258          * RPCs to complete */
2259         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE)
2260                 cli->cl_w_in_flight--;
2261         else
2262                 cli->cl_r_in_flight--;
2263
2264         async = cfs_list_empty(&aa->aa_oaps);
2265         if (!async) { /* from osc_send_oap_rpc() */
2266                 struct osc_async_page *oap, *tmp;
2267                 /* the caller may re-use the oap after the completion call so
2268                  * we need to clean it up a little */
2269                 cfs_list_for_each_entry_safe(oap, tmp, &aa->aa_oaps,
2270                                              oap_rpc_item) {
2271                         cfs_list_del_init(&oap->oap_rpc_item);
2272                         osc_ap_completion(env, cli, aa->aa_oa, oap, 1, rc);
2273                 }
2274                 OBDO_FREE(aa->aa_oa);
2275         } else { /* from async_internal() */
2276                 obd_count i;
2277                 for (i = 0; i < aa->aa_page_count; i++)
2278                         osc_release_write_grant(aa->aa_cli, aa->aa_ppga[i], 1);
2279         }
2280         osc_wake_cache_waiters(cli);
2281         osc_check_rpcs(env, cli);
2282         client_obd_list_unlock(&cli->cl_loi_list_lock);
2283         if (!async)
2284                 cl_req_completion(env, aa->aa_clerq, rc);
2285         osc_release_ppga(aa->aa_ppga, aa->aa_page_count);
2286
2287         RETURN(rc);
2288 }
2289
2290 static struct ptlrpc_request *osc_build_req(const struct lu_env *env,
2291                                             struct client_obd *cli,
2292                                             cfs_list_t *rpc_list,
2293                                             int page_count, int cmd)
2294 {
2295         struct ptlrpc_request *req;
2296         struct brw_page **pga = NULL;
2297         struct osc_brw_async_args *aa;
2298         struct obdo *oa = NULL;
2299         const struct obd_async_page_ops *ops = NULL;
2300         void *caller_data = NULL;
2301         struct osc_async_page *oap;
2302         struct osc_async_page *tmp;
2303         struct ost_body *body;
2304         struct cl_req *clerq = NULL;
2305         enum cl_req_type crt = (cmd & OBD_BRW_WRITE) ? CRT_WRITE : CRT_READ;
2306         struct ldlm_lock *lock = NULL;
2307         struct cl_req_attr crattr;
2308         int i, rc, mpflag = 0;
2309
2310         ENTRY;
2311         LASSERT(!cfs_list_empty(rpc_list));
2312
2313         if (cmd & OBD_BRW_MEMALLOC)
2314                 mpflag = cfs_memory_pressure_get_and_set();
2315
2316         memset(&crattr, 0, sizeof crattr);
2317         OBD_ALLOC(pga, sizeof(*pga) * page_count);
2318         if (pga == NULL)
2319                 GOTO(out, req = ERR_PTR(-ENOMEM));
2320
2321         OBDO_ALLOC(oa);
2322         if (oa == NULL)
2323                 GOTO(out, req = ERR_PTR(-ENOMEM));
2324
2325         i = 0;
2326         cfs_list_for_each_entry(oap, rpc_list, oap_rpc_item) {
2327                 struct cl_page *page = osc_oap2cl_page(oap);
2328                 if (ops == NULL) {
2329                         ops = oap->oap_caller_ops;
2330                         caller_data = oap->oap_caller_data;
2331
2332                         clerq = cl_req_alloc(env, page, crt,
2333                                              1 /* only 1-object rpcs for
2334                                                 * now */);
2335                         if (IS_ERR(clerq))
2336                                 GOTO(out, req = (void *)clerq);
2337                         lock = oap->oap_ldlm_lock;
2338                 }
2339                 pga[i] = &oap->oap_brw_page;
2340                 pga[i]->off = oap->oap_obj_off + oap->oap_page_off;
2341                 CDEBUG(0, "put page %p index %lu oap %p flg %x to pga\n",
2342                        pga[i]->pg, cfs_page_index(oap->oap_page), oap, pga[i]->flag);
2343                 i++;
2344                 cl_req_page_add(env, clerq, page);
2345         }
2346
2347         /* always get the data for the obdo for the rpc */
2348         LASSERT(ops != NULL);
2349         crattr.cra_oa = oa;
2350         crattr.cra_capa = NULL;
2351         cl_req_attr_set(env, clerq, &crattr, ~0ULL);
2352         if (lock) {
2353                 oa->o_handle = lock->l_remote_handle;
2354                 oa->o_valid |= OBD_MD_FLHANDLE;
2355         }
2356
2357         rc = cl_req_prep(env, clerq);
2358         if (rc != 0) {
2359                 CERROR("cl_req_prep failed: %d\n", rc);
2360                 GOTO(out, req = ERR_PTR(rc));
2361         }
2362
2363         sort_brw_pages(pga, page_count);
2364         rc = osc_brw_prep_request(cmd, cli, oa, NULL, page_count,
2365                                   pga, &req, crattr.cra_capa, 1, 0);
2366         if (rc != 0) {
2367                 CERROR("prep_req failed: %d\n", rc);
2368                 GOTO(out, req = ERR_PTR(rc));
2369         }
2370
2371         if (cmd & OBD_BRW_MEMALLOC)
2372                 req->rq_memalloc = 1;
2373
2374         /* Need to update the timestamps after the request is built in case
2375          * we race with setattr (locally or in queue at OST).  If OST gets
2376          * later setattr before earlier BRW (as determined by the request xid),
2377          * the OST will not use BRW timestamps.  Sadly, there is no obvious
2378          * way to do this in a single call.  bug 10150 */
2379         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
2380         cl_req_attr_set(env, clerq, &crattr,
2381                         OBD_MD_FLMTIME|OBD_MD_FLCTIME|OBD_MD_FLATIME);
2382
2383         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
2384         aa = ptlrpc_req_async_args(req);
2385         CFS_INIT_LIST_HEAD(&aa->aa_oaps);
2386         cfs_list_splice(rpc_list, &aa->aa_oaps);
2387         CFS_INIT_LIST_HEAD(rpc_list);
2388         aa->aa_clerq = clerq;
2389 out:
2390         if (cmd & OBD_BRW_MEMALLOC)
2391                 cfs_memory_pressure_restore(mpflag);
2392
2393         capa_put(crattr.cra_capa);
2394         if (IS_ERR(req)) {
2395                 if (oa)
2396                         OBDO_FREE(oa);
2397                 if (pga)
2398                         OBD_FREE(pga, sizeof(*pga) * page_count);
2399                 /* this should happen rarely and is pretty bad, it makes the
2400                  * pending list not follow the dirty order */
2401                 client_obd_list_lock(&cli->cl_loi_list_lock);
2402                 cfs_list_for_each_entry_safe(oap, tmp, rpc_list, oap_rpc_item) {
2403                         cfs_list_del_init(&oap->oap_rpc_item);
2404
2405                         /* queued sync pages can be torn down while the pages
2406                          * were between the pending list and the rpc */
2407                         if (oap->oap_interrupted) {
2408                                 CDEBUG(D_INODE, "oap %p interrupted\n", oap);
2409                                 osc_ap_completion(env, cli, NULL, oap, 0,
2410                                                   oap->oap_count);
2411                                 continue;
2412                         }
2413                         osc_ap_completion(env, cli, NULL, oap, 0, PTR_ERR(req));
2414                 }
2415                 if (clerq && !IS_ERR(clerq))
2416                         cl_req_completion(env, clerq, PTR_ERR(req));
2417         }
2418         RETURN(req);
2419 }
2420
2421 /**
2422  * prepare pages for ASYNC io and put pages in send queue.
2423  *
2424  * \param cmd OBD_BRW_* macroses
2425  * \param lop pending pages
2426  *
2427  * \return zero if no page added to send queue.
2428  * \return 1 if pages successfully added to send queue.
2429  * \return negative on errors.
2430  */
2431 static int
2432 osc_send_oap_rpc(const struct lu_env *env, struct client_obd *cli,
2433                  struct lov_oinfo *loi,
2434                  int cmd, struct loi_oap_pages *lop)
2435 {
2436         struct ptlrpc_request *req;
2437         obd_count page_count = 0;
2438         struct osc_async_page *oap = NULL, *tmp;
2439         struct osc_brw_async_args *aa;
2440         const struct obd_async_page_ops *ops;
2441         CFS_LIST_HEAD(rpc_list);
2442         int srvlock = 0, mem_tight = 0;
2443         struct cl_object *clob = NULL;
2444         obd_off starting_offset = OBD_OBJECT_EOF;
2445         unsigned int ending_offset;
2446         int starting_page_off = 0;
2447         ENTRY;
2448
2449         /* ASYNC_HP pages first. At present, when the lock the pages is
2450          * to be canceled, the pages covered by the lock will be sent out
2451          * with ASYNC_HP. We have to send out them as soon as possible. */
2452         cfs_list_for_each_entry_safe(oap, tmp, &lop->lop_urgent, oap_urgent_item) {
2453                 if (oap->oap_async_flags & ASYNC_HP)
2454                         cfs_list_move(&oap->oap_pending_item, &lop->lop_pending);
2455                 if (++page_count >= cli->cl_max_pages_per_rpc)
2456                         break;
2457         }
2458         page_count = 0;
2459
2460         /* first we find the pages we're allowed to work with */
2461         cfs_list_for_each_entry_safe(oap, tmp, &lop->lop_pending,
2462                                      oap_pending_item) {
2463                 ops = oap->oap_caller_ops;
2464
2465                 LASSERTF(oap->oap_magic == OAP_MAGIC, "Bad oap magic: oap %p, "
2466                          "magic 0x%x\n", oap, oap->oap_magic);
2467
2468                 if (clob == NULL) {
2469                         /* pin object in memory, so that completion call-backs
2470                          * can be safely called under client_obd_list lock. */
2471                         clob = osc_oap2cl_page(oap)->cp_obj;
2472                         cl_object_get(clob);
2473                 }
2474
2475                 if (page_count != 0 &&
2476                     srvlock != !!(oap->oap_brw_flags & OBD_BRW_SRVLOCK)) {
2477                         CDEBUG(D_PAGE, "SRVLOCK flag mismatch,"
2478                                " oap %p, page %p, srvlock %u\n",
2479                                oap, oap->oap_brw_page.pg, (unsigned)!srvlock);
2480                         break;
2481                 }
2482
2483                 /* If there is a gap at the start of this page, it can't merge
2484                  * with any previous page, so we'll hand the network a
2485                  * "fragmented" page array that it can't transfer in 1 RDMA */
2486                 if (oap->oap_obj_off < starting_offset) {
2487                         if (starting_page_off != 0)
2488                                 break;
2489
2490                         starting_page_off = oap->oap_page_off;
2491                         starting_offset = oap->oap_obj_off + starting_page_off;
2492                 } else if (oap->oap_page_off != 0)
2493                         break;
2494
2495                 /* in llite being 'ready' equates to the page being locked
2496                  * until completion unlocks it.  commit_write submits a page
2497                  * as not ready because its unlock will happen unconditionally
2498                  * as the call returns.  if we race with commit_write giving
2499                  * us that page we don't want to create a hole in the page
2500                  * stream, so we stop and leave the rpc to be fired by
2501                  * another dirtier or kupdated interval (the not ready page
2502                  * will still be on the dirty list).  we could call in
2503                  * at the end of ll_file_write to process the queue again. */
2504                 if (!(oap->oap_async_flags & ASYNC_READY)) {
2505                         int rc = ops->ap_make_ready(env, oap->oap_caller_data,
2506                                                     cmd);
2507                         if (rc < 0)
2508                                 CDEBUG(D_INODE, "oap %p page %p returned %d "
2509                                                 "instead of ready\n", oap,
2510                                                 oap->oap_page, rc);
2511                         switch (rc) {
2512                         case -EAGAIN:
2513                                 /* llite is telling us that the page is still
2514                                  * in commit_write and that we should try
2515                                  * and put it in an rpc again later.  we
2516                                  * break out of the loop so we don't create
2517                                  * a hole in the sequence of pages in the rpc
2518                                  * stream.*/
2519                                 oap = NULL;
2520                                 break;
2521                         case -EINTR:
2522                                 /* the io isn't needed.. tell the checks
2523                                  * below to complete the rpc with EINTR */
2524                                 cfs_spin_lock(&oap->oap_lock);
2525                                 oap->oap_async_flags |= ASYNC_COUNT_STABLE;
2526                                 cfs_spin_unlock(&oap->oap_lock);
2527                                 oap->oap_count = -EINTR;
2528                                 break;
2529                         case 0:
2530                                 cfs_spin_lock(&oap->oap_lock);
2531                                 oap->oap_async_flags |= ASYNC_READY;
2532                                 cfs_spin_unlock(&oap->oap_lock);
2533                                 break;
2534                         default:
2535                                 LASSERTF(0, "oap %p page %p returned %d "
2536                                             "from make_ready\n", oap,
2537                                             oap->oap_page, rc);
2538                                 break;
2539                         }
2540                 }
2541                 if (oap == NULL)
2542                         break;
2543                 /*
2544                  * Page submitted for IO has to be locked. Either by
2545                  * ->ap_make_ready() or by higher layers.
2546                  */
2547 #if defined(__KERNEL__) && defined(__linux__)
2548                 {
2549                         struct cl_page *page;
2550
2551                         page = osc_oap2cl_page(oap);
2552
2553                         if (page->cp_type == CPT_CACHEABLE &&
2554                             !(PageLocked(oap->oap_page) &&
2555                               (CheckWriteback(oap->oap_page, cmd)))) {
2556                                 CDEBUG(D_PAGE, "page %p lost wb %lx/%x\n",
2557                                        oap->oap_page,
2558                                        (long)oap->oap_page->flags,
2559                                        oap->oap_async_flags);
2560                                 LBUG();
2561                         }
2562                 }
2563 #endif
2564
2565                 /* take the page out of our book-keeping */
2566                 cfs_list_del_init(&oap->oap_pending_item);
2567                 lop_update_pending(cli, lop, cmd, -1);
2568                 cfs_list_del_init(&oap->oap_urgent_item);
2569
2570                 /* ask the caller for the size of the io as the rpc leaves. */
2571                 if (!(oap->oap_async_flags & ASYNC_COUNT_STABLE)) {
2572                         oap->oap_count =
2573                                 ops->ap_refresh_count(env, oap->oap_caller_data,
2574                                                       cmd);
2575                         LASSERT(oap->oap_page_off + oap->oap_count <= CFS_PAGE_SIZE);
2576                 }
2577                 if (oap->oap_count <= 0) {
2578                         CDEBUG(D_CACHE, "oap %p count %d, completing\n", oap,
2579                                oap->oap_count);
2580                         osc_ap_completion(env, cli, NULL,
2581                                           oap, 0, oap->oap_count);
2582                         continue;
2583                 }
2584
2585                 /* now put the page back in our accounting */
2586                 cfs_list_add_tail(&oap->oap_rpc_item, &rpc_list);
2587                 if (page_count++ == 0) {
2588                         srvlock = !!(oap->oap_brw_flags & OBD_BRW_SRVLOCK);
2589                         starting_offset = (oap->oap_obj_off+oap->oap_page_off) &
2590                                           (PTLRPC_MAX_BRW_SIZE - 1);
2591                 }
2592
2593                 if (oap->oap_brw_flags & OBD_BRW_MEMALLOC)
2594                         mem_tight = 1;
2595
2596                 /* End on a PTLRPC_MAX_BRW_SIZE boundary.  We want full-sized
2597                  * RPCs aligned on PTLRPC_MAX_BRW_SIZE boundaries to help reads
2598                  * have the same alignment as the initial writes that allocated
2599                  * extents on the server. */
2600                 ending_offset = oap->oap_obj_off + oap->oap_page_off +
2601                                 oap->oap_count;
2602                 if (!(ending_offset & (PTLRPC_MAX_BRW_SIZE - 1)))
2603                         break;
2604
2605                 if (page_count >= cli->cl_max_pages_per_rpc)
2606                         break;
2607
2608                 /* If there is a gap at the end of this page, it can't merge
2609                  * with any subsequent pages, so we'll hand the network a
2610                  * "fragmented" page array that it can't transfer in 1 RDMA */
2611                 if (oap->oap_page_off + oap->oap_count < CFS_PAGE_SIZE)
2612                         break;
2613         }
2614
2615         osc_wake_cache_waiters(cli);
2616
2617         loi_list_maint(cli, loi);
2618
2619         client_obd_list_unlock(&cli->cl_loi_list_lock);
2620
2621         if (clob != NULL)
2622                 cl_object_put(env, clob);
2623
2624         if (page_count == 0) {
2625                 client_obd_list_lock(&cli->cl_loi_list_lock);
2626                 RETURN(0);
2627         }
2628
2629         req = osc_build_req(env, cli, &rpc_list, page_count,
2630                             mem_tight ? (cmd | OBD_BRW_MEMALLOC) : cmd);
2631         if (IS_ERR(req)) {
2632                 LASSERT(cfs_list_empty(&rpc_list));
2633                 loi_list_maint(cli, loi);
2634                 RETURN(PTR_ERR(req));
2635         }
2636
2637         aa = ptlrpc_req_async_args(req);
2638
2639         starting_offset &= PTLRPC_MAX_BRW_SIZE - 1;
2640         if (cmd == OBD_BRW_READ) {
2641                 lprocfs_oh_tally_log2(&cli->cl_read_page_hist, page_count);
2642                 lprocfs_oh_tally(&cli->cl_read_rpc_hist, cli->cl_r_in_flight);
2643                 lprocfs_oh_tally_log2(&cli->cl_read_offset_hist,
2644                                       (starting_offset >> CFS_PAGE_SHIFT) + 1);
2645         } else {
2646                 lprocfs_oh_tally_log2(&cli->cl_write_page_hist, page_count);
2647                 lprocfs_oh_tally(&cli->cl_write_rpc_hist,
2648                                  cli->cl_w_in_flight);
2649                 lprocfs_oh_tally_log2(&cli->cl_write_offset_hist,
2650                                       (starting_offset >> CFS_PAGE_SHIFT) + 1);
2651         }
2652         ptlrpc_lprocfs_brw(req, aa->aa_requested_nob);
2653
2654         client_obd_list_lock(&cli->cl_loi_list_lock);
2655
2656         if (cmd == OBD_BRW_READ)
2657                 cli->cl_r_in_flight++;
2658         else
2659                 cli->cl_w_in_flight++;
2660
2661         /* queued sync pages can be torn down while the pages
2662          * were between the pending list and the rpc */
2663         tmp = NULL;
2664         cfs_list_for_each_entry(oap, &aa->aa_oaps, oap_rpc_item) {
2665                 /* only one oap gets a request reference */
2666                 if (tmp == NULL)
2667                         tmp = oap;
2668                 if (oap->oap_interrupted && !req->rq_intr) {
2669                         CDEBUG(D_INODE, "oap %p in req %p interrupted\n",
2670                                oap, req);
2671                         ptlrpc_mark_interrupted(req);
2672                 }
2673         }
2674         if (tmp != NULL)
2675                 tmp->oap_request = ptlrpc_request_addref(req);
2676
2677         DEBUG_REQ(D_INODE, req, "%d pages, aa %p. now %dr/%dw in flight",
2678                   page_count, aa, cli->cl_r_in_flight, cli->cl_w_in_flight);
2679
2680         req->rq_interpret_reply = brw_interpret;
2681         ptlrpcd_add_req(req, PSCOPE_BRW);
2682         RETURN(1);
2683 }
2684
2685 #define LOI_DEBUG(LOI, STR, args...)                                     \
2686         CDEBUG(D_INODE, "loi ready %d wr %d:%d rd %d:%d " STR,           \
2687                !cfs_list_empty(&(LOI)->loi_ready_item) ||                \
2688                !cfs_list_empty(&(LOI)->loi_hp_ready_item),               \
2689                (LOI)->loi_write_lop.lop_num_pending,                     \
2690                !cfs_list_empty(&(LOI)->loi_write_lop.lop_urgent),        \
2691                (LOI)->loi_read_lop.lop_num_pending,                      \
2692                !cfs_list_empty(&(LOI)->loi_read_lop.lop_urgent),         \
2693                args)                                                     \
2694
2695 /* This is called by osc_check_rpcs() to find which objects have pages that
2696  * we could be sending.  These lists are maintained by lop_makes_rpc(). */
2697 struct lov_oinfo *osc_next_loi(struct client_obd *cli)
2698 {
2699         ENTRY;
2700
2701         /* First return objects that have blocked locks so that they
2702          * will be flushed quickly and other clients can get the lock,
2703          * then objects which have pages ready to be stuffed into RPCs */
2704         if (!cfs_list_empty(&cli->cl_loi_hp_ready_list))
2705                 RETURN(cfs_list_entry(cli->cl_loi_hp_ready_list.next,
2706                                       struct lov_oinfo, loi_hp_ready_item));
2707         if (!cfs_list_empty(&cli->cl_loi_ready_list))
2708                 RETURN(cfs_list_entry(cli->cl_loi_ready_list.next,
2709                                       struct lov_oinfo, loi_ready_item));
2710
2711         /* then if we have cache waiters, return all objects with queued
2712          * writes.  This is especially important when many small files
2713          * have filled up the cache and not been fired into rpcs because
2714          * they don't pass the nr_pending/object threshhold */
2715         if (!cfs_list_empty(&cli->cl_cache_waiters) &&
2716             !cfs_list_empty(&cli->cl_loi_write_list))
2717                 RETURN(cfs_list_entry(cli->cl_loi_write_list.next,
2718                                       struct lov_oinfo, loi_write_item));
2719
2720         /* then return all queued objects when we have an invalid import
2721          * so that they get flushed */
2722         if (cli->cl_import == NULL || cli->cl_import->imp_invalid) {
2723                 if (!cfs_list_empty(&cli->cl_loi_write_list))
2724                         RETURN(cfs_list_entry(cli->cl_loi_write_list.next,
2725                                               struct lov_oinfo,
2726                                               loi_write_item));
2727                 if (!cfs_list_empty(&cli->cl_loi_read_list))
2728                         RETURN(cfs_list_entry(cli->cl_loi_read_list.next,
2729                                               struct lov_oinfo, loi_read_item));
2730         }
2731         RETURN(NULL);
2732 }
2733
2734 static int osc_max_rpc_in_flight(struct client_obd *cli, struct lov_oinfo *loi)
2735 {
2736         struct osc_async_page *oap;
2737         int hprpc = 0;
2738
2739         if (!cfs_list_empty(&loi->loi_write_lop.lop_urgent)) {
2740                 oap = cfs_list_entry(loi->loi_write_lop.lop_urgent.next,
2741                                      struct osc_async_page, oap_urgent_item);
2742                 hprpc = !!(oap->oap_async_flags & ASYNC_HP);
2743         }
2744
2745         if (!hprpc && !cfs_list_empty(&loi->loi_read_lop.lop_urgent)) {
2746                 oap = cfs_list_entry(loi->loi_read_lop.lop_urgent.next,
2747                                      struct osc_async_page, oap_urgent_item);
2748                 hprpc = !!(oap->oap_async_flags & ASYNC_HP);
2749         }
2750
2751         return rpcs_in_flight(cli) >= cli->cl_max_rpcs_in_flight + hprpc;
2752 }
2753
2754 /* called with the loi list lock held */
2755 void osc_check_rpcs(const struct lu_env *env, struct client_obd *cli)
2756 {
2757         struct lov_oinfo *loi;
2758         int rc = 0, race_counter = 0;
2759         ENTRY;
2760
2761         while ((loi = osc_next_loi(cli)) != NULL) {
2762                 LOI_DEBUG(loi, "%lu in flight\n", rpcs_in_flight(cli));
2763
2764                 if (osc_max_rpc_in_flight(cli, loi))
2765                         break;
2766
2767                 /* attempt some read/write balancing by alternating between
2768                  * reads and writes in an object.  The makes_rpc checks here
2769                  * would be redundant if we were getting read/write work items
2770                  * instead of objects.  we don't want send_oap_rpc to drain a
2771                  * partial read pending queue when we're given this object to
2772                  * do io on writes while there are cache waiters */
2773                 if (lop_makes_rpc(cli, &loi->loi_write_lop, OBD_BRW_WRITE)) {
2774                         rc = osc_send_oap_rpc(env, cli, loi, OBD_BRW_WRITE,
2775                                               &loi->loi_write_lop);
2776                         if (rc < 0) {
2777                                 CERROR("Write request failed with %d\n", rc);
2778
2779                                 /* osc_send_oap_rpc failed, mostly because of
2780                                  * memory pressure.
2781                                  *
2782                                  * It can't break here, because if:
2783                                  *  - a page was submitted by osc_io_submit, so
2784                                  *    page locked;
2785                                  *  - no request in flight
2786                                  *  - no subsequent request
2787                                  * The system will be in live-lock state,
2788                                  * because there is no chance to call
2789                                  * osc_io_unplug() and osc_check_rpcs() any
2790                                  * more. pdflush can't help in this case,
2791                                  * because it might be blocked at grabbing
2792                                  * the page lock as we mentioned.
2793                                  *
2794                                  * Anyway, continue to drain pages. */
2795                                 /* break; */
2796                         }
2797
2798                         if (rc > 0)
2799                                 race_counter = 0;
2800                         else if (rc == 0)
2801                                 race_counter++;
2802                 }
2803                 if (lop_makes_rpc(cli, &loi->loi_read_lop, OBD_BRW_READ)) {
2804                         rc = osc_send_oap_rpc(env, cli, loi, OBD_BRW_READ,
2805                                               &loi->loi_read_lop);
2806                         if (rc < 0)
2807                                 CERROR("Read request failed with %d\n", rc);
2808
2809                         if (rc > 0)
2810                                 race_counter = 0;
2811                         else if (rc == 0)
2812                                 race_counter++;
2813                 }
2814
2815                 /* attempt some inter-object balancing by issuing rpcs
2816                  * for each object in turn */
2817                 if (!cfs_list_empty(&loi->loi_hp_ready_item))
2818                         cfs_list_del_init(&loi->loi_hp_ready_item);
2819                 if (!cfs_list_empty(&loi->loi_ready_item))
2820                         cfs_list_del_init(&loi->loi_ready_item);
2821                 if (!cfs_list_empty(&loi->loi_write_item))
2822                         cfs_list_del_init(&loi->loi_write_item);
2823                 if (!cfs_list_empty(&loi->loi_read_item))
2824                         cfs_list_del_init(&loi->loi_read_item);
2825
2826                 loi_list_maint(cli, loi);
2827
2828                 /* send_oap_rpc fails with 0 when make_ready tells it to
2829                  * back off.  llite's make_ready does this when it tries
2830                  * to lock a page queued for write that is already locked.
2831                  * we want to try sending rpcs from many objects, but we
2832                  * don't want to spin failing with 0.  */
2833                 if (race_counter == 10)
2834                         break;
2835         }
2836         EXIT;
2837 }
2838
2839 /* we're trying to queue a page in the osc so we're subject to the
2840  * 'cl_dirty_max' limit on the number of pages that can be queued in the osc.
2841  * If the osc's queued pages are already at that limit, then we want to sleep
2842  * until there is space in the osc's queue for us.  We also may be waiting for
2843  * write credits from the OST if there are RPCs in flight that may return some
2844  * before we fall back to sync writes.
2845  *
2846  * We need this know our allocation was granted in the presence of signals */
2847 static int ocw_granted(struct client_obd *cli, struct osc_cache_waiter *ocw)
2848 {
2849         int rc;
2850         ENTRY;
2851         client_obd_list_lock(&cli->cl_loi_list_lock);
2852         rc = cfs_list_empty(&ocw->ocw_entry) || rpcs_in_flight(cli) == 0;
2853         client_obd_list_unlock(&cli->cl_loi_list_lock);
2854         RETURN(rc);
2855 };
2856
2857 /**
2858  * Non-blocking version of osc_enter_cache() that consumes grant only when it
2859  * is available.
2860  */
2861 int osc_enter_cache_try(const struct lu_env *env,
2862                         struct client_obd *cli, struct lov_oinfo *loi,
2863                         struct osc_async_page *oap, int transient)
2864 {
2865         int has_grant;
2866
2867         has_grant = cli->cl_avail_grant >= CFS_PAGE_SIZE;
2868         if (has_grant) {
2869                 osc_consume_write_grant(cli, &oap->oap_brw_page);
2870                 if (transient) {
2871                         cli->cl_dirty_transit += CFS_PAGE_SIZE;
2872                         cfs_atomic_inc(&obd_dirty_transit_pages);
2873                         oap->oap_brw_flags |= OBD_BRW_NOCACHE;
2874                 }
2875         }
2876         return has_grant;
2877 }
2878
2879 /* Caller must hold loi_list_lock - we drop/regain it if we need to wait for
2880  * grant or cache space. */
2881 static int osc_enter_cache(const struct lu_env *env,
2882                            struct client_obd *cli, struct lov_oinfo *loi,
2883                            struct osc_async_page *oap)
2884 {
2885         struct osc_cache_waiter ocw;
2886         struct l_wait_info lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP, NULL);
2887
2888         ENTRY;
2889
2890         CDEBUG(D_CACHE, "dirty: %ld/%d dirty_max: %ld/%d dropped: %lu "
2891                "grant: %lu\n", cli->cl_dirty, cfs_atomic_read(&obd_dirty_pages),
2892                cli->cl_dirty_max, obd_max_dirty_pages,
2893                cli->cl_lost_grant, cli->cl_avail_grant);
2894
2895         /* force the caller to try sync io.  this can jump the list
2896          * of queued writes and create a discontiguous rpc stream */
2897         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_NO_GRANT) ||
2898             cli->cl_dirty_max < CFS_PAGE_SIZE     ||
2899             cli->cl_ar.ar_force_sync || loi->loi_ar.ar_force_sync)
2900                 RETURN(-EDQUOT);
2901
2902         /* Hopefully normal case - cache space and write credits available */
2903         if (cli->cl_dirty + CFS_PAGE_SIZE <= cli->cl_dirty_max &&
2904             cfs_atomic_read(&obd_dirty_pages) + 1 <= obd_max_dirty_pages &&
2905             osc_enter_cache_try(env, cli, loi, oap, 0))
2906                 RETURN(0);
2907
2908         /* It is safe to block as a cache waiter as long as there is grant
2909          * space available or the hope of additional grant being returned
2910          * when an in flight write completes.  Using the write back cache
2911          * if possible is preferable to sending the data synchronously
2912          * because write pages can then be merged in to large requests.
2913          * The addition of this cache waiter will causing pending write
2914          * pages to be sent immediately. */
2915         if (cli->cl_w_in_flight || cli->cl_avail_grant >= CFS_PAGE_SIZE) {
2916                 cfs_list_add_tail(&ocw.ocw_entry, &cli->cl_cache_waiters);
2917                 cfs_waitq_init(&ocw.ocw_waitq);
2918                 ocw.ocw_oap = oap;
2919                 ocw.ocw_rc = 0;
2920
2921                 loi_list_maint(cli, loi);
2922                 osc_check_rpcs(env, cli);
2923                 client_obd_list_unlock(&cli->cl_loi_list_lock);
2924
2925                 CDEBUG(D_CACHE, "sleeping for cache space\n");
2926                 l_wait_event(ocw.ocw_waitq, ocw_granted(cli, &ocw), &lwi);
2927
2928                 client_obd_list_lock(&cli->cl_loi_list_lock);
2929                 if (!cfs_list_empty(&ocw.ocw_entry)) {
2930                         cfs_list_del(&ocw.ocw_entry);
2931                         RETURN(-EINTR);
2932                 }
2933                 RETURN(ocw.ocw_rc);
2934         }
2935
2936         RETURN(-EDQUOT);
2937 }
2938
2939
2940 int osc_prep_async_page(struct obd_export *exp, struct lov_stripe_md *lsm,
2941                         struct lov_oinfo *loi, cfs_page_t *page,
2942                         obd_off offset, const struct obd_async_page_ops *ops,
2943                         void *data, void **res, int nocache,
2944                         struct lustre_handle *lockh)
2945 {
2946         struct osc_async_page *oap;
2947
2948         ENTRY;
2949
2950         if (!page)
2951                 return cfs_size_round(sizeof(*oap));
2952
2953         oap = *res;
2954         oap->oap_magic = OAP_MAGIC;
2955         oap->oap_cli = &exp->exp_obd->u.cli;
2956         oap->oap_loi = loi;
2957
2958         oap->oap_caller_ops = ops;
2959         oap->oap_caller_data = data;
2960
2961         oap->oap_page = page;
2962         oap->oap_obj_off = offset;
2963         if (!client_is_remote(exp) &&
2964             cfs_capable(CFS_CAP_SYS_RESOURCE))
2965                 oap->oap_brw_flags = OBD_BRW_NOQUOTA;
2966
2967         LASSERT(!(offset & ~CFS_PAGE_MASK));
2968
2969         CFS_INIT_LIST_HEAD(&oap->oap_pending_item);
2970         CFS_INIT_LIST_HEAD(&oap->oap_urgent_item);
2971         CFS_INIT_LIST_HEAD(&oap->oap_rpc_item);
2972         CFS_INIT_LIST_HEAD(&oap->oap_page_list);
2973
2974         cfs_spin_lock_init(&oap->oap_lock);
2975         CDEBUG(D_CACHE, "oap %p page %p obj off "LPU64"\n", oap, page, offset);
2976         RETURN(0);
2977 }
2978
2979 int osc_queue_async_io(const struct lu_env *env, struct obd_export *exp,
2980                        struct lov_stripe_md *lsm, struct lov_oinfo *loi,
2981                        struct osc_async_page *oap, int cmd, int off,
2982                        int count, obd_flag brw_flags, enum async_flags async_flags)
2983 {
2984         struct client_obd *cli = &exp->exp_obd->u.cli;
2985         int rc = 0;
2986         ENTRY;
2987
2988         if (oap->oap_magic != OAP_MAGIC)
2989                 RETURN(-EINVAL);
2990
2991         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
2992                 RETURN(-EIO);
2993
2994         if (!cfs_list_empty(&oap->oap_pending_item) ||
2995             !cfs_list_empty(&oap->oap_urgent_item) ||
2996             !cfs_list_empty(&oap->oap_rpc_item))
2997                 RETURN(-EBUSY);
2998
2999         /* check if the file's owner/group is over quota */
3000         if ((cmd & OBD_BRW_WRITE) && !(cmd & OBD_BRW_NOQUOTA)) {
3001                 struct cl_object *obj;
3002                 struct cl_attr    attr; /* XXX put attr into thread info */
3003                 unsigned int qid[MAXQUOTAS];
3004
3005                 obj = cl_object_top(osc_oap2cl_page(oap)->cp_obj);
3006
3007                 cl_object_attr_lock(obj);
3008                 rc = cl_object_attr_get(env, obj, &attr);
3009                 cl_object_attr_unlock(obj);
3010
3011                 qid[USRQUOTA] = attr.cat_uid;
3012                 qid[GRPQUOTA] = attr.cat_gid;
3013                 if (rc == 0 &&
3014                     lquota_chkdq(quota_interface, cli, qid) == NO_QUOTA)
3015                         rc = -EDQUOT;
3016                 if (rc)
3017                         RETURN(rc);
3018         }
3019
3020         if (loi == NULL)
3021                 loi = lsm->lsm_oinfo[0];
3022
3023         client_obd_list_lock(&cli->cl_loi_list_lock);
3024
3025         LASSERT(off + count <= CFS_PAGE_SIZE);
3026         oap->oap_cmd = cmd;
3027         oap->oap_page_off = off;
3028         oap->oap_count = count;
3029         oap->oap_brw_flags = brw_flags;
3030         /* Give a hint to OST that requests are coming from kswapd - bug19529 */
3031         if (cfs_memory_pressure_get())
3032                 oap->oap_brw_flags |= OBD_BRW_MEMALLOC;
3033         cfs_spin_lock(&oap->oap_lock);
3034         oap->oap_async_flags = async_flags;
3035         cfs_spin_unlock(&oap->oap_lock);
3036
3037         if (cmd & OBD_BRW_WRITE) {
3038                 rc = osc_enter_cache(env, cli, loi, oap);
3039                 if (rc) {
3040                         client_obd_list_unlock(&cli->cl_loi_list_lock);
3041                         RETURN(rc);
3042                 }
3043         }
3044
3045         osc_oap_to_pending(oap);
3046         loi_list_maint(cli, loi);
3047
3048         LOI_DEBUG(loi, "oap %p page %p added for cmd %d\n", oap, oap->oap_page,
3049                   cmd);
3050
3051         osc_check_rpcs(env, cli);
3052         client_obd_list_unlock(&cli->cl_loi_list_lock);
3053
3054         RETURN(0);
3055 }
3056
3057 /* aka (~was & now & flag), but this is more clear :) */
3058 #define SETTING(was, now, flag) (!(was & flag) && (now & flag))
3059
3060 int osc_set_async_flags_base(struct client_obd *cli,
3061                              struct lov_oinfo *loi, struct osc_async_page *oap,
3062                              obd_flag async_flags)
3063 {
3064         struct loi_oap_pages *lop;
3065         int flags = 0;
3066         ENTRY;
3067
3068         LASSERT(!cfs_list_empty(&oap->oap_pending_item));
3069
3070         if (oap->oap_cmd & OBD_BRW_WRITE) {
3071                 lop = &loi->loi_write_lop;
3072         } else {
3073                 lop = &loi->loi_read_lop;
3074         }
3075
3076         if ((oap->oap_async_flags & async_flags) == async_flags)
3077                 RETURN(0);
3078
3079         if (SETTING(oap->oap_async_flags, async_flags, ASYNC_READY))
3080                 flags |= ASYNC_READY;
3081
3082         if (SETTING(oap->oap_async_flags, async_flags, ASYNC_URGENT) &&
3083             cfs_list_empty(&oap->oap_rpc_item)) {
3084                 if (oap->oap_async_flags & ASYNC_HP)
3085                         cfs_list_add(&oap->oap_urgent_item, &lop->lop_urgent);
3086                 else
3087                         cfs_list_add_tail(&oap->oap_urgent_item,
3088                                           &lop->lop_urgent);
3089                 flags |= ASYNC_URGENT;
3090                 loi_list_maint(cli, loi);
3091         }
3092         cfs_spin_lock(&oap->oap_lock);
3093         oap->oap_async_flags |= flags;
3094         cfs_spin_unlock(&oap->oap_lock);
3095
3096         LOI_DEBUG(loi, "oap %p page %p has flags %x\n", oap, oap->oap_page,
3097                         oap->oap_async_flags);
3098         RETURN(0);
3099 }
3100
3101 int osc_teardown_async_page(struct obd_export *exp, struct lov_stripe_md *lsm,
3102                             struct lov_oinfo *loi, struct osc_async_page *oap)
3103 {
3104         struct client_obd *cli = &exp->exp_obd->u.cli;
3105         struct loi_oap_pages *lop;
3106         int rc = 0;
3107         ENTRY;
3108
3109         if (oap->oap_magic != OAP_MAGIC)
3110                 RETURN(-EINVAL);
3111
3112         if (loi == NULL)
3113                 loi = lsm->lsm_oinfo[0];
3114
3115         if (oap->oap_cmd & OBD_BRW_WRITE) {
3116                 lop = &loi->loi_write_lop;
3117         } else {
3118                 lop = &loi->loi_read_lop;
3119         }
3120
3121         client_obd_list_lock(&cli->cl_loi_list_lock);
3122
3123         if (!cfs_list_empty(&oap->oap_rpc_item))
3124                 GOTO(out, rc = -EBUSY);
3125
3126         osc_exit_cache(cli, oap, 0);
3127         osc_wake_cache_waiters(cli);
3128
3129         if (!cfs_list_empty(&oap->oap_urgent_item)) {
3130                 cfs_list_del_init(&oap->oap_urgent_item);
3131                 cfs_spin_lock(&oap->oap_lock);
3132                 oap->oap_async_flags &= ~(ASYNC_URGENT | ASYNC_HP);
3133                 cfs_spin_unlock(&oap->oap_lock);
3134         }
3135         if (!cfs_list_empty(&oap->oap_pending_item)) {
3136                 cfs_list_del_init(&oap->oap_pending_item);
3137                 lop_update_pending(cli, lop, oap->oap_cmd, -1);
3138         }
3139         loi_list_maint(cli, loi);
3140         LOI_DEBUG(loi, "oap %p page %p torn down\n", oap, oap->oap_page);
3141 out:
3142         client_obd_list_unlock(&cli->cl_loi_list_lock);
3143         RETURN(rc);
3144 }
3145
3146 static int osc_set_lock_data_with_check(struct ldlm_lock *lock,
3147                                         struct ldlm_enqueue_info *einfo)
3148 {
3149         void *data = einfo->ei_cbdata;
3150         int set = 0;
3151
3152         LASSERT(lock != NULL);
3153         LASSERT(lock->l_blocking_ast == einfo->ei_cb_bl);
3154         LASSERT(lock->l_resource->lr_type == einfo->ei_type);
3155         LASSERT(lock->l_completion_ast == einfo->ei_cb_cp);
3156         LASSERT(lock->l_glimpse_ast == einfo->ei_cb_gl);
3157
3158         lock_res_and_lock(lock);
3159         cfs_spin_lock(&osc_ast_guard);
3160
3161         if (lock->l_ast_data == NULL)
3162                 lock->l_ast_data = data;
3163         if (lock->l_ast_data == data)
3164                 set = 1;
3165
3166         cfs_spin_unlock(&osc_ast_guard);
3167         unlock_res_and_lock(lock);
3168
3169         return set;
3170 }
3171
3172 static int osc_set_data_with_check(struct lustre_handle *lockh,
3173                                    struct ldlm_enqueue_info *einfo)
3174 {
3175         struct ldlm_lock *lock = ldlm_handle2lock(lockh);
3176         int set = 0;
3177
3178         if (lock != NULL) {
3179                 set = osc_set_lock_data_with_check(lock, einfo);
3180                 LDLM_LOCK_PUT(lock);
3181         } else
3182                 CERROR("lockh %p, data %p - client evicted?\n",
3183                        lockh, einfo->ei_cbdata);
3184         return set;
3185 }
3186
3187 static int osc_change_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
3188                              ldlm_iterator_t replace, void *data)
3189 {
3190         struct ldlm_res_id res_id;
3191         struct obd_device *obd = class_exp2obd(exp);
3192
3193         osc_build_res_name(lsm->lsm_object_id, lsm->lsm_object_seq, &res_id);
3194         ldlm_resource_iterate(obd->obd_namespace, &res_id, replace, data);
3195         return 0;
3196 }
3197
3198 /* find any ldlm lock of the inode in osc
3199  * return 0    not find
3200  *        1    find one
3201  *      < 0    error */
3202 static int osc_find_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
3203                            ldlm_iterator_t replace, void *data)
3204 {
3205         struct ldlm_res_id res_id;
3206         struct obd_device *obd = class_exp2obd(exp);
3207         int rc = 0;
3208
3209         osc_build_res_name(lsm->lsm_object_id, lsm->lsm_object_seq, &res_id);
3210         rc = ldlm_resource_iterate(obd->obd_namespace, &res_id, replace, data);
3211         if (rc == LDLM_ITER_STOP)
3212                 return(1);
3213         if (rc == LDLM_ITER_CONTINUE)
3214                 return(0);
3215         return(rc);
3216 }
3217
3218 static int osc_enqueue_fini(struct ptlrpc_request *req, struct ost_lvb *lvb,
3219                             obd_enqueue_update_f upcall, void *cookie,
3220                             int *flags, int rc)
3221 {
3222         int intent = *flags & LDLM_FL_HAS_INTENT;
3223         ENTRY;
3224
3225         if (intent) {
3226                 /* The request was created before ldlm_cli_enqueue call. */
3227                 if (rc == ELDLM_LOCK_ABORTED) {
3228                         struct ldlm_reply *rep;
3229                         rep = req_capsule_server_get(&req->rq_pill,
3230                                                      &RMF_DLM_REP);
3231
3232                         LASSERT(rep != NULL);
3233                         if (rep->lock_policy_res1)
3234                                 rc = rep->lock_policy_res1;
3235                 }
3236         }
3237
3238         if ((intent && rc == ELDLM_LOCK_ABORTED) || !rc) {
3239                 *flags |= LDLM_FL_LVB_READY;
3240                 CDEBUG(D_INODE,"got kms "LPU64" blocks "LPU64" mtime "LPU64"\n",
3241                        lvb->lvb_size, lvb->lvb_blocks, lvb->lvb_mtime);
3242         }
3243
3244         /* Call the update callback. */
3245         rc = (*upcall)(cookie, rc);
3246         RETURN(rc);
3247 }
3248
3249 static int osc_enqueue_interpret(const struct lu_env *env,
3250                                  struct ptlrpc_request *req,
3251                                  struct osc_enqueue_args *aa, int rc)
3252 {
3253         struct ldlm_lock *lock;
3254         struct lustre_handle handle;
3255         __u32 mode;
3256
3257         /* Make a local copy of a lock handle and a mode, because aa->oa_*
3258          * might be freed anytime after lock upcall has been called. */
3259         lustre_handle_copy(&handle, aa->oa_lockh);
3260         mode = aa->oa_ei->ei_mode;
3261
3262         /* ldlm_cli_enqueue is holding a reference on the lock, so it must
3263          * be valid. */
3264         lock = ldlm_handle2lock(&handle);
3265
3266         /* Take an additional reference so that a blocking AST that
3267          * ldlm_cli_enqueue_fini() might post for a failed lock, is guaranteed
3268          * to arrive after an upcall has been executed by
3269          * osc_enqueue_fini(). */
3270         ldlm_lock_addref(&handle, mode);
3271
3272         /* Let CP AST to grant the lock first. */
3273         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_ENQ_RACE, 1);
3274
3275         /* Complete obtaining the lock procedure. */
3276         rc = ldlm_cli_enqueue_fini(aa->oa_exp, req, aa->oa_ei->ei_type, 1,
3277                                    mode, aa->oa_flags, aa->oa_lvb,
3278                                    sizeof(*aa->oa_lvb), &handle, rc);
3279         /* Complete osc stuff. */
3280         rc = osc_enqueue_fini(req, aa->oa_lvb,
3281                               aa->oa_upcall, aa->oa_cookie, aa->oa_flags, rc);
3282
3283         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_CANCEL_RACE, 10);
3284
3285         /* Release the lock for async request. */
3286         if (lustre_handle_is_used(&handle) && rc == ELDLM_OK)
3287                 /*
3288                  * Releases a reference taken by ldlm_cli_enqueue(), if it is
3289                  * not already released by
3290                  * ldlm_cli_enqueue_fini()->failed_lock_cleanup()
3291                  */
3292                 ldlm_lock_decref(&handle, mode);
3293
3294         LASSERTF(lock != NULL, "lockh %p, req %p, aa %p - client evicted?\n",
3295                  aa->oa_lockh, req, aa);
3296         ldlm_lock_decref(&handle, mode);
3297         LDLM_LOCK_PUT(lock);
3298         return rc;
3299 }
3300
3301 void osc_update_enqueue(struct lustre_handle *lov_lockhp,
3302                         struct lov_oinfo *loi, int flags,
3303                         struct ost_lvb *lvb, __u32 mode, int rc)
3304 {
3305         if (rc == ELDLM_OK) {
3306                 struct ldlm_lock *lock = ldlm_handle2lock(lov_lockhp);
3307                 __u64 tmp;
3308
3309                 LASSERT(lock != NULL);
3310                 loi->loi_lvb = *lvb;
3311                 tmp = loi->loi_lvb.lvb_size;
3312                 /* Extend KMS up to the end of this lock and no further
3313                  * A lock on [x,y] means a KMS of up to y + 1 bytes! */
3314                 if (tmp > lock->l_policy_data.l_extent.end)
3315                         tmp = lock->l_policy_data.l_extent.end + 1;
3316                 if (tmp >= loi->loi_kms) {
3317                         LDLM_DEBUG(lock, "lock acquired, setting rss="LPU64
3318                                    ", kms="LPU64, loi->loi_lvb.lvb_size, tmp);
3319                         loi_kms_set(loi, tmp);
3320                 } else {
3321                         LDLM_DEBUG(lock, "lock acquired, setting rss="
3322                                    LPU64"; leaving kms="LPU64", end="LPU64,
3323                                    loi->loi_lvb.lvb_size, loi->loi_kms,
3324                                    lock->l_policy_data.l_extent.end);
3325                 }
3326                 ldlm_lock_allow_match(lock);
3327                 LDLM_LOCK_PUT(lock);
3328         } else if (rc == ELDLM_LOCK_ABORTED && (flags & LDLM_FL_HAS_INTENT)) {
3329                 loi->loi_lvb = *lvb;
3330                 CDEBUG(D_INODE, "glimpsed, setting rss="LPU64"; leaving"
3331                        " kms="LPU64"\n", loi->loi_lvb.lvb_size, loi->loi_kms);
3332                 rc = ELDLM_OK;
3333         }
3334 }
3335 EXPORT_SYMBOL(osc_update_enqueue);
3336
3337 struct ptlrpc_request_set *PTLRPCD_SET = (void *)1;
3338
3339 /* When enqueuing asynchronously, locks are not ordered, we can obtain a lock
3340  * from the 2nd OSC before a lock from the 1st one. This does not deadlock with
3341  * other synchronous requests, however keeping some locks and trying to obtain
3342  * others may take a considerable amount of time in a case of ost failure; and
3343  * when other sync requests do not get released lock from a client, the client
3344  * is excluded from the cluster -- such scenarious make the life difficult, so
3345  * release locks just after they are obtained. */
3346 int osc_enqueue_base(struct obd_export *exp, struct ldlm_res_id *res_id,
3347                      int *flags, ldlm_policy_data_t *policy,
3348                      struct ost_lvb *lvb, int kms_valid,
3349                      obd_enqueue_update_f upcall, void *cookie,
3350                      struct ldlm_enqueue_info *einfo,
3351                      struct lustre_handle *lockh,
3352                      struct ptlrpc_request_set *rqset, int async)
3353 {
3354         struct obd_device *obd = exp->exp_obd;
3355         struct ptlrpc_request *req = NULL;
3356         int intent = *flags & LDLM_FL_HAS_INTENT;
3357         ldlm_mode_t mode;
3358         int rc;
3359         ENTRY;
3360
3361         /* Filesystem lock extents are extended to page boundaries so that
3362          * dealing with the page cache is a little smoother.  */
3363         policy->l_extent.start -= policy->l_extent.start & ~CFS_PAGE_MASK;
3364         policy->l_extent.end |= ~CFS_PAGE_MASK;
3365
3366         /*
3367          * kms is not valid when either object is completely fresh (so that no
3368          * locks are cached), or object was evicted. In the latter case cached
3369          * lock cannot be used, because it would prime inode state with
3370          * potentially stale LVB.
3371          */
3372         if (!kms_valid)
3373                 goto no_match;
3374
3375         /* Next, search for already existing extent locks that will cover us */
3376         /* If we're trying to read, we also search for an existing PW lock.  The
3377          * VFS and page cache already protect us locally, so lots of readers/
3378          * writers can share a single PW lock.
3379          *
3380          * There are problems with conversion deadlocks, so instead of
3381          * converting a read lock to a write lock, we'll just enqueue a new
3382          * one.
3383          *
3384          * At some point we should cancel the read lock instead of making them
3385          * send us a blocking callback, but there are problems with canceling
3386          * locks out from other users right now, too. */
3387         mode = einfo->ei_mode;
3388         if (einfo->ei_mode == LCK_PR)
3389                 mode |= LCK_PW;
3390         mode = ldlm_lock_match(obd->obd_namespace,
3391                                *flags | LDLM_FL_LVB_READY, res_id,
3392                                einfo->ei_type, policy, mode, lockh, 0);
3393         if (mode) {
3394                 struct ldlm_lock *matched = ldlm_handle2lock(lockh);
3395
3396                 if (osc_set_lock_data_with_check(matched, einfo)) {
3397                         /* addref the lock only if not async requests and PW
3398                          * lock is matched whereas we asked for PR. */
3399                         if (!rqset && einfo->ei_mode != mode)
3400                                 ldlm_lock_addref(lockh, LCK_PR);
3401                         if (intent) {
3402                                 /* I would like to be able to ASSERT here that
3403                                  * rss <= kms, but I can't, for reasons which
3404                                  * are explained in lov_enqueue() */
3405                         }
3406
3407                         /* We already have a lock, and it's referenced */
3408                         (*upcall)(cookie, ELDLM_OK);
3409
3410                         /* For async requests, decref the lock. */
3411                         if (einfo->ei_mode != mode)
3412                                 ldlm_lock_decref(lockh, LCK_PW);
3413                         else if (rqset)
3414                                 ldlm_lock_decref(lockh, einfo->ei_mode);
3415                         LDLM_LOCK_PUT(matched);
3416                         RETURN(ELDLM_OK);
3417                 } else
3418                         ldlm_lock_decref(lockh, mode);
3419                 LDLM_LOCK_PUT(matched);
3420         }
3421
3422  no_match:
3423         if (intent) {
3424                 CFS_LIST_HEAD(cancels);
3425                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
3426                                            &RQF_LDLM_ENQUEUE_LVB);
3427                 if (req == NULL)
3428                         RETURN(-ENOMEM);
3429
3430                 rc = ldlm_prep_enqueue_req(exp, req, &cancels, 0);
3431                 if (rc) {
3432                         ptlrpc_request_free(req);
3433                         RETURN(rc);
3434                 }
3435
3436                 req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
3437                                      sizeof *lvb);
3438                 ptlrpc_request_set_replen(req);
3439         }
3440
3441         /* users of osc_enqueue() can pass this flag for ldlm_lock_match() */
3442         *flags &= ~LDLM_FL_BLOCK_GRANTED;
3443
3444         rc = ldlm_cli_enqueue(exp, &req, einfo, res_id, policy, flags, lvb,
3445                               sizeof(*lvb), lockh, async);
3446         if (rqset) {
3447                 if (!rc) {
3448                         struct osc_enqueue_args *aa;
3449                         CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
3450                         aa = ptlrpc_req_async_args(req);
3451                         aa->oa_ei = einfo;
3452                         aa->oa_exp = exp;
3453                         aa->oa_flags  = flags;
3454                         aa->oa_upcall = upcall;
3455                         aa->oa_cookie = cookie;
3456                         aa->oa_lvb    = lvb;
3457                         aa->oa_lockh  = lockh;
3458
3459                         req->rq_interpret_reply =
3460                                 (ptlrpc_interpterer_t)osc_enqueue_interpret;
3461                         if (rqset == PTLRPCD_SET)
3462                                 ptlrpcd_add_req(req, PSCOPE_OTHER);
3463                         else
3464                                 ptlrpc_set_add_req(rqset, req);
3465                 } else if (intent) {
3466                         ptlrpc_req_finished(req);
3467                 }
3468                 RETURN(rc);
3469         }
3470
3471         rc = osc_enqueue_fini(req, lvb, upcall, cookie, flags, rc);
3472         if (intent)
3473                 ptlrpc_req_finished(req);
3474
3475         RETURN(rc);
3476 }
3477
3478 static int osc_enqueue(struct obd_export *exp, struct obd_info *oinfo,
3479                        struct ldlm_enqueue_info *einfo,
3480                        struct ptlrpc_request_set *rqset)
3481 {
3482         struct ldlm_res_id res_id;
3483         int rc;
3484         ENTRY;
3485
3486         osc_build_res_name(oinfo->oi_md->lsm_object_id,
3487                            oinfo->oi_md->lsm_object_seq, &res_id);
3488
3489         rc = osc_enqueue_base(exp, &res_id, &oinfo->oi_flags, &oinfo->oi_policy,
3490                               &oinfo->oi_md->lsm_oinfo[0]->loi_lvb,
3491                               oinfo->oi_md->lsm_oinfo[0]->loi_kms_valid,
3492                               oinfo->oi_cb_up, oinfo, einfo, oinfo->oi_lockh,
3493                               rqset, rqset != NULL);
3494         RETURN(rc);
3495 }
3496
3497 int osc_match_base(struct obd_export *exp, struct ldlm_res_id *res_id,
3498                    __u32 type, ldlm_policy_data_t *policy, __u32 mode,
3499                    int *flags, void *data, struct lustre_handle *lockh,
3500                    int unref)
3501 {
3502         struct obd_device *obd = exp->exp_obd;
3503         int lflags = *flags;
3504         ldlm_mode_t rc;
3505         ENTRY;
3506
3507         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_MATCH))
3508                 RETURN(-EIO);
3509
3510         /* Filesystem lock extents are extended to page boundaries so that
3511          * dealing with the page cache is a little smoother */
3512         policy->l_extent.start -= policy->l_extent.start & ~CFS_PAGE_MASK;
3513         policy->l_extent.end |= ~CFS_PAGE_MASK;
3514
3515         /* Next, search for already existing extent locks that will cover us */
3516         /* If we're trying to read, we also search for an existing PW lock.  The
3517          * VFS and page cache already protect us locally, so lots of readers/
3518          * writers can share a single PW lock. */
3519         rc = mode;
3520         if (mode == LCK_PR)
3521                 rc |= LCK_PW;
3522         rc = ldlm_lock_match(obd->obd_namespace, lflags,
3523                              res_id, type, policy, rc, lockh, unref);
3524         if (rc) {
3525                 if (data != NULL) {
3526                         if (!osc_set_data_with_check(lockh, data)) {
3527                                 if (!(lflags & LDLM_FL_TEST_LOCK))
3528                                         ldlm_lock_decref(lockh, rc);
3529                                 RETURN(0);
3530                         }
3531                 }
3532                 if (!(lflags & LDLM_FL_TEST_LOCK) && mode != rc) {
3533                         ldlm_lock_addref(lockh, LCK_PR);
3534                         ldlm_lock_decref(lockh, LCK_PW);
3535                 }
3536                 RETURN(rc);
3537         }
3538         RETURN(rc);
3539 }
3540
3541 int osc_cancel_base(struct lustre_handle *lockh, __u32 mode)
3542 {
3543         ENTRY;
3544
3545         if (unlikely(mode == LCK_GROUP))
3546                 ldlm_lock_decref_and_cancel(lockh, mode);
3547         else
3548                 ldlm_lock_decref(lockh, mode);
3549
3550         RETURN(0);
3551 }
3552
3553 static int osc_cancel(struct obd_export *exp, struct lov_stripe_md *md,
3554                       __u32 mode, struct lustre_handle *lockh)
3555 {
3556         ENTRY;
3557         RETURN(osc_cancel_base(lockh, mode));
3558 }
3559
3560 static int osc_cancel_unused(struct obd_export *exp,
3561                              struct lov_stripe_md *lsm,
3562                              ldlm_cancel_flags_t flags,
3563                              void *opaque)
3564 {
3565         struct obd_device *obd = class_exp2obd(exp);
3566         struct ldlm_res_id res_id, *resp = NULL;
3567
3568         if (lsm != NULL) {
3569                 resp = osc_build_res_name(lsm->lsm_object_id,
3570                                           lsm->lsm_object_seq, &res_id);
3571         }
3572
3573         return ldlm_cli_cancel_unused(obd->obd_namespace, resp, flags, opaque);
3574 }
3575
3576 static int osc_statfs_interpret(const struct lu_env *env,
3577                                 struct ptlrpc_request *req,
3578                                 struct osc_async_args *aa, int rc)
3579 {
3580         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
3581         struct obd_statfs *msfs;
3582         __u64 used;
3583         ENTRY;
3584
3585         if (rc == -EBADR)
3586                 /* The request has in fact never been sent
3587                  * due to issues at a higher level (LOV).
3588                  * Exit immediately since the caller is
3589                  * aware of the problem and takes care
3590                  * of the clean up */
3591                  RETURN(rc);
3592
3593         if ((rc == -ENOTCONN || rc == -EAGAIN) &&
3594             (aa->aa_oi->oi_flags & OBD_STATFS_NODELAY))
3595                 GOTO(out, rc = 0);
3596
3597         if (rc != 0)
3598                 GOTO(out, rc);
3599
3600         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
3601         if (msfs == NULL) {
3602                 GOTO(out, rc = -EPROTO);
3603         }
3604
3605         /* Reinitialize the RDONLY and DEGRADED flags at the client
3606          * on each statfs, so they don't stay set permanently. */
3607         cfs_spin_lock(&cli->cl_oscc.oscc_lock);
3608
3609         if (unlikely(msfs->os_state & OS_STATE_DEGRADED))
3610                 cli->cl_oscc.oscc_flags |= OSCC_FLAG_DEGRADED;
3611         else if (unlikely(cli->cl_oscc.oscc_flags & OSCC_FLAG_DEGRADED))
3612                 cli->cl_oscc.oscc_flags &= ~OSCC_FLAG_DEGRADED;
3613
3614         if (unlikely(msfs->os_state & OS_STATE_READONLY))
3615                 cli->cl_oscc.oscc_flags |= OSCC_FLAG_RDONLY;
3616         else if (unlikely(cli->cl_oscc.oscc_flags & OSCC_FLAG_RDONLY))
3617                 cli->cl_oscc.oscc_flags &= ~OSCC_FLAG_RDONLY;
3618
3619         /* Add a bit of hysteresis so this flag isn't continually flapping,
3620          * and ensure that new files don't get extremely fragmented due to
3621          * only a small amount of available space in the filesystem.
3622          * We want to set the NOSPC flag when there is less than ~0.1% free
3623          * and clear it when there is at least ~0.2% free space, so:
3624          *                   avail < ~0.1% max          max = avail + used
3625          *            1025 * avail < avail + used       used = blocks - free
3626          *            1024 * avail < used
3627          *            1024 * avail < blocks - free
3628          *                   avail < ((blocks - free) >> 10)
3629          *
3630          * On very large disk, say 16TB 0.1% will be 16 GB. We don't want to
3631          * lose that amount of space so in those cases we report no space left
3632          * if their is less than 1 GB left.                             */
3633         used = min_t(__u64,(msfs->os_blocks - msfs->os_bfree) >> 10, 1 << 30);
3634         if (unlikely(((cli->cl_oscc.oscc_flags & OSCC_FLAG_NOSPC) == 0) &&
3635                      ((msfs->os_ffree < 32) || (msfs->os_bavail < used))))
3636                 cli->cl_oscc.oscc_flags |= OSCC_FLAG_NOSPC;
3637         else if (unlikely(((cli->cl_oscc.oscc_flags & OSCC_FLAG_NOSPC) != 0) &&
3638                 (msfs->os_ffree > 64) && (msfs->os_bavail > (used << 1))))
3639                         cli->cl_oscc.oscc_flags &= ~OSCC_FLAG_NOSPC;
3640
3641         cfs_spin_unlock(&cli->cl_oscc.oscc_lock);
3642
3643         *aa->aa_oi->oi_osfs = *msfs;
3644 out:
3645         rc = aa->aa_oi->oi_cb_up(aa->aa_oi, rc);
3646         RETURN(rc);
3647 }
3648
3649 static int osc_statfs_async(struct obd_device *obd, struct obd_info *oinfo,
3650                             __u64 max_age, struct ptlrpc_request_set *rqset)
3651 {
3652         struct ptlrpc_request *req;
3653         struct osc_async_args *aa;
3654         int                    rc;
3655         ENTRY;
3656
3657         /* We could possibly pass max_age in the request (as an absolute
3658          * timestamp or a "seconds.usec ago") so the target can avoid doing
3659          * extra calls into the filesystem if that isn't necessary (e.g.
3660          * during mount that would help a bit).  Having relative timestamps
3661          * is not so great if request processing is slow, while absolute
3662          * timestamps are not ideal because they need time synchronization. */
3663         req = ptlrpc_request_alloc(obd->u.cli.cl_import, &RQF_OST_STATFS);
3664         if (req == NULL)
3665                 RETURN(-ENOMEM);
3666
3667         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
3668         if (rc) {
3669                 ptlrpc_request_free(req);
3670                 RETURN(rc);
3671         }
3672         ptlrpc_request_set_replen(req);
3673         req->rq_request_portal = OST_CREATE_PORTAL;
3674         ptlrpc_at_set_req_timeout(req);
3675
3676         if (oinfo->oi_flags & OBD_STATFS_NODELAY) {
3677                 /* procfs requests not want stat in wait for avoid deadlock */
3678                 req->rq_no_resend = 1;
3679                 req->rq_no_delay = 1;
3680         }
3681
3682         req->rq_interpret_reply = (ptlrpc_interpterer_t)osc_statfs_interpret;
3683         CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
3684         aa = ptlrpc_req_async_args(req);
3685         aa->aa_oi = oinfo;
3686
3687         ptlrpc_set_add_req(rqset, req);
3688         RETURN(0);
3689 }
3690
3691 static int osc_statfs(struct obd_device *obd, struct obd_statfs *osfs,
3692                       __u64 max_age, __u32 flags)
3693 {
3694         struct obd_statfs     *msfs;
3695         struct ptlrpc_request *req;
3696         struct obd_import     *imp = NULL;
3697         int rc;
3698         ENTRY;
3699
3700         /*Since the request might also come from lprocfs, so we need
3701          *sync this with client_disconnect_export Bug15684*/
3702         cfs_down_read(&obd->u.cli.cl_sem);
3703         if (obd->u.cli.cl_import)
3704                 imp = class_import_get(obd->u.cli.cl_import);
3705         cfs_up_read(&obd->u.cli.cl_sem);
3706         if (!imp)
3707                 RETURN(-ENODEV);
3708
3709         /* We could possibly pass max_age in the request (as an absolute
3710          * timestamp or a "seconds.usec ago") so the target can avoid doing
3711          * extra calls into the filesystem if that isn't necessary (e.g.
3712          * during mount that would help a bit).  Having relative timestamps
3713          * is not so great if request processing is slow, while absolute
3714          * timestamps are not ideal because they need time synchronization. */
3715         req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
3716
3717         class_import_put(imp);
3718
3719         if (req == NULL)
3720                 RETURN(-ENOMEM);
3721
3722         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
3723         if (rc) {
3724                 ptlrpc_request_free(req);
3725                 RETURN(rc);
3726         }
3727         ptlrpc_request_set_replen(req);
3728         req->rq_request_portal = OST_CREATE_PORTAL;
3729         ptlrpc_at_set_req_timeout(req);
3730
3731         if (flags & OBD_STATFS_NODELAY) {
3732                 /* procfs requests not want stat in wait for avoid deadlock */
3733                 req->rq_no_resend = 1;
3734                 req->rq_no_delay = 1;
3735         }
3736
3737         rc = ptlrpc_queue_wait(req);
3738         if (rc)
3739                 GOTO(out, rc);
3740
3741         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
3742         if (msfs == NULL) {
3743                 GOTO(out, rc = -EPROTO);
3744         }
3745
3746         *osfs = *msfs;
3747
3748         EXIT;
3749  out:
3750         ptlrpc_req_finished(req);
3751         return rc;
3752 }
3753
3754 /* Retrieve object striping information.
3755  *
3756  * @lmmu is a pointer to an in-core struct with lmm_ost_count indicating
3757  * the maximum number of OST indices which will fit in the user buffer.
3758  * lmm_magic must be LOV_MAGIC (we only use 1 slot here).
3759  */
3760 static int osc_getstripe(struct lov_stripe_md *lsm, struct lov_user_md *lump)
3761 {
3762         /* we use lov_user_md_v3 because it is larger than lov_user_md_v1 */
3763         struct lov_user_md_v3 lum, *lumk;
3764         struct lov_user_ost_data_v1 *lmm_objects;
3765         int rc = 0, lum_size;
3766         ENTRY;
3767
3768         if (!lsm)
3769                 RETURN(-ENODATA);
3770
3771         /* we only need the header part from user space to get lmm_magic and
3772          * lmm_stripe_count, (the header part is common to v1 and v3) */
3773         lum_size = sizeof(struct lov_user_md_v1);
3774         if (cfs_copy_from_user(&lum, lump, lum_size))
3775                 RETURN(-EFAULT);
3776
3777         if ((lum.lmm_magic != LOV_USER_MAGIC_V1) &&
3778             (lum.lmm_magic != LOV_USER_MAGIC_V3))
3779                 RETURN(-EINVAL);
3780
3781         /* lov_user_md_vX and lov_mds_md_vX must have the same size */
3782         LASSERT(sizeof(struct lov_user_md_v1) == sizeof(struct lov_mds_md_v1));
3783         LASSERT(sizeof(struct lov_user_md_v3) == sizeof(struct lov_mds_md_v3));
3784         LASSERT(sizeof(lum.lmm_objects[0]) == sizeof(lumk->lmm_objects[0]));
3785
3786         /* we can use lov_mds_md_size() to compute lum_size
3787          * because lov_user_md_vX and lov_mds_md_vX have the same size */
3788         if (lum.lmm_stripe_count > 0) {
3789                 lum_size = lov_mds_md_size(lum.lmm_stripe_count, lum.lmm_magic);
3790                 OBD_ALLOC(lumk, lum_size);
3791                 if (!lumk)
3792                         RETURN(-ENOMEM);
3793
3794                 if (lum.lmm_magic == LOV_USER_MAGIC_V1)
3795                         lmm_objects = &(((struct lov_user_md_v1 *)lumk)->lmm_objects[0]);
3796                 else
3797                         lmm_objects = &(lumk->lmm_objects[0]);
3798                 lmm_objects->l_object_id = lsm->lsm_object_id;
3799         } else {
3800                 lum_size = lov_mds_md_size(0, lum.lmm_magic);
3801                 lumk = &lum;
3802         }
3803
3804         lumk->lmm_object_id = lsm->lsm_object_id;
3805         lumk->lmm_object_seq = lsm->lsm_object_seq;
3806         lumk->lmm_stripe_count = 1;
3807
3808         if (cfs_copy_to_user(lump, lumk, lum_size))
3809                 rc = -EFAULT;
3810
3811         if (lumk != &lum)
3812                 OBD_FREE(lumk, lum_size);
3813
3814         RETURN(rc);
3815 }
3816
3817
3818 static int osc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
3819                          void *karg, void *uarg)
3820 {
3821         struct obd_device *obd = exp->exp_obd;
3822         struct obd_ioctl_data *data = karg;
3823         int err = 0;
3824         ENTRY;
3825
3826         if (!cfs_try_module_get(THIS_MODULE)) {
3827                 CERROR("Can't get module. Is it alive?");
3828                 return -EINVAL;
3829         }
3830         switch (cmd) {
3831         case OBD_IOC_LOV_GET_CONFIG: {
3832                 char *buf;
3833                 struct lov_desc *desc;
3834                 struct obd_uuid uuid;
3835
3836                 buf = NULL;
3837                 len = 0;
3838                 if (obd_ioctl_getdata(&buf, &len, (void *)uarg))
3839                         GOTO(out, err = -EINVAL);
3840
3841                 data = (struct obd_ioctl_data *)buf;
3842
3843                 if (sizeof(*desc) > data->ioc_inllen1) {
3844                         obd_ioctl_freedata(buf, len);
3845                         GOTO(out, err = -EINVAL);
3846                 }
3847
3848                 if (data->ioc_inllen2 < sizeof(uuid)) {
3849                         obd_ioctl_freedata(buf, len);
3850                         GOTO(out, err = -EINVAL);
3851                 }
3852
3853                 desc = (struct lov_desc *)data->ioc_inlbuf1;
3854                 desc->ld_tgt_count = 1;
3855                 desc->ld_active_tgt_count = 1;
3856                 desc->ld_default_stripe_count = 1;
3857                 desc->ld_default_stripe_size = 0;
3858                 desc->ld_default_stripe_offset = 0;
3859                 desc->ld_pattern = 0;
3860                 memcpy(&desc->ld_uuid, &obd->obd_uuid, sizeof(uuid));
3861
3862                 memcpy(data->ioc_inlbuf2, &obd->obd_uuid, sizeof(uuid));
3863
3864                 err = cfs_copy_to_user((void *)uarg, buf, len);
3865                 if (err)
3866                         err = -EFAULT;
3867                 obd_ioctl_freedata(buf, len);
3868                 GOTO(out, err);
3869         }
3870         case LL_IOC_LOV_SETSTRIPE:
3871                 err = obd_alloc_memmd(exp, karg);
3872                 if (err > 0)
3873                         err = 0;
3874                 GOTO(out, err);
3875         case LL_IOC_LOV_GETSTRIPE:
3876                 err = osc_getstripe(karg, uarg);
3877                 GOTO(out, err);
3878         case OBD_IOC_CLIENT_RECOVER:
3879                 err = ptlrpc_recover_import(obd->u.cli.cl_import,
3880                                             data->ioc_inlbuf1);
3881                 if (err > 0)
3882                         err = 0;
3883                 GOTO(out, err);
3884         case IOC_OSC_SET_ACTIVE:
3885                 err = ptlrpc_set_import_active(obd->u.cli.cl_import,
3886                                                data->ioc_offset);
3887                 GOTO(out, err);
3888         case OBD_IOC_POLL_QUOTACHECK:
3889                 err = lquota_poll_check(quota_interface, exp,
3890                                         (struct if_quotacheck *)karg);
3891                 GOTO(out, err);
3892         case OBD_IOC_PING_TARGET:
3893                 err = ptlrpc_obd_ping(obd);
3894                 GOTO(out, err);
3895         default:
3896                 CDEBUG(D_INODE, "unrecognised ioctl %#x by %s\n",
3897                        cmd, cfs_curproc_comm());
3898                 GOTO(out, err = -ENOTTY);
3899         }
3900 out:
3901         cfs_module_put(THIS_MODULE);
3902         return err;
3903 }
3904
3905 static int osc_get_info(struct obd_export *exp, obd_count keylen,
3906                         void *key, __u32 *vallen, void *val,
3907                         struct lov_stripe_md *lsm)
3908 {
3909         ENTRY;
3910         if (!vallen || !val)
3911                 RETURN(-EFAULT);
3912
3913         if (KEY_IS(KEY_LOCK_TO_STRIPE)) {
3914                 __u32 *stripe = val;
3915                 *vallen = sizeof(*stripe);
3916                 *stripe = 0;
3917                 RETURN(0);
3918         } else if (KEY_IS(KEY_LAST_ID)) {
3919                 struct ptlrpc_request *req;
3920                 obd_id                *reply;
3921                 char                  *tmp;
3922                 int                    rc;
3923
3924                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
3925                                            &RQF_OST_GET_INFO_LAST_ID);
3926                 if (req == NULL)
3927                         RETURN(-ENOMEM);
3928
3929                 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
3930                                      RCL_CLIENT, keylen);
3931                 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GET_INFO);
3932                 if (rc) {
3933                         ptlrpc_request_free(req);
3934                         RETURN(rc);
3935                 }
3936
3937                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
3938                 memcpy(tmp, key, keylen);
3939
3940                 req->rq_no_delay = req->rq_no_resend = 1;
3941                 ptlrpc_request_set_replen(req);
3942                 rc = ptlrpc_queue_wait(req);
3943                 if (rc)
3944                         GOTO(out, rc);
3945
3946                 reply = req_capsule_server_get(&req->rq_pill, &RMF_OBD_ID);
3947                 if (reply == NULL)
3948                         GOTO(out, rc = -EPROTO);
3949
3950                 *((obd_id *)val) = *reply;
3951         out:
3952                 ptlrpc_req_finished(req);
3953                 RETURN(rc);
3954         } else if (KEY_IS(KEY_FIEMAP)) {
3955                 struct ptlrpc_request *req;
3956                 struct ll_user_fiemap *reply;
3957                 char *tmp;
3958                 int rc;
3959
3960                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
3961                                            &RQF_OST_GET_INFO_FIEMAP);
3962                 if (req == NULL)
3963                         RETURN(-ENOMEM);
3964
3965                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_KEY,
3966                                      RCL_CLIENT, keylen);
3967                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_VAL,
3968                                      RCL_CLIENT, *vallen);
3969                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_VAL,
3970                                      RCL_SERVER, *vallen);
3971
3972                 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GET_INFO);
3973                 if (rc) {
3974                         ptlrpc_request_free(req);
3975                         RETURN(rc);
3976                 }
3977
3978                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_FIEMAP_KEY);
3979                 memcpy(tmp, key, keylen);
3980                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_FIEMAP_VAL);
3981                 memcpy(tmp, val, *vallen);
3982
3983                 ptlrpc_request_set_replen(req);
3984                 rc = ptlrpc_queue_wait(req);
3985                 if (rc)
3986                         GOTO(out1, rc);
3987
3988                 reply = req_capsule_server_get(&req->rq_pill, &RMF_FIEMAP_VAL);
3989                 if (reply == NULL)
3990                         GOTO(out1, rc = -EPROTO);
3991
3992                 memcpy(val, reply, *vallen);
3993         out1:
3994                 ptlrpc_req_finished(req);
3995
3996                 RETURN(rc);
3997         }
3998
3999         RETURN(-EINVAL);
4000 }
4001
4002 static int osc_setinfo_mds_connect_import(struct obd_import *imp)
4003 {
4004         struct llog_ctxt *ctxt;
4005         int rc = 0;
4006         ENTRY;
4007
4008         ctxt = llog_get_context(imp->imp_obd, LLOG_MDS_OST_ORIG_CTXT);
4009         if (ctxt) {
4010                 rc = llog_initiator_connect(ctxt);
4011                 llog_ctxt_put(ctxt);
4012         } else {
4013                 /* XXX return an error? skip setting below flags? */
4014         }
4015
4016         cfs_spin_lock(&imp->imp_lock);
4017         imp->imp_server_timeout = 1;
4018         imp->imp_pingable = 1;
4019         cfs_spin_unlock(&imp->imp_lock);
4020         CDEBUG(D_RPCTRACE, "pinging OST %s\n", obd2cli_tgt(imp->imp_obd));
4021
4022         RETURN(rc);
4023 }
4024
4025 static int osc_setinfo_mds_conn_interpret(const struct lu_env *env,
4026                                           struct ptlrpc_request *req,
4027                                           void *aa, int rc)
4028 {
4029         ENTRY;
4030         if (rc != 0)
4031                 RETURN(rc);
4032
4033         RETURN(osc_setinfo_mds_connect_import(req->rq_import));
4034 }
4035
4036 static int osc_set_info_async(struct obd_export *exp, obd_count keylen,
4037                               void *key, obd_count vallen, void *val,
4038                               struct ptlrpc_request_set *set)
4039 {
4040         struct ptlrpc_request *req;
4041         struct obd_device     *obd = exp->exp_obd;
4042         struct obd_import     *imp = class_exp2cliimp(exp);
4043         char                  *tmp;
4044         int                    rc;
4045         ENTRY;
4046
4047         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_SHUTDOWN, 10);
4048
4049         if (KEY_IS(KEY_NEXT_ID)) {
4050                 obd_id new_val;
4051                 struct osc_creator *oscc = &obd->u.cli.cl_oscc;
4052
4053                 if (vallen != sizeof(obd_id))
4054                         RETURN(-ERANGE);
4055                 if (val == NULL)
4056                         RETURN(-EINVAL);
4057
4058                 if (vallen != sizeof(obd_id))
4059                         RETURN(-EINVAL);
4060
4061                 /* avoid race between allocate new object and set next id
4062                  * from ll_sync thread */
4063                 cfs_spin_lock(&oscc->oscc_lock);
4064                 new_val = *((obd_id*)val) + 1;
4065                 if (new_val > oscc->oscc_next_id)
4066                         oscc->oscc_next_id = new_val;
4067                 cfs_spin_unlock(&oscc->oscc_lock);
4068                 CDEBUG(D_HA, "%s: set oscc_next_id = "LPU64"\n",
4069                        exp->exp_obd->obd_name,
4070                        obd->u.cli.cl_oscc.oscc_next_id);
4071
4072                 RETURN(0);
4073         }
4074
4075         if (KEY_IS(KEY_CHECKSUM)) {
4076                 if (vallen != sizeof(int))
4077                         RETURN(-EINVAL);
4078                 exp->exp_obd->u.cli.cl_checksum = (*(int *)val) ? 1 : 0;
4079                 RETURN(0);
4080         }
4081
4082         if (KEY_IS(KEY_SPTLRPC_CONF)) {
4083                 sptlrpc_conf_client_adapt(obd);
4084                 RETURN(0);
4085         }
4086
4087         if (KEY_IS(KEY_FLUSH_CTX)) {
4088                 sptlrpc_import_flush_my_ctx(imp);
4089                 RETURN(0);
4090         }
4091
4092         if (!set && !KEY_IS(KEY_GRANT_SHRINK))
4093                 RETURN(-EINVAL);
4094
4095         /* We pass all other commands directly to OST. Since nobody calls osc
4096            methods directly and everybody is supposed to go through LOV, we
4097            assume lov checked invalid values for us.
4098            The only recognised values so far are evict_by_nid and mds_conn.
4099            Even if something bad goes through, we'd get a -EINVAL from OST
4100            anyway. */
4101
4102         if (KEY_IS(KEY_GRANT_SHRINK))
4103                 req = ptlrpc_request_alloc(imp, &RQF_OST_SET_GRANT_INFO);
4104         else
4105                 req = ptlrpc_request_alloc(imp, &RQF_OBD_SET_INFO);
4106
4107         if (req == NULL)
4108                 RETURN(-ENOMEM);
4109
4110         req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
4111                              RCL_CLIENT, keylen);
4112         req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_VAL,
4113                              RCL_CLIENT, vallen);
4114         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SET_INFO);
4115         if (rc) {
4116                 ptlrpc_request_free(req);
4117                 RETURN(rc);
4118         }
4119
4120         tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
4121         memcpy(tmp, key, keylen);
4122         tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_VAL);
4123         memcpy(tmp, val, vallen);
4124
4125         if (KEY_IS(KEY_MDS_CONN)) {
4126                 struct osc_creator *oscc = &obd->u.cli.cl_oscc;
4127
4128                 oscc->oscc_oa.o_seq = (*(__u32 *)val);
4129                 oscc->oscc_oa.o_valid |= OBD_MD_FLGROUP;
4130                 LASSERT_SEQ_IS_MDT(oscc->oscc_oa.o_seq);
4131                 req->rq_no_delay = req->rq_no_resend = 1;
4132                 req->rq_interpret_reply = osc_setinfo_mds_conn_interpret;
4133         } else if (KEY_IS(KEY_GRANT_SHRINK)) {
4134                 struct osc_grant_args *aa;
4135                 struct obdo *oa;
4136
4137                 CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
4138                 aa = ptlrpc_req_async_args(req);
4139                 OBDO_ALLOC(oa);
4140                 if (!oa) {
4141                         ptlrpc_req_finished(req);
4142                         RETURN(-ENOMEM);
4143                 }
4144                 *oa = ((struct ost_body *)val)->oa;
4145                 aa->aa_oa = oa;
4146                 req->rq_interpret_reply = osc_shrink_grant_interpret;
4147         }
4148
4149         ptlrpc_request_set_replen(req);
4150         if (!KEY_IS(KEY_GRANT_SHRINK)) {
4151                 LASSERT(set != NULL);
4152                 ptlrpc_set_add_req(set, req);
4153                 ptlrpc_check_set(NULL, set);
4154         } else
4155                 ptlrpcd_add_req(req, PSCOPE_OTHER);
4156
4157         RETURN(0);
4158 }
4159
4160
4161 static struct llog_operations osc_size_repl_logops = {
4162         lop_cancel: llog_obd_repl_cancel
4163 };
4164
4165 static struct llog_operations osc_mds_ost_orig_logops;
4166
4167 static int __osc_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
4168                            struct obd_device *tgt, struct llog_catid *catid)
4169 {
4170         int rc;
4171         ENTRY;
4172
4173         rc = llog_setup(obd, &obd->obd_olg, LLOG_MDS_OST_ORIG_CTXT, tgt, 1,
4174                         &catid->lci_logid, &osc_mds_ost_orig_logops);
4175         if (rc) {
4176                 CERROR("failed LLOG_MDS_OST_ORIG_CTXT\n");
4177                 GOTO(out, rc);
4178         }
4179
4180         rc = llog_setup(obd, &obd->obd_olg, LLOG_SIZE_REPL_CTXT, tgt, 1,
4181                         NULL, &osc_size_repl_logops);
4182         if (rc) {
4183                 struct llog_ctxt *ctxt =
4184                         llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
4185                 if (ctxt)
4186                         llog_cleanup(ctxt);
4187                 CERROR("failed LLOG_SIZE_REPL_CTXT\n");
4188         }
4189         GOTO(out, rc);
4190 out:
4191         if (rc) {
4192                 CERROR("osc '%s' tgt '%s' catid %p rc=%d\n",
4193                        obd->obd_name, tgt->obd_name, catid, rc);
4194                 CERROR("logid "LPX64":0x%x\n",
4195                        catid->lci_logid.lgl_oid, catid->lci_logid.lgl_ogen);
4196         }
4197         return rc;
4198 }
4199
4200 static int osc_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
4201                          struct obd_device *disk_obd, int *index)
4202 {
4203         struct llog_catid catid;
4204         static char name[32] = CATLIST;
4205         int rc;
4206         ENTRY;
4207
4208         LASSERT(olg == &obd->obd_olg);
4209
4210         cfs_mutex_down(&olg->olg_cat_processing);
4211         rc = llog_get_cat_list(disk_obd, name, *index, 1, &catid);
4212         if (rc) {
4213                 CERROR("rc: %d\n", rc);
4214                 GOTO(out, rc);
4215         }
4216
4217         CDEBUG(D_INFO, "%s: Init llog for %d - catid "LPX64"/"LPX64":%x\n",
4218                obd->obd_name, *index, catid.lci_logid.lgl_oid,
4219                catid.lci_logid.lgl_oseq, catid.lci_logid.lgl_ogen);
4220
4221         rc = __osc_llog_init(obd, olg, disk_obd, &catid);
4222         if (rc) {
4223                 CERROR("rc: %d\n", rc);
4224                 GOTO(out, rc);
4225         }
4226
4227         rc = llog_put_cat_list(disk_obd, name, *index, 1, &catid);
4228         if (rc) {
4229                 CERROR("rc: %d\n", rc);
4230                 GOTO(out, rc);
4231         }
4232
4233  out:
4234         cfs_mutex_up(&olg->olg_cat_processing);
4235
4236         return rc;
4237 }
4238
4239 static int osc_llog_finish(struct obd_device *obd, int count)
4240 {
4241         struct llog_ctxt *ctxt;
4242         int rc = 0, rc2 = 0;
4243         ENTRY;
4244
4245         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
4246         if (ctxt)
4247                 rc = llog_cleanup(ctxt);
4248
4249         ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
4250         if (ctxt)
4251                 rc2 = llog_cleanup(ctxt);
4252         if (!rc)
4253                 rc = rc2;
4254
4255         RETURN(rc);
4256 }
4257
4258 static int osc_reconnect(const struct lu_env *env,
4259                          struct obd_export *exp, struct obd_device *obd,
4260                          struct obd_uuid *cluuid,
4261                          struct obd_connect_data *data,
4262                          void *localdata)
4263 {
4264         struct client_obd *cli = &obd->u.cli;
4265
4266         if (data != NULL && (data->ocd_connect_flags & OBD_CONNECT_GRANT)) {
4267                 long lost_grant;
4268
4269                 client_obd_list_lock(&cli->cl_loi_list_lock);
4270                 data->ocd_grant = (cli->cl_avail_grant + cli->cl_dirty) ?:
4271                                 2 * cli->cl_max_pages_per_rpc << CFS_PAGE_SHIFT;
4272                 lost_grant = cli->cl_lost_grant;
4273                 cli->cl_lost_grant = 0;
4274                 client_obd_list_unlock(&cli->cl_loi_list_lock);
4275
4276                 CDEBUG(D_CACHE, "request ocd_grant: %d cl_avail_grant: %ld "
4277                        "cl_dirty: %ld cl_lost_grant: %ld\n", data->ocd_grant,
4278                        cli->cl_avail_grant, cli->cl_dirty, lost_grant);
4279                 CDEBUG(D_RPCTRACE, "ocd_connect_flags: "LPX64" ocd_version: %d"
4280                        " ocd_grant: %d\n", data->ocd_connect_flags,
4281                        data->ocd_version, data->ocd_grant);
4282         }
4283
4284         RETURN(0);
4285 }
4286
4287 static int osc_disconnect(struct obd_export *exp)
4288 {
4289         struct obd_device *obd = class_exp2obd(exp);
4290         struct llog_ctxt  *ctxt;
4291         int rc;
4292
4293         ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
4294         if (ctxt) {
4295                 if (obd->u.cli.cl_conn_count == 1) {
4296                         /* Flush any remaining cancel messages out to the
4297                          * target */
4298                         llog_sync(ctxt, exp);
4299                 }
4300                 llog_ctxt_put(ctxt);
4301         } else {
4302                 CDEBUG(D_HA, "No LLOG_SIZE_REPL_CTXT found in obd %p\n",
4303                        obd);
4304         }
4305
4306         rc = client_disconnect_export(exp);
4307         /**
4308          * Initially we put del_shrink_grant before disconnect_export, but it
4309          * causes the following problem if setup (connect) and cleanup
4310          * (disconnect) are tangled together.
4311          *      connect p1                     disconnect p2
4312          *   ptlrpc_connect_import
4313          *     ...............               class_manual_cleanup
4314          *                                     osc_disconnect
4315          *                                     del_shrink_grant
4316          *   ptlrpc_connect_interrupt
4317          *     init_grant_shrink
4318          *   add this client to shrink list
4319          *                                      cleanup_osc
4320          * Bang! pinger trigger the shrink.
4321          * So the osc should be disconnected from the shrink list, after we
4322          * are sure the import has been destroyed. BUG18662
4323          */
4324         if (obd->u.cli.cl_import == NULL)
4325                 osc_del_shrink_grant(&obd->u.cli);
4326         return rc;
4327 }
4328
4329 static int osc_import_event(struct obd_device *obd,
4330                             struct obd_import *imp,
4331                             enum obd_import_event event)
4332 {
4333         struct client_obd *cli;
4334         int rc = 0;
4335
4336         ENTRY;
4337         LASSERT(imp->imp_obd == obd);
4338
4339         switch (event) {
4340         case IMP_EVENT_DISCON: {
4341                 /* Only do this on the MDS OSC's */
4342                 if (imp->imp_server_timeout) {
4343                         struct osc_creator *oscc = &obd->u.cli.cl_oscc;
4344
4345                         cfs_spin_lock(&oscc->oscc_lock);
4346                         oscc->oscc_flags |= OSCC_FLAG_RECOVERING;
4347                         cfs_spin_unlock(&oscc->oscc_lock);
4348                 }
4349                 cli = &obd->u.cli;
4350                 client_obd_list_lock(&cli->cl_loi_list_lock);
4351                 cli->cl_avail_grant = 0;
4352                 cli->cl_lost_grant = 0;
4353                 client_obd_list_unlock(&cli->cl_loi_list_lock);
4354                 break;
4355         }
4356         case IMP_EVENT_INACTIVE: {
4357                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_INACTIVE, NULL);
4358                 break;
4359         }
4360         case IMP_EVENT_INVALIDATE: {
4361                 struct ldlm_namespace *ns = obd->obd_namespace;
4362                 struct lu_env         *env;
4363                 int                    refcheck;
4364
4365                 env = cl_env_get(&refcheck);
4366                 if (!IS_ERR(env)) {
4367                         /* Reset grants */
4368                         cli = &obd->u.cli;
4369                         client_obd_list_lock(&cli->cl_loi_list_lock);
4370                         /* all pages go to failing rpcs due to the invalid
4371                          * import */
4372                         osc_check_rpcs(env, cli);
4373                         client_obd_list_unlock(&cli->cl_loi_list_lock);
4374
4375                         ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
4376                         cl_env_put(env, &refcheck);
4377                 } else
4378                         rc = PTR_ERR(env);
4379                 break;
4380         }
4381         case IMP_EVENT_ACTIVE: {
4382                 /* Only do this on the MDS OSC's */
4383                 if (imp->imp_server_timeout) {
4384                         struct osc_creator *oscc = &obd->u.cli.cl_oscc;
4385
4386                         cfs_spin_lock(&oscc->oscc_lock);
4387                         oscc->oscc_flags &= ~OSCC_FLAG_NOSPC;
4388                         cfs_spin_unlock(&oscc->oscc_lock);
4389                 }
4390                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE, NULL);
4391                 break;
4392         }
4393         case IMP_EVENT_OCD: {
4394                 struct obd_connect_data *ocd = &imp->imp_connect_data;
4395
4396                 if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT)
4397                         osc_init_grant(&obd->u.cli, ocd);
4398
4399                 /* See bug 7198 */
4400                 if (ocd->ocd_connect_flags & OBD_CONNECT_REQPORTAL)
4401                         imp->imp_client->cli_request_portal =OST_REQUEST_PORTAL;
4402
4403                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_OCD, NULL);
4404                 break;
4405         }
4406         case IMP_EVENT_DEACTIVATE: {
4407                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_DEACTIVATE, NULL);
4408                 break;
4409         }
4410         case IMP_EVENT_ACTIVATE: {
4411                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVATE, NULL);
4412                 break;
4413         }
4414         default:
4415                 CERROR("Unknown import event %d\n", event);
4416                 LBUG();
4417         }
4418         RETURN(rc);
4419 }
4420
4421 /**
4422  * Determine whether the lock can be canceled before replaying the lock
4423  * during recovery, see bug16774 for detailed information.
4424  *
4425  * \retval zero the lock can't be canceled
4426  * \retval other ok to cancel
4427  */
4428 static int osc_cancel_for_recovery(struct ldlm_lock *lock)
4429 {
4430         check_res_locked(lock->l_resource);
4431
4432         /*
4433          * Cancel all unused extent lock in granted mode LCK_PR or LCK_CR.
4434          *
4435          * XXX as a future improvement, we can also cancel unused write lock
4436          * if it doesn't have dirty data and active mmaps.
4437          */
4438         if (lock->l_resource->lr_type == LDLM_EXTENT &&
4439             (lock->l_granted_mode == LCK_PR ||
4440              lock->l_granted_mode == LCK_CR) &&
4441             (osc_dlm_lock_pageref(lock) == 0))
4442                 RETURN(1);
4443
4444         RETURN(0);
4445 }
4446
4447 int osc_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
4448 {
4449         int rc;
4450         ENTRY;
4451
4452         ENTRY;
4453         rc = ptlrpcd_addref();
4454         if (rc)
4455                 RETURN(rc);
4456
4457         rc = client_obd_setup(obd, lcfg);
4458         if (rc) {
4459                 ptlrpcd_decref();
4460         } else {
4461                 struct lprocfs_static_vars lvars = { 0 };
4462                 struct client_obd *cli = &obd->u.cli;
4463
4464                 cli->cl_grant_shrink_interval = GRANT_SHRINK_INTERVAL;
4465                 lprocfs_osc_init_vars(&lvars);
4466                 if (lprocfs_obd_setup(obd, lvars.obd_vars) == 0) {
4467                         lproc_osc_attach_seqstat(obd);
4468                         sptlrpc_lprocfs_cliobd_attach(obd);
4469                         ptlrpc_lprocfs_register_obd(obd);
4470                 }
4471
4472                 oscc_init(obd);
4473                 /* We need to allocate a few requests more, because
4474                    brw_interpret tries to create new requests before freeing
4475                    previous ones. Ideally we want to have 2x max_rpcs_in_flight
4476                    reserved, but I afraid that might be too much wasted RAM
4477                    in fact, so 2 is just my guess and still should work. */
4478                 cli->cl_import->imp_rq_pool =
4479                         ptlrpc_init_rq_pool(cli->cl_max_rpcs_in_flight + 2,
4480                                             OST_MAXREQSIZE,
4481                                             ptlrpc_add_rqs_to_pool);
4482
4483                 CFS_INIT_LIST_HEAD(&cli->cl_grant_shrink_list);
4484                 cfs_sema_init(&cli->cl_grant_sem, 1);
4485
4486                 ns_register_cancel(obd->obd_namespace, osc_cancel_for_recovery);
4487         }
4488
4489         RETURN(rc);
4490 }
4491
4492 static int osc_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
4493 {
4494         int rc = 0;
4495         ENTRY;
4496
4497         switch (stage) {
4498         case OBD_CLEANUP_EARLY: {
4499                 struct obd_import *imp;
4500                 imp = obd->u.cli.cl_import;
4501                 CDEBUG(D_HA, "Deactivating import %s\n", obd->obd_name);
4502                 /* ptlrpc_abort_inflight to stop an mds_lov_synchronize */
4503                 ptlrpc_deactivate_import(imp);
4504                 cfs_spin_lock(&imp->imp_lock);
4505                 imp->imp_pingable = 0;
4506                 cfs_spin_unlock(&imp->imp_lock);
4507                 break;
4508         }
4509         case OBD_CLEANUP_EXPORTS: {
4510                 /* LU-464
4511                  * for echo client, export may be on zombie list, wait for
4512                  * zombie thread to cull it, because cli.cl_import will be
4513                  * cleared in client_disconnect_export():
4514                  *   class_export_destroy() -> obd_cleanup() ->
4515                  *   echo_device_free() -> echo_client_cleanup() ->
4516                  *   obd_disconnect() -> osc_disconnect() ->
4517                  *   client_disconnect_export()
4518                  */
4519                 obd_zombie_barrier();
4520                 /* If we set up but never connected, the
4521                    client import will not have been cleaned. */
4522                 if (obd->u.cli.cl_import) {
4523                         struct obd_import *imp;
4524                         cfs_down_write(&obd->u.cli.cl_sem);
4525                         imp = obd->u.cli.cl_import;
4526                         CDEBUG(D_CONFIG, "%s: client import never connected\n",
4527                                obd->obd_name);
4528                         ptlrpc_invalidate_import(imp);
4529                         if (imp->imp_rq_pool) {
4530                                 ptlrpc_free_rq_pool(imp->imp_rq_pool);
4531                                 imp->imp_rq_pool = NULL;
4532                         }
4533                         class_destroy_import(imp);
4534                         cfs_up_write(&obd->u.cli.cl_sem);
4535                         obd->u.cli.cl_import = NULL;
4536                 }
4537                 rc = obd_llog_finish(obd, 0);
4538                 if (rc != 0)
4539                         CERROR("failed to cleanup llogging subsystems\n");
4540                 break;
4541                 }
4542         }
4543         RETURN(rc);
4544 }
4545
4546 int osc_cleanup(struct obd_device *obd)
4547 {
4548         int rc;
4549
4550         ENTRY;
4551         ptlrpc_lprocfs_unregister_obd(obd);
4552         lprocfs_obd_cleanup(obd);
4553
4554         /* free memory of osc quota cache */
4555         lquota_cleanup(quota_interface, obd);
4556
4557         rc = client_obd_cleanup(obd);
4558
4559         ptlrpcd_decref();
4560         RETURN(rc);
4561 }
4562
4563 int osc_process_config_base(struct obd_device *obd, struct lustre_cfg *lcfg)
4564 {
4565         struct lprocfs_static_vars lvars = { 0 };
4566         int rc = 0;
4567
4568         lprocfs_osc_init_vars(&lvars);
4569
4570         switch (lcfg->lcfg_command) {
4571         default:
4572                 rc = class_process_proc_param(PARAM_OSC, lvars.obd_vars,
4573                                               lcfg, obd);
4574                 if (rc > 0)
4575                         rc = 0;
4576                 break;
4577         }
4578
4579         return(rc);
4580 }
4581
4582 static int osc_process_config(struct obd_device *obd, obd_count len, void *buf)
4583 {
4584         return osc_process_config_base(obd, buf);
4585 }
4586
4587 struct obd_ops osc_obd_ops = {
4588         .o_owner                = THIS_MODULE,
4589         .o_setup                = osc_setup,
4590         .o_precleanup           = osc_precleanup,
4591         .o_cleanup              = osc_cleanup,
4592         .o_add_conn             = client_import_add_conn,
4593         .o_del_conn             = client_import_del_conn,
4594         .o_connect              = client_connect_import,
4595         .o_reconnect            = osc_reconnect,
4596         .o_disconnect           = osc_disconnect,
4597         .o_statfs               = osc_statfs,
4598         .o_statfs_async         = osc_statfs_async,
4599         .o_packmd               = osc_packmd,
4600         .o_unpackmd             = osc_unpackmd,
4601         .o_precreate            = osc_precreate,
4602         .o_create               = osc_create,
4603         .o_create_async         = osc_create_async,
4604         .o_destroy              = osc_destroy,
4605         .o_getattr              = osc_getattr,
4606         .o_getattr_async        = osc_getattr_async,
4607         .o_setattr              = osc_setattr,
4608         .o_setattr_async        = osc_setattr_async,
4609         .o_brw                  = osc_brw,
4610         .o_punch                = osc_punch,
4611         .o_sync                 = osc_sync,
4612         .o_enqueue              = osc_enqueue,
4613         .o_change_cbdata        = osc_change_cbdata,
4614         .o_find_cbdata          = osc_find_cbdata,
4615         .o_cancel               = osc_cancel,
4616         .o_cancel_unused        = osc_cancel_unused,
4617         .o_iocontrol            = osc_iocontrol,
4618         .o_get_info             = osc_get_info,
4619         .o_set_info_async       = osc_set_info_async,
4620         .o_import_event         = osc_import_event,
4621         .o_llog_init            = osc_llog_init,
4622         .o_llog_finish          = osc_llog_finish,
4623         .o_process_config       = osc_process_config,
4624 };
4625
4626 extern struct lu_kmem_descr osc_caches[];
4627 extern cfs_spinlock_t       osc_ast_guard;
4628 extern cfs_lock_class_key_t osc_ast_guard_class;
4629
4630 int __init osc_init(void)
4631 {
4632         struct lprocfs_static_vars lvars = { 0 };
4633         int rc;
4634         ENTRY;
4635
4636         /* print an address of _any_ initialized kernel symbol from this
4637          * module, to allow debugging with gdb that doesn't support data
4638          * symbols from modules.*/
4639         CDEBUG(D_CONSOLE, "Lustre OSC module (%p).\n", &osc_caches);
4640
4641         rc = lu_kmem_init(osc_caches);
4642
4643         lprocfs_osc_init_vars(&lvars);
4644
4645         cfs_request_module("lquota");
4646         quota_interface = PORTAL_SYMBOL_GET(osc_quota_interface);
4647         lquota_init(quota_interface);
4648         init_obd_quota_ops(quota_interface, &osc_obd_ops);
4649
4650         rc = class_register_type(&osc_obd_ops, NULL, lvars.module_vars,
4651                                  LUSTRE_OSC_NAME, &osc_device_type);
4652         if (rc) {
4653                 if (quota_interface)
4654                         PORTAL_SYMBOL_PUT(osc_quota_interface);
4655                 lu_kmem_fini(osc_caches);
4656                 RETURN(rc);
4657         }
4658
4659         cfs_spin_lock_init(&osc_ast_guard);
4660         cfs_lockdep_set_class(&osc_ast_guard, &osc_ast_guard_class);
4661
4662         osc_mds_ost_orig_logops = llog_lvfs_ops;
4663         osc_mds_ost_orig_logops.lop_setup = llog_obd_origin_setup;
4664         osc_mds_ost_orig_logops.lop_cleanup = llog_obd_origin_cleanup;
4665         osc_mds_ost_orig_logops.lop_add = llog_obd_origin_add;
4666         osc_mds_ost_orig_logops.lop_connect = llog_origin_connect;
4667
4668         RETURN(rc);
4669 }
4670
4671 #ifdef __KERNEL__
4672 static void /*__exit*/ osc_exit(void)
4673 {
4674         lu_device_type_fini(&osc_device_type);
4675
4676         lquota_exit(quota_interface);
4677         if (quota_interface)
4678                 PORTAL_SYMBOL_PUT(osc_quota_interface);
4679
4680         class_unregister_type(LUSTRE_OSC_NAME);
4681         lu_kmem_fini(osc_caches);
4682 }
4683
4684 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
4685 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC)");
4686 MODULE_LICENSE("GPL");
4687
4688 cfs_module(osc, LUSTRE_VERSION_STRING, osc_init, osc_exit);
4689 #endif