lustre/osc/osc_request.c

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