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