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         unlock_res_and_lock(lock);
3048         LDLM_LOCK_PUT(lock);
3049 }
3050
3051 static int osc_change_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
3052                              ldlm_iterator_t replace, void *data)
3053 {
3054         struct ldlm_res_id res_id;
3055         struct obd_device *obd = class_exp2obd(exp);
3056
3057         osc_build_res_name(lsm->lsm_object_id, lsm->lsm_object_gr, &res_id);
3058         ldlm_resource_iterate(obd->obd_namespace, &res_id, replace, data);
3059         return 0;
3060 }
3061
3062 static int osc_enqueue_fini(struct obd_device *obd, struct ptlrpc_request *req,
3063                             struct obd_info *oinfo, int intent, int rc)
3064 {
3065         ENTRY;
3066
3067         if (intent) {
3068                 /* The request was created before ldlm_cli_enqueue call. */
3069                 if (rc == ELDLM_LOCK_ABORTED) {
3070                         struct ldlm_reply *rep;
3071                         rep = req_capsule_server_get(&req->rq_pill,
3072                                                      &RMF_DLM_REP);
3073
3074                         LASSERT(rep != NULL);
3075                         if (rep->lock_policy_res1)
3076                                 rc = rep->lock_policy_res1;
3077                 }
3078         }
3079
3080         if ((intent && rc == ELDLM_LOCK_ABORTED) || !rc) {
3081                 CDEBUG(D_INODE,"got kms "LPU64" blocks "LPU64" mtime "LPU64"\n",
3082                        oinfo->oi_md->lsm_oinfo[0]->loi_lvb.lvb_size,
3083                        oinfo->oi_md->lsm_oinfo[0]->loi_lvb.lvb_blocks,
3084                        oinfo->oi_md->lsm_oinfo[0]->loi_lvb.lvb_mtime);
3085         }
3086
3087         if (!rc)
3088                 cache_add_lock(obd->u.cli.cl_cache, oinfo->oi_lockh);
3089
3090         /* Call the update callback. */
3091         rc = oinfo->oi_cb_up(oinfo, rc);
3092         RETURN(rc);
3093 }
3094
3095 static int osc_enqueue_interpret(struct ptlrpc_request *req,
3096                                  struct osc_enqueue_args *aa, int rc)
3097 {
3098         int intent = aa->oa_oi->oi_flags & LDLM_FL_HAS_INTENT;
3099         struct lov_stripe_md *lsm = aa->oa_oi->oi_md;
3100         struct ldlm_lock *lock;
3101
3102         /* ldlm_cli_enqueue is holding a reference on the lock, so it must
3103          * be valid. */
3104         lock = ldlm_handle2lock(aa->oa_oi->oi_lockh);
3105
3106         /* Complete obtaining the lock procedure. */
3107         rc = ldlm_cli_enqueue_fini(aa->oa_exp, req, aa->oa_ei->ei_type, 1,
3108                                    aa->oa_ei->ei_mode,
3109                                    &aa->oa_oi->oi_flags,
3110                                    &lsm->lsm_oinfo[0]->loi_lvb,
3111                                    sizeof(lsm->lsm_oinfo[0]->loi_lvb),
3112                                    lustre_swab_ost_lvb,
3113                                    aa->oa_oi->oi_lockh, rc);
3114
3115         /* Complete osc stuff. */
3116         rc = osc_enqueue_fini(aa->oa_exp->exp_obd, req, aa->oa_oi, intent, rc);
3117
3118         /* Release the lock for async request. */
3119         if (lustre_handle_is_used(aa->oa_oi->oi_lockh) && rc == ELDLM_OK)
3120                 ldlm_lock_decref(aa->oa_oi->oi_lockh, aa->oa_ei->ei_mode);
3121
3122         LASSERTF(lock != NULL, "lockh %p, req %p, aa %p - client evicted?\n",
3123                  aa->oa_oi->oi_lockh, req, aa);
3124         LDLM_LOCK_PUT(lock);
3125         return rc;
3126 }
3127
3128 /* When enqueuing asynchronously, locks are not ordered, we can obtain a lock
3129  * from the 2nd OSC before a lock from the 1st one. This does not deadlock with
3130  * other synchronous requests, however keeping some locks and trying to obtain
3131  * others may take a considerable amount of time in a case of ost failure; and
3132  * when other sync requests do not get released lock from a client, the client
3133  * is excluded from the cluster -- such scenarious make the life difficult, so
3134  * release locks just after they are obtained. */
3135 static int osc_enqueue(struct obd_export *exp, struct obd_info *oinfo,
3136                        struct ldlm_enqueue_info *einfo,
3137                        struct ptlrpc_request_set *rqset)
3138 {
3139         struct ldlm_res_id res_id;
3140         struct obd_device *obd = exp->exp_obd;
3141         struct ptlrpc_request *req = NULL;
3142         int intent = oinfo->oi_flags & LDLM_FL_HAS_INTENT;
3143         ldlm_mode_t mode;
3144         int rc;
3145         ENTRY;
3146
3147
3148         osc_build_res_name(oinfo->oi_md->lsm_object_id,
3149                            oinfo->oi_md->lsm_object_gr, &res_id);
3150         /* Filesystem lock extents are extended to page boundaries so that
3151          * dealing with the page cache is a little smoother.  */
3152         oinfo->oi_policy.l_extent.start -=
3153                 oinfo->oi_policy.l_extent.start & ~CFS_PAGE_MASK;
3154         oinfo->oi_policy.l_extent.end |= ~CFS_PAGE_MASK;
3155
3156         if (oinfo->oi_md->lsm_oinfo[0]->loi_kms_valid == 0)
3157                 goto no_match;
3158
3159         /* Next, search for already existing extent locks that will cover us */
3160         /* If we're trying to read, we also search for an existing PW lock.  The
3161          * VFS and page cache already protect us locally, so lots of readers/
3162          * writers can share a single PW lock.
3163          *
3164          * There are problems with conversion deadlocks, so instead of
3165          * converting a read lock to a write lock, we'll just enqueue a new
3166          * one.
3167          *
3168          * At some point we should cancel the read lock instead of making them
3169          * send us a blocking callback, but there are problems with canceling
3170          * locks out from other users right now, too. */
3171         mode = einfo->ei_mode;
3172         if (einfo->ei_mode == LCK_PR)
3173                 mode |= LCK_PW;
3174         mode = ldlm_lock_match(obd->obd_namespace,
3175                                oinfo->oi_flags | LDLM_FL_LVB_READY, &res_id,
3176                                einfo->ei_type, &oinfo->oi_policy, mode,
3177                                oinfo->oi_lockh);
3178         if (mode) {
3179                 /* addref the lock only if not async requests and PW lock is
3180                  * matched whereas we asked for PR. */
3181                 if (!rqset && einfo->ei_mode != mode)
3182                         ldlm_lock_addref(oinfo->oi_lockh, LCK_PR);
3183                 osc_set_data_with_check(oinfo->oi_lockh, einfo->ei_cbdata,
3184                                         oinfo->oi_flags);
3185                 if (intent) {
3186                         /* I would like to be able to ASSERT here that rss <=
3187                          * kms, but I can't, for reasons which are explained in
3188                          * lov_enqueue() */
3189                 }
3190
3191                 /* We already have a lock, and it's referenced */
3192                 oinfo->oi_cb_up(oinfo, ELDLM_OK);
3193
3194                 /* For async requests, decref the lock. */
3195                 if (einfo->ei_mode != mode)
3196                         ldlm_lock_decref(oinfo->oi_lockh, LCK_PW);
3197                 else if (rqset)
3198                         ldlm_lock_decref(oinfo->oi_lockh, einfo->ei_mode);
3199
3200                 RETURN(ELDLM_OK);
3201         }
3202
3203  no_match:
3204         if (intent) {
3205                 CFS_LIST_HEAD(cancels);
3206                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
3207                                            &RQF_LDLM_ENQUEUE_LVB);
3208                 if (req == NULL)
3209                         RETURN(-ENOMEM);
3210
3211                 rc = ldlm_prep_enqueue_req(exp, req, &cancels, 0);
3212                 if (rc)
3213                         RETURN(rc);
3214
3215                 req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
3216                                      sizeof(oinfo->oi_md->lsm_oinfo[0]->loi_lvb));
3217                 ptlrpc_request_set_replen(req);
3218         }
3219
3220         /* users of osc_enqueue() can pass this flag for ldlm_lock_match() */
3221         oinfo->oi_flags &= ~LDLM_FL_BLOCK_GRANTED;
3222
3223         rc = ldlm_cli_enqueue(exp, &req, einfo, &res_id,
3224                               &oinfo->oi_policy, &oinfo->oi_flags,
3225                               &oinfo->oi_md->lsm_oinfo[0]->loi_lvb,
3226                               sizeof(oinfo->oi_md->lsm_oinfo[0]->loi_lvb),
3227                               lustre_swab_ost_lvb, oinfo->oi_lockh,
3228                               rqset ? 1 : 0);
3229         if (rqset) {
3230                 if (!rc) {
3231                         struct osc_enqueue_args *aa;
3232                         CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
3233                         aa = ptlrpc_req_async_args(req);
3234                         aa->oa_oi = oinfo;
3235                         aa->oa_ei = einfo;
3236                         aa->oa_exp = exp;
3237
3238                         req->rq_interpret_reply = osc_enqueue_interpret;
3239                         ptlrpc_set_add_req(rqset, req);
3240                 } else if (intent) {
3241                         ptlrpc_req_finished(req);
3242                 }
3243                 RETURN(rc);
3244         }
3245
3246         rc = osc_enqueue_fini(obd, req, oinfo, intent, rc);
3247         if (intent)
3248                 ptlrpc_req_finished(req);
3249
3250         RETURN(rc);
3251 }
3252
3253 static int osc_match(struct obd_export *exp, struct lov_stripe_md *lsm,
3254                      __u32 type, ldlm_policy_data_t *policy, __u32 mode,
3255                      int *flags, void *data, struct lustre_handle *lockh)
3256 {
3257         struct ldlm_res_id res_id;
3258         struct obd_device *obd = exp->exp_obd;
3259         int lflags = *flags;
3260         ldlm_mode_t rc;
3261         ENTRY;
3262
3263         osc_build_res_name(lsm->lsm_object_id, lsm->lsm_object_gr, &res_id);
3264
3265         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_MATCH))
3266                 RETURN(-EIO);
3267
3268         /* Filesystem lock extents are extended to page boundaries so that
3269          * dealing with the page cache is a little smoother */
3270         policy->l_extent.start -= policy->l_extent.start & ~CFS_PAGE_MASK;
3271         policy->l_extent.end |= ~CFS_PAGE_MASK;
3272
3273         /* Next, search for already existing extent locks that will cover us */
3274         /* If we're trying to read, we also search for an existing PW lock.  The
3275          * VFS and page cache already protect us locally, so lots of readers/
3276          * writers can share a single PW lock. */
3277         rc = mode;
3278         if (mode == LCK_PR)
3279                 rc |= LCK_PW;
3280         rc = ldlm_lock_match(obd->obd_namespace, lflags | LDLM_FL_LVB_READY,
3281                              &res_id, type, policy, rc, lockh);
3282         if (rc) {
3283                 osc_set_data_with_check(lockh, data, lflags);
3284                 if (!(lflags & LDLM_FL_TEST_LOCK) && mode != rc) {
3285                         ldlm_lock_addref(lockh, LCK_PR);
3286                         ldlm_lock_decref(lockh, LCK_PW);
3287                 }
3288                 RETURN(rc);
3289         }
3290         RETURN(rc);
3291 }
3292
3293 static int osc_cancel(struct obd_export *exp, struct lov_stripe_md *md,
3294                       __u32 mode, struct lustre_handle *lockh)
3295 {
3296         ENTRY;
3297
3298         if (unlikely(mode == LCK_GROUP))
3299                 ldlm_lock_decref_and_cancel(lockh, mode);
3300         else
3301                 ldlm_lock_decref(lockh, mode);
3302
3303         RETURN(0);
3304 }
3305
3306 static int osc_cancel_unused(struct obd_export *exp,
3307                              struct lov_stripe_md *lsm, int flags,
3308                              void *opaque)
3309 {
3310         struct obd_device *obd = class_exp2obd(exp);
3311         struct ldlm_res_id res_id, *resp = NULL;
3312
3313         if (lsm != NULL) {
3314                 resp = osc_build_res_name(lsm->lsm_object_id,
3315                                           lsm->lsm_object_gr, &res_id);
3316         }
3317
3318         return ldlm_cli_cancel_unused(obd->obd_namespace, resp, flags, opaque);
3319 }
3320
3321 static int osc_statfs_interpret(const struct lu_env *env,
3322                                 struct ptlrpc_request *req,
3323                                 struct osc_async_args *aa, int rc)
3324 {
3325         struct obd_statfs *msfs;
3326         ENTRY;
3327
3328         if (rc != 0)
3329                 GOTO(out, rc);
3330
3331         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
3332         if (msfs == NULL) {
3333                 GOTO(out, rc = -EPROTO);
3334         }
3335
3336         *aa->aa_oi->oi_osfs = *msfs;
3337 out:
3338         rc = aa->aa_oi->oi_cb_up(aa->aa_oi, rc);
3339         RETURN(rc);
3340 }
3341
3342 static int osc_statfs_async(struct obd_device *obd, struct obd_info *oinfo,
3343                             __u64 max_age, struct ptlrpc_request_set *rqset)
3344 {
3345         struct ptlrpc_request *req;
3346         struct osc_async_args *aa;
3347         int                    rc;
3348         ENTRY;
3349
3350         /* We could possibly pass max_age in the request (as an absolute
3351          * timestamp or a "seconds.usec ago") so the target can avoid doing
3352          * extra calls into the filesystem if that isn't necessary (e.g.
3353          * during mount that would help a bit).  Having relative timestamps
3354          * is not so great if request processing is slow, while absolute
3355          * timestamps are not ideal because they need time synchronization. */
3356         req = ptlrpc_request_alloc(obd->u.cli.cl_import, &RQF_OST_STATFS);
3357         if (req == NULL)
3358                 RETURN(-ENOMEM);
3359
3360         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
3361         if (rc) {
3362                 ptlrpc_request_free(req);
3363                 RETURN(rc);
3364         }
3365         ptlrpc_request_set_replen(req);
3366         req->rq_request_portal = OST_CREATE_PORTAL;
3367         ptlrpc_at_set_req_timeout(req);
3368
3369         if (oinfo->oi_flags & OBD_STATFS_NODELAY) {
3370                 /* procfs requests not want stat in wait for avoid deadlock */
3371                 req->rq_no_resend = 1;
3372                 req->rq_no_delay = 1;
3373         }
3374
3375         req->rq_interpret_reply = osc_statfs_interpret;
3376         CLASSERT (sizeof(*aa) <= sizeof(req->rq_async_args));
3377         aa = ptlrpc_req_async_args(req);
3378         aa->aa_oi = oinfo;
3379
3380         ptlrpc_set_add_req(rqset, req);
3381         RETURN(0);
3382 }
3383
3384 static int osc_statfs(struct obd_device *obd, struct obd_statfs *osfs,
3385                       __u64 max_age, __u32 flags)
3386 {
3387         struct obd_statfs     *msfs;
3388         struct ptlrpc_request *req;
3389         struct obd_import     *imp = NULL;
3390         int rc;
3391         ENTRY;
3392
3393         /*Since the request might also come from lprocfs, so we need
3394          *sync this with client_disconnect_export Bug15684*/
3395         down_read(&obd->u.cli.cl_sem);
3396         if (obd->u.cli.cl_import)
3397                 imp = class_import_get(obd->u.cli.cl_import);
3398         up_read(&obd->u.cli.cl_sem);
3399         if (!imp)
3400                 RETURN(-ENODEV);
3401
3402         /* We could possibly pass max_age in the request (as an absolute
3403          * timestamp or a "seconds.usec ago") so the target can avoid doing
3404          * extra calls into the filesystem if that isn't necessary (e.g.
3405          * during mount that would help a bit).  Having relative timestamps
3406          * is not so great if request processing is slow, while absolute
3407          * timestamps are not ideal because they need time synchronization. */
3408         req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
3409
3410         class_import_put(imp);
3411
3412         if (req == NULL)
3413                 RETURN(-ENOMEM);
3414
3415         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
3416         if (rc) {
3417                 ptlrpc_request_free(req);
3418                 RETURN(rc);
3419         }
3420         ptlrpc_request_set_replen(req);
3421         req->rq_request_portal = OST_CREATE_PORTAL;
3422         ptlrpc_at_set_req_timeout(req);
3423
3424         if (flags & OBD_STATFS_NODELAY) {
3425                 /* procfs requests not want stat in wait for avoid deadlock */
3426                 req->rq_no_resend = 1;
3427                 req->rq_no_delay = 1;
3428         }
3429
3430         rc = ptlrpc_queue_wait(req);
3431         if (rc)
3432                 GOTO(out, rc);
3433
3434         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
3435         if (msfs == NULL) {
3436                 GOTO(out, rc = -EPROTO);
3437         }
3438
3439         *osfs = *msfs;
3440
3441         EXIT;
3442  out:
3443         ptlrpc_req_finished(req);
3444         return rc;
3445 }
3446
3447 /* Retrieve object striping information.
3448  *
3449  * @lmmu is a pointer to an in-core struct with lmm_ost_count indicating
3450  * the maximum number of OST indices which will fit in the user buffer.
3451  * lmm_magic must be LOV_MAGIC (we only use 1 slot here).
3452  */
3453 static int osc_getstripe(struct lov_stripe_md *lsm, struct lov_user_md *lump)
3454 {
3455         /* we use lov_user_md_v3 because it is larger than lov_user_md_v1 */
3456         struct lov_user_md_v3 lum, *lumk;
3457         struct lov_user_ost_data_v1 *lmm_objects;
3458         int rc = 0, lum_size;
3459         ENTRY;
3460
3461         if (!lsm)
3462                 RETURN(-ENODATA);
3463
3464         /* we only need the header part from user space to get lmm_magic and
3465          * lmm_stripe_count, (the header part is common to v1 and v3) */
3466         lum_size = sizeof(struct lov_user_md_v1);
3467         if (copy_from_user(&lum, lump, lum_size))
3468                 RETURN(-EFAULT);
3469
3470         if ((lum.lmm_magic != LOV_USER_MAGIC_V1) &&
3471             (lum.lmm_magic != LOV_USER_MAGIC_V3))
3472                 RETURN(-EINVAL);
3473
3474         /* lov_user_md_vX and lov_mds_md_vX must have the same size */
3475         LASSERT(sizeof(struct lov_user_md_v1) == sizeof(struct lov_mds_md_v1));
3476         LASSERT(sizeof(struct lov_user_md_v3) == sizeof(struct lov_mds_md_v3));
3477         LASSERT(sizeof(lum.lmm_objects[0]) == sizeof(lumk->lmm_objects[0]));
3478
3479         /* we can use lov_mds_md_size() to compute lum_size
3480          * because lov_user_md_vX and lov_mds_md_vX have the same size */
3481         if (lum.lmm_stripe_count > 0) {
3482                 lum_size = lov_mds_md_size(lum.lmm_stripe_count, lum.lmm_magic);
3483                 OBD_ALLOC(lumk, lum_size);
3484                 if (!lumk)
3485                         RETURN(-ENOMEM);
3486
3487                 if (lum.lmm_magic == LOV_USER_MAGIC_V1)
3488                         lmm_objects = &(((struct lov_user_md_v1 *)lumk)->lmm_objects[0]);
3489                 else
3490                         lmm_objects = &(lumk->lmm_objects[0]);
3491                 lmm_objects->l_object_id = lsm->lsm_object_id;
3492         } else {
3493                 lum_size = lov_mds_md_size(0, lum.lmm_magic);
3494                 lumk = &lum;
3495         }
3496
3497         lumk->lmm_object_id = lsm->lsm_object_id;
3498         lumk->lmm_object_gr = lsm->lsm_object_gr;
3499         lumk->lmm_stripe_count = 1;
3500
3501         if (copy_to_user(lump, lumk, lum_size))
3502                 rc = -EFAULT;
3503
3504         if (lumk != &lum)
3505                 OBD_FREE(lumk, lum_size);
3506
3507         RETURN(rc);
3508 }
3509
3510
3511 static int osc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
3512                          void *karg, void *uarg)
3513 {
3514         struct obd_device *obd = exp->exp_obd;
3515         struct obd_ioctl_data *data = karg;
3516         int err = 0;
3517         ENTRY;
3518
3519         if (!try_module_get(THIS_MODULE)) {
3520                 CERROR("Can't get module. Is it alive?");
3521                 return -EINVAL;
3522         }
3523         switch (cmd) {
3524         case OBD_IOC_LOV_GET_CONFIG: {
3525                 char *buf;
3526                 struct lov_desc *desc;
3527                 struct obd_uuid uuid;
3528
3529                 buf = NULL;
3530                 len = 0;
3531                 if (obd_ioctl_getdata(&buf, &len, (void *)uarg))
3532                         GOTO(out, err = -EINVAL);
3533
3534                 data = (struct obd_ioctl_data *)buf;
3535
3536                 if (sizeof(*desc) > data->ioc_inllen1) {
3537                         obd_ioctl_freedata(buf, len);
3538                         GOTO(out, err = -EINVAL);
3539                 }
3540
3541                 if (data->ioc_inllen2 < sizeof(uuid)) {
3542                         obd_ioctl_freedata(buf, len);
3543                         GOTO(out, err = -EINVAL);
3544                 }
3545
3546                 desc = (struct lov_desc *)data->ioc_inlbuf1;
3547                 desc->ld_tgt_count = 1;
3548                 desc->ld_active_tgt_count = 1;
3549                 desc->ld_default_stripe_count = 1;
3550                 desc->ld_default_stripe_size = 0;
3551                 desc->ld_default_stripe_offset = 0;
3552                 desc->ld_pattern = 0;
3553                 memcpy(&desc->ld_uuid, &obd->obd_uuid, sizeof(uuid));
3554
3555                 memcpy(data->ioc_inlbuf2, &obd->obd_uuid, sizeof(uuid));
3556
3557                 err = copy_to_user((void *)uarg, buf, len);
3558                 if (err)
3559                         err = -EFAULT;
3560                 obd_ioctl_freedata(buf, len);
3561                 GOTO(out, err);
3562         }
3563         case LL_IOC_LOV_SETSTRIPE:
3564                 err = obd_alloc_memmd(exp, karg);
3565                 if (err > 0)
3566                         err = 0;
3567                 GOTO(out, err);
3568         case LL_IOC_LOV_GETSTRIPE:
3569                 err = osc_getstripe(karg, uarg);
3570                 GOTO(out, err);
3571         case OBD_IOC_CLIENT_RECOVER:
3572                 err = ptlrpc_recover_import(obd->u.cli.cl_import,
3573                                             data->ioc_inlbuf1);
3574                 if (err > 0)
3575                         err = 0;
3576                 GOTO(out, err);
3577         case IOC_OSC_SET_ACTIVE:
3578                 err = ptlrpc_set_import_active(obd->u.cli.cl_import,
3579                                                data->ioc_offset);
3580                 GOTO(out, err);
3581         case OBD_IOC_POLL_QUOTACHECK:
3582                 err = lquota_poll_check(quota_interface, exp,
3583                                         (struct if_quotacheck *)karg);
3584                 GOTO(out, err);
3585         default:
3586                 CDEBUG(D_INODE, "unrecognised ioctl %#x by %s\n",
3587                        cmd, cfs_curproc_comm());
3588                 GOTO(out, err = -ENOTTY);
3589         }
3590 out:
3591         module_put(THIS_MODULE);
3592         return err;
3593 }
3594
3595 static int osc_get_info(struct obd_export *exp, obd_count keylen,
3596                         void *key, __u32 *vallen, void *val,
3597                         struct lov_stripe_md *lsm)
3598 {
3599         ENTRY;
3600         if (!vallen || !val)
3601                 RETURN(-EFAULT);
3602
3603         if (KEY_IS(KEY_LOCK_TO_STRIPE)) {
3604                 __u32 *stripe = val;
3605                 *vallen = sizeof(*stripe);
3606                 *stripe = 0;
3607                 RETURN(0);
3608         } else if (KEY_IS(KEY_LAST_ID)) {
3609                 struct ptlrpc_request *req;
3610                 obd_id                *reply;
3611                 char                  *tmp;
3612                 int                    rc;
3613
3614                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
3615                                            &RQF_OST_GET_INFO_LAST_ID);
3616                 if (req == NULL)
3617                         RETURN(-ENOMEM);
3618
3619                 req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
3620                                      RCL_CLIENT, keylen);
3621                 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GET_INFO);
3622                 if (rc) {
3623                         ptlrpc_request_free(req);
3624                         RETURN(rc);
3625                 }
3626
3627                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
3628                 memcpy(tmp, key, keylen);
3629
3630                 ptlrpc_request_set_replen(req);
3631                 rc = ptlrpc_queue_wait(req);
3632                 if (rc)
3633                         GOTO(out, rc);
3634
3635                 reply = req_capsule_server_get(&req->rq_pill, &RMF_OBD_ID);
3636                 if (reply == NULL)
3637                         GOTO(out, rc = -EPROTO);
3638
3639                 *((obd_id *)val) = *reply;
3640         out:
3641                 ptlrpc_req_finished(req);
3642                 RETURN(rc);
3643         } else if (KEY_IS(KEY_FIEMAP)) {
3644                 struct ptlrpc_request *req;
3645                 struct ll_user_fiemap *reply;
3646                 char *tmp;
3647                 int rc;
3648
3649                 req = ptlrpc_request_alloc(class_exp2cliimp(exp),
3650                                            &RQF_OST_GET_INFO_FIEMAP);
3651                 if (req == NULL)
3652                         RETURN(-ENOMEM);
3653
3654                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_KEY,
3655                                      RCL_CLIENT, keylen);
3656                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_VAL,
3657                                      RCL_CLIENT, *vallen);
3658                 req_capsule_set_size(&req->rq_pill, &RMF_FIEMAP_VAL,
3659                                      RCL_SERVER, *vallen);
3660
3661                 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GET_INFO);
3662                 if (rc) {
3663                         ptlrpc_request_free(req);
3664                         RETURN(rc);
3665                 }
3666
3667                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_FIEMAP_KEY);
3668                 memcpy(tmp, key, keylen);
3669                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_FIEMAP_VAL);
3670                 memcpy(tmp, val, *vallen);
3671
3672                 ptlrpc_request_set_replen(req);
3673                 rc = ptlrpc_queue_wait(req);
3674                 if (rc)
3675                         GOTO(out1, rc);
3676
3677                 reply = req_capsule_server_get(&req->rq_pill, &RMF_FIEMAP_VAL);
3678                 if (reply == NULL)
3679                         GOTO(out1, rc = -EPROTO);
3680
3681                 memcpy(val, reply, *vallen);
3682         out1:
3683                 ptlrpc_req_finished(req);
3684
3685                 RETURN(rc);
3686         }
3687
3688         RETURN(-EINVAL);
3689 }
3690
3691 static int osc_setinfo_mds_conn_interpret(struct ptlrpc_request *req,
3692                                           void *aa, int rc)
3693 {
3694         struct llog_ctxt *ctxt;
3695         struct obd_import *imp = req->rq_import;
3696         ENTRY;
3697
3698         if (rc != 0)
3699                 RETURN(rc);
3700
3701         ctxt = llog_get_context(imp->imp_obd, LLOG_MDS_OST_ORIG_CTXT);
3702         if (ctxt) {
3703                 if (rc == 0)
3704                         rc = llog_initiator_connect(ctxt);
3705                 else
3706                         CERROR("cannot establish connection for "
3707                                "ctxt %p: %d\n", ctxt, rc);
3708         }
3709
3710         llog_ctxt_put(ctxt);
3711         spin_lock(&imp->imp_lock);
3712         imp->imp_server_timeout = 1;
3713         imp->imp_pingable = 1;
3714         spin_unlock(&imp->imp_lock);
3715         CDEBUG(D_RPCTRACE, "pinging OST %s\n", obd2cli_tgt(imp->imp_obd));
3716
3717         RETURN(rc);
3718 }
3719
3720 static int osc_set_info_async(struct obd_export *exp, obd_count keylen,
3721                               void *key, obd_count vallen, void *val,
3722                               struct ptlrpc_request_set *set)
3723 {
3724         struct ptlrpc_request *req;
3725         struct obd_device     *obd = exp->exp_obd;
3726         struct obd_import     *imp = class_exp2cliimp(exp);
3727         char                  *tmp;
3728         int                    rc;
3729         ENTRY;
3730
3731         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_SHUTDOWN, 10);
3732
3733         if (KEY_IS(KEY_NEXT_ID)) {
3734                 if (vallen != sizeof(obd_id))
3735                         RETURN(-ERANGE);
3736                 if (val == NULL)
3737                         RETURN(-EINVAL);
3738                 obd->u.cli.cl_oscc.oscc_next_id = *((obd_id*)val) + 1;
3739                 CDEBUG(D_HA, "%s: set oscc_next_id = "LPU64"\n",
3740                        exp->exp_obd->obd_name,
3741                        obd->u.cli.cl_oscc.oscc_next_id);
3742
3743                 RETURN(0);
3744         }
3745
3746         if (KEY_IS(KEY_UNLINKED)) {
3747                 struct osc_creator *oscc = &obd->u.cli.cl_oscc;
3748                 spin_lock(&oscc->oscc_lock);
3749                 oscc->oscc_flags &= ~OSCC_FLAG_NOSPC;
3750                 spin_unlock(&oscc->oscc_lock);
3751                 RETURN(0);
3752         }
3753
3754         if (KEY_IS(KEY_INIT_RECOV)) {
3755                 if (vallen != sizeof(int))
3756                         RETURN(-EINVAL);
3757                 spin_lock(&imp->imp_lock);
3758                 imp->imp_initial_recov = *(int *)val;
3759                 spin_unlock(&imp->imp_lock);
3760                 CDEBUG(D_HA, "%s: set imp_initial_recov = %d\n",
3761                        exp->exp_obd->obd_name,
3762                        imp->imp_initial_recov);
3763                 RETURN(0);
3764         }
3765
3766         if (KEY_IS(KEY_CHECKSUM)) {
3767                 if (vallen != sizeof(int))
3768                         RETURN(-EINVAL);
3769                 exp->exp_obd->u.cli.cl_checksum = (*(int *)val) ? 1 : 0;
3770                 RETURN(0);
3771         }
3772
3773         if (KEY_IS(KEY_FLUSH_CTX)) {
3774                 sptlrpc_import_flush_my_ctx(imp);
3775                 RETURN(0);
3776         }
3777
3778         if (!set)
3779                 RETURN(-EINVAL);
3780
3781         /* We pass all other commands directly to OST. Since nobody calls osc
3782            methods directly and everybody is supposed to go through LOV, we
3783            assume lov checked invalid values for us.
3784            The only recognised values so far are evict_by_nid and mds_conn.
3785            Even if something bad goes through, we'd get a -EINVAL from OST
3786            anyway. */
3787
3788
3789         req = ptlrpc_request_alloc(imp, &RQF_OST_SET_INFO);
3790         if (req == NULL)
3791                 RETURN(-ENOMEM);
3792
3793         req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
3794                              RCL_CLIENT, keylen);
3795         req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_VAL,
3796                              RCL_CLIENT, vallen);
3797         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SET_INFO);
3798         if (rc) {
3799                 ptlrpc_request_free(req);
3800                 RETURN(rc);
3801         }
3802
3803         tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
3804         memcpy(tmp, key, keylen);
3805         tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_VAL);
3806         memcpy(tmp, val, vallen);
3807
3808         if (KEY_IS(KEY_MDS_CONN)) {
3809                 struct osc_creator *oscc = &obd->u.cli.cl_oscc;
3810
3811                 oscc->oscc_oa.o_gr = (*(__u32 *)val);
3812                 oscc->oscc_oa.o_valid |= OBD_MD_FLGROUP;
3813                 LASSERT(oscc->oscc_oa.o_gr > 0);
3814                 req->rq_interpret_reply = osc_setinfo_mds_conn_interpret;
3815         }
3816
3817         ptlrpc_request_set_replen(req);
3818         ptlrpc_set_add_req(set, req);
3819         ptlrpc_check_set(set);
3820
3821         RETURN(0);
3822 }
3823
3824
3825 static struct llog_operations osc_size_repl_logops = {
3826         lop_cancel: llog_obd_repl_cancel
3827 };
3828
3829 static struct llog_operations osc_mds_ost_orig_logops;
3830 static int osc_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
3831                          struct obd_device *tgt, int count,
3832                          struct llog_catid *catid, struct obd_uuid *uuid)
3833 {
3834         int rc;
3835         ENTRY;
3836
3837         LASSERT(olg == &obd->obd_olg);
3838         spin_lock(&obd->obd_dev_lock);
3839         if (osc_mds_ost_orig_logops.lop_setup != llog_obd_origin_setup) {
3840                 osc_mds_ost_orig_logops = llog_lvfs_ops;
3841                 osc_mds_ost_orig_logops.lop_setup = llog_obd_origin_setup;
3842                 osc_mds_ost_orig_logops.lop_cleanup = llog_obd_origin_cleanup;
3843                 osc_mds_ost_orig_logops.lop_add = llog_obd_origin_add;
3844                 osc_mds_ost_orig_logops.lop_connect = llog_origin_connect;
3845         }
3846         spin_unlock(&obd->obd_dev_lock);
3847
3848         rc = llog_setup(obd, &obd->obd_olg, LLOG_MDS_OST_ORIG_CTXT, tgt, count,
3849                         &catid->lci_logid, &osc_mds_ost_orig_logops);
3850         if (rc) {
3851                 CERROR("failed LLOG_MDS_OST_ORIG_CTXT\n");
3852                 GOTO (out, rc);
3853         }
3854
3855         rc = llog_setup(obd, &obd->obd_olg, LLOG_SIZE_REPL_CTXT, tgt, count,
3856                         NULL, &osc_size_repl_logops);
3857         if (rc)
3858                 CERROR("failed LLOG_SIZE_REPL_CTXT\n");
3859 out:
3860         if (rc) {
3861                 CERROR("osc '%s' tgt '%s' cnt %d catid %p rc=%d\n",
3862                        obd->obd_name, tgt->obd_name, count, catid, rc);
3863                 CERROR("logid "LPX64":0x%x\n",
3864                        catid->lci_logid.lgl_oid, catid->lci_logid.lgl_ogen);
3865         }
3866         RETURN(rc);
3867 }
3868
3869 static int osc_llog_finish(struct obd_device *obd, int count)
3870 {
3871         struct llog_ctxt *ctxt;
3872         int rc = 0, rc2 = 0;
3873         ENTRY;
3874
3875         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
3876         if (ctxt)
3877                 rc = llog_cleanup(ctxt);
3878
3879         ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
3880         if (ctxt)
3881                 rc2 = llog_cleanup(ctxt);
3882         if (!rc)
3883                 rc = rc2;
3884
3885         RETURN(rc);
3886 }
3887
3888 static int osc_reconnect(const struct lu_env *env,
3889                          struct obd_export *exp, struct obd_device *obd,
3890                          struct obd_uuid *cluuid,
3891                          struct obd_connect_data *data)
3892 {
3893         struct client_obd *cli = &obd->u.cli;
3894
3895         if (data != NULL && (data->ocd_connect_flags & OBD_CONNECT_GRANT)) {
3896                 long lost_grant;
3897
3898                 client_obd_list_lock(&cli->cl_loi_list_lock);
3899                 data->ocd_grant = cli->cl_avail_grant ?:
3900                                 2 * cli->cl_max_pages_per_rpc << CFS_PAGE_SHIFT;
3901                 lost_grant = cli->cl_lost_grant;
3902                 cli->cl_lost_grant = 0;
3903                 client_obd_list_unlock(&cli->cl_loi_list_lock);
3904
3905                 CDEBUG(D_CACHE, "request ocd_grant: %d cl_avail_grant: %ld "
3906                        "cl_lost_grant: %ld\n", data->ocd_grant,
3907                        cli->cl_avail_grant, lost_grant);
3908                 CDEBUG(D_RPCTRACE, "ocd_connect_flags: "LPX64" ocd_version: %d"
3909                        " ocd_grant: %d\n", data->ocd_connect_flags,
3910                        data->ocd_version, data->ocd_grant);
3911         }
3912
3913         RETURN(0);
3914 }
3915
3916 static int osc_disconnect(struct obd_export *exp)
3917 {
3918         struct obd_device *obd = class_exp2obd(exp);
3919         struct llog_ctxt *ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
3920         int rc;
3921
3922         if (obd->u.cli.cl_conn_count == 1)
3923                 /* flush any remaining cancel messages out to the target */
3924                 llog_sync(ctxt, exp);
3925
3926         llog_ctxt_put(ctxt);
3927
3928         rc = client_disconnect_export(exp);
3929         return rc;
3930 }
3931
3932 static int osc_import_event(struct obd_device *obd,
3933                             struct obd_import *imp,
3934                             enum obd_import_event event)
3935 {
3936         struct client_obd *cli;
3937         int rc = 0;
3938
3939         ENTRY;
3940         LASSERT(imp->imp_obd == obd);
3941
3942         switch (event) {
3943         case IMP_EVENT_DISCON: {
3944                 /* Only do this on the MDS OSC's */
3945                 if (imp->imp_server_timeout) {
3946                         struct osc_creator *oscc = &obd->u.cli.cl_oscc;
3947
3948                         spin_lock(&oscc->oscc_lock);
3949                         oscc->oscc_flags |= OSCC_FLAG_RECOVERING;
3950                         spin_unlock(&oscc->oscc_lock);
3951                 }
3952                 cli = &obd->u.cli;
3953                 client_obd_list_lock(&cli->cl_loi_list_lock);
3954                 cli->cl_avail_grant = 0;
3955                 cli->cl_lost_grant = 0;
3956                 client_obd_list_unlock(&cli->cl_loi_list_lock);
3957                 break;
3958         }
3959         case IMP_EVENT_INACTIVE: {
3960                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_INACTIVE, NULL);
3961                 break;
3962         }
3963         case IMP_EVENT_INVALIDATE: {
3964                 struct ldlm_namespace *ns = obd->obd_namespace;
3965
3966                 /* Reset grants */
3967                 cli = &obd->u.cli;
3968                 client_obd_list_lock(&cli->cl_loi_list_lock);
3969                 /* all pages go to failing rpcs due to the invalid import */
3970                 osc_check_rpcs(cli);
3971                 client_obd_list_unlock(&cli->cl_loi_list_lock);
3972
3973                 ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
3974
3975                 break;
3976         }
3977         case IMP_EVENT_ACTIVE: {
3978                 /* Only do this on the MDS OSC's */
3979                 if (imp->imp_server_timeout) {
3980                         struct osc_creator *oscc = &obd->u.cli.cl_oscc;
3981
3982                         spin_lock(&oscc->oscc_lock);
3983                         oscc->oscc_flags &= ~OSCC_FLAG_NOSPC;
3984                         spin_unlock(&oscc->oscc_lock);
3985                 }
3986                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE, NULL);
3987                 break;
3988         }
3989         case IMP_EVENT_OCD: {
3990                 struct obd_connect_data *ocd = &imp->imp_connect_data;
3991
3992                 if (ocd->ocd_connect_flags & OBD_CONNECT_GRANT)
3993                         osc_init_grant(&obd->u.cli, ocd);
3994
3995                 /* See bug 7198 */
3996                 if (ocd->ocd_connect_flags & OBD_CONNECT_REQPORTAL)
3997                         imp->imp_client->cli_request_portal =OST_REQUEST_PORTAL;
3998
3999                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_OCD, NULL);
4000                 break;
4001         }
4002         default:
4003                 CERROR("Unknown import event %d\n", event);
4004                 LBUG();
4005         }
4006         RETURN(rc);
4007 }
4008
4009 int osc_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
4010 {
4011         int rc;
4012         ENTRY;
4013
4014         ENTRY;
4015         rc = ptlrpcd_addref();
4016         if (rc)
4017                 RETURN(rc);
4018
4019         rc = client_obd_setup(obd, lcfg);
4020         if (rc) {
4021                 ptlrpcd_decref();
4022         } else {
4023                 struct lprocfs_static_vars lvars = { 0 };
4024                 struct client_obd *cli = &obd->u.cli;
4025
4026                 lprocfs_osc_init_vars(&lvars);
4027                 if (lprocfs_obd_setup(obd, lvars.obd_vars) == 0) {
4028                         lproc_osc_attach_seqstat(obd);
4029                         sptlrpc_lprocfs_cliobd_attach(obd);
4030                         ptlrpc_lprocfs_register_obd(obd);
4031                 }
4032
4033                 oscc_init(obd);
4034                 /* We need to allocate a few requests more, because
4035                    brw_interpret tries to create new requests before freeing
4036                    previous ones. Ideally we want to have 2x max_rpcs_in_flight
4037                    reserved, but I afraid that might be too much wasted RAM
4038                    in fact, so 2 is just my guess and still should work. */
4039                 cli->cl_import->imp_rq_pool =
4040                         ptlrpc_init_rq_pool(cli->cl_max_rpcs_in_flight + 2,
4041                                             OST_MAXREQSIZE,
4042                                             ptlrpc_add_rqs_to_pool);
4043                 cli->cl_cache = cache_create(obd);
4044                 if (!cli->cl_cache) {
4045                         osc_cleanup(obd);
4046                         rc = -ENOMEM;
4047                 }
4048         }
4049
4050         RETURN(rc);
4051 }
4052
4053 static int osc_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
4054 {
4055         int rc = 0;
4056         ENTRY;
4057
4058         switch (stage) {
4059         case OBD_CLEANUP_EARLY: {
4060                 struct obd_import *imp;
4061                 imp = obd->u.cli.cl_import;
4062                 CDEBUG(D_HA, "Deactivating import %s\n", obd->obd_name);
4063                 /* ptlrpc_abort_inflight to stop an mds_lov_synchronize */
4064                 ptlrpc_deactivate_import(imp);
4065                 spin_lock(&imp->imp_lock);
4066                 imp->imp_pingable = 0;
4067                 spin_unlock(&imp->imp_lock);
4068                 break;
4069         }
4070         case OBD_CLEANUP_EXPORTS: {
4071                 /* If we set up but never connected, the
4072                    client import will not have been cleaned. */
4073                 if (obd->u.cli.cl_import) {
4074                         struct obd_import *imp;
4075                         imp = obd->u.cli.cl_import;
4076                         CDEBUG(D_CONFIG, "%s: client import never connected\n",
4077                                obd->obd_name);
4078                         ptlrpc_invalidate_import(imp);
4079                         ptlrpc_free_rq_pool(imp->imp_rq_pool);
4080                         class_destroy_import(imp);
4081                         obd->u.cli.cl_import = NULL;
4082                 }
4083                 rc = obd_llog_finish(obd, 0);
4084                 if (rc != 0)
4085                         CERROR("failed to cleanup llogging subsystems\n");
4086                 break;
4087                 }
4088         }
4089         RETURN(rc);
4090 }
4091
4092 int osc_cleanup(struct obd_device *obd)
4093 {
4094         struct osc_creator *oscc = &obd->u.cli.cl_oscc;
4095         int rc;
4096
4097         ENTRY;
4098         ptlrpc_lprocfs_unregister_obd(obd);
4099         lprocfs_obd_cleanup(obd);
4100
4101         spin_lock(&oscc->oscc_lock);
4102         oscc->oscc_flags &= ~OSCC_FLAG_RECOVERING;
4103         oscc->oscc_flags |= OSCC_FLAG_EXITING;
4104         spin_unlock(&oscc->oscc_lock);
4105
4106         /* free memory of osc quota cache */
4107         lquota_cleanup(quota_interface, obd);
4108
4109         cache_destroy(obd->u.cli.cl_cache);
4110         rc = client_obd_cleanup(obd);
4111
4112         ptlrpcd_decref();
4113         RETURN(rc);
4114 }
4115
4116 static int osc_register_page_removal_cb(struct obd_export *exp,
4117                                         obd_page_removal_cb_t func,
4118                                         obd_pin_extent_cb pin_cb)
4119 {
4120         return cache_add_extent_removal_cb(exp->exp_obd->u.cli.cl_cache, func,
4121                                            pin_cb);
4122 }
4123
4124 static int osc_unregister_page_removal_cb(struct obd_export *exp,
4125                                           obd_page_removal_cb_t func)
4126 {
4127         return cache_del_extent_removal_cb(exp->exp_obd->u.cli.cl_cache, func);
4128 }
4129
4130 static int osc_register_lock_cancel_cb(struct obd_export *exp,
4131                                        obd_lock_cancel_cb cb)
4132 {
4133         LASSERT(exp->exp_obd->u.cli.cl_ext_lock_cancel_cb == NULL);
4134
4135         exp->exp_obd->u.cli.cl_ext_lock_cancel_cb = cb;
4136         return 0;
4137 }
4138
4139 static int osc_unregister_lock_cancel_cb(struct obd_export *exp,
4140                                          obd_lock_cancel_cb cb)
4141 {
4142         if (exp->exp_obd->u.cli.cl_ext_lock_cancel_cb != cb) {
4143                 CERROR("Unregistering cancel cb %p, while only %p was "
4144                        "registered\n", cb,
4145                        exp->exp_obd->u.cli.cl_ext_lock_cancel_cb);
4146                 RETURN(-EINVAL);
4147         }
4148
4149         exp->exp_obd->u.cli.cl_ext_lock_cancel_cb = NULL;
4150         return 0;
4151 }
4152
4153 static int osc_process_config(struct obd_device *obd, obd_count len, void *buf)
4154 {
4155         struct lustre_cfg *lcfg = buf;
4156         struct lprocfs_static_vars lvars = { 0 };
4157         int rc = 0;
4158
4159         lprocfs_osc_init_vars(&lvars);
4160
4161         switch (lcfg->lcfg_command) {
4162         case LCFG_SPTLRPC_CONF:
4163                 rc = sptlrpc_cliobd_process_config(obd, lcfg);
4164                 break;
4165         default:
4166                 rc = class_process_proc_param(PARAM_OSC, lvars.obd_vars,
4167                                               lcfg, obd);
4168                 break;
4169         }
4170
4171         return(rc);
4172 }
4173
4174 struct obd_ops osc_obd_ops = {
4175         .o_owner                = THIS_MODULE,
4176         .o_setup                = osc_setup,
4177         .o_precleanup           = osc_precleanup,
4178         .o_cleanup              = osc_cleanup,
4179         .o_add_conn             = client_import_add_conn,
4180         .o_del_conn             = client_import_del_conn,
4181         .o_connect              = client_connect_import,
4182         .o_reconnect            = osc_reconnect,
4183         .o_disconnect           = osc_disconnect,
4184         .o_statfs               = osc_statfs,
4185         .o_statfs_async         = osc_statfs_async,
4186         .o_packmd               = osc_packmd,
4187         .o_unpackmd             = osc_unpackmd,
4188         .o_precreate            = osc_precreate,
4189         .o_create               = osc_create,
4190         .o_destroy              = osc_destroy,
4191         .o_getattr              = osc_getattr,
4192         .o_getattr_async        = osc_getattr_async,
4193         .o_setattr              = osc_setattr,
4194         .o_setattr_async        = osc_setattr_async,
4195         .o_brw                  = osc_brw,
4196         .o_brw_async            = osc_brw_async,
4197         .o_prep_async_page      = osc_prep_async_page,
4198         .o_reget_short_lock     = osc_reget_short_lock,
4199         .o_release_short_lock   = osc_release_short_lock,
4200         .o_queue_async_io       = osc_queue_async_io,
4201         .o_set_async_flags      = osc_set_async_flags,
4202         .o_queue_group_io       = osc_queue_group_io,
4203         .o_trigger_group_io     = osc_trigger_group_io,
4204         .o_teardown_async_page  = osc_teardown_async_page,
4205         .o_punch                = osc_punch,
4206         .o_sync                 = osc_sync,
4207         .o_enqueue              = osc_enqueue,
4208         .o_match                = osc_match,
4209         .o_change_cbdata        = osc_change_cbdata,
4210         .o_cancel               = osc_cancel,
4211         .o_cancel_unused        = osc_cancel_unused,
4212         .o_iocontrol            = osc_iocontrol,
4213         .o_get_info             = osc_get_info,
4214         .o_set_info_async       = osc_set_info_async,
4215         .o_import_event         = osc_import_event,
4216         .o_llog_init            = osc_llog_init,
4217         .o_llog_finish          = osc_llog_finish,
4218         .o_process_config       = osc_process_config,
4219         .o_register_page_removal_cb = osc_register_page_removal_cb,
4220         .o_unregister_page_removal_cb = osc_unregister_page_removal_cb,
4221         .o_register_lock_cancel_cb = osc_register_lock_cancel_cb,
4222         .o_unregister_lock_cancel_cb = osc_unregister_lock_cancel_cb,
4223 };
4224 int __init osc_init(void)
4225 {
4226         struct lprocfs_static_vars lvars = { 0 };
4227         int rc;
4228         ENTRY;
4229
4230         lprocfs_osc_init_vars(&lvars);
4231
4232         request_module("lquota");
4233         quota_interface = PORTAL_SYMBOL_GET(osc_quota_interface);
4234         lquota_init(quota_interface);
4235         init_obd_quota_ops(quota_interface, &osc_obd_ops);
4236
4237         rc = class_register_type(&osc_obd_ops, NULL, lvars.module_vars,
4238                                  LUSTRE_OSC_NAME, NULL);
4239         if (rc) {
4240                 if (quota_interface)
4241                         PORTAL_SYMBOL_PUT(osc_quota_interface);
4242                 RETURN(rc);
4243         }
4244
4245         RETURN(rc);
4246 }
4247
4248 #ifdef __KERNEL__
4249 static void /*__exit*/ osc_exit(void)
4250 {
4251         lquota_exit(quota_interface);
4252         if (quota_interface)
4253                 PORTAL_SYMBOL_PUT(osc_quota_interface);
4254
4255         class_unregister_type(LUSTRE_OSC_NAME);
4256 }
4257
4258 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
4259 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC)");
4260 MODULE_LICENSE("GPL");
4261
4262 cfs_module(osc, LUSTRE_VERSION_STRING, osc_init, osc_exit);
4263 #endif