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