lustre/osc/osc_request.c

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