lustre/osc/osc_request.c

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