lustre/osc/osc_request.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (C) 2001-2003 Cluster File Systems, Inc.
   5  *   Author Peter Braam <braam@clusterfs.com>
   6  *
   7  *   This file is part of Lustre, http://www.lustre.org.
   8  *
   9  *   Lustre is free software; you can redistribute it and/or
  10  *   modify it under the terms of version 2 of the GNU General Public
  11  *   License as published by the Free Software Foundation.
  12  *
  13  *   Lustre is distributed in the hope that it will be useful,
  14  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  *   GNU General Public License for more details.
  17  *
  18  *   You should have received a copy of the GNU General Public License
  19  *   along with Lustre; if not, write to the Free Software
  20  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  21  *
  22  *  For testing and management it is treated as an obd_device,
  23  *  although * it does not export a full OBD method table (the
  24  *  requests are coming * in over the wire, so object target modules
  25  *  do not have a full * method table.)
  26  *
  27  */
  28
  29 #ifndef EXPORT_SYMTAB
  30 # define EXPORT_SYMTAB
  31 #endif
  32 #define DEBUG_SUBSYSTEM S_OSC
  33
  34 #ifdef __KERNEL__
  35 # include <linux/version.h>
  36 # include <linux/module.h>
  37 # include <linux/mm.h>
  38 # include <linux/highmem.h>
  39 # include <linux/ctype.h>
  40 # include <linux/init.h>
  41 # if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
  42 #  include <linux/workqueue.h>
  43 #  include <linux/smp_lock.h>
  44 # else
  45 #  include <linux/locks.h>
  46 # endif
  47 #else /* __KERNEL__ */
  48 # include <liblustre.h>
  49 #endif
  50
  51 # include <linux/lustre_dlm.h>
  52 #include <linux/kp30.h>
  53 #include <linux/lustre_net.h>
  54 #include <lustre/lustre_user.h>
  55 #include <linux/obd_ost.h>
  56 #include <linux/obd_lov.h>
  57
  58 #ifdef  __CYGWIN__
  59 # include <ctype.h>
  60 #endif
  61
  62 #include <linux/lustre_ha.h>
  63 #include <linux/lprocfs_status.h>
  64 #include <linux/lustre_log.h>
  65 #include "osc_internal.h"
  66
  67
  68 static int osc_attach(struct obd_device *dev, obd_count len, void *data)
  69 {
  70         struct lprocfs_static_vars lvars;
  71         int rc;
  72         ENTRY;
  73
  74         lprocfs_init_vars(osc,&lvars);
  75         rc = lprocfs_obd_attach(dev, lvars.obd_vars);
  76         if (rc < 0)
  77                 RETURN(rc);
  78
  79         rc = lproc_osc_attach_seqstat(dev);
  80         if (rc < 0) {
  81                 lprocfs_obd_detach(dev);
  82                 RETURN(rc);
  83         }
  84
  85         ptlrpc_lprocfs_register_obd(dev);
  86         RETURN(0);
  87 }
  88
  89 static int osc_detach(struct obd_device *dev)
  90 {
  91         ptlrpc_lprocfs_unregister_obd(dev);
  92         return lprocfs_obd_detach(dev);
  93 }
  94
  95
  96 /* Pack OSC object metadata for disk storage (LE byte order). */
  97 static int osc_packmd(struct obd_export *exp, struct lov_mds_md **lmmp,
  98                       struct lov_stripe_md *lsm)
  99 {
 100         int lmm_size;
 101         ENTRY;
 102
 103         lmm_size = sizeof(**lmmp);
 104         if (!lmmp)
 105                 RETURN(lmm_size);
 106
 107         if (*lmmp && !lsm) {
 108                 OBD_FREE(*lmmp, lmm_size);
 109                 *lmmp = NULL;
 110                 RETURN(0);
 111         }
 112
 113         if (!*lmmp) {
 114                 OBD_ALLOC(*lmmp, lmm_size);
 115                 if (!*lmmp)
 116                         RETURN(-ENOMEM);
 117         }
 118
 119         if (lsm) {
 120                 LASSERT(lsm->lsm_object_id);
 121                 (*lmmp)->lmm_object_id = cpu_to_le64(lsm->lsm_object_id);
 122         }
 123
 124         RETURN(lmm_size);
 125 }
 126
 127 /* Unpack OSC object metadata from disk storage (LE byte order). */
 128 static int osc_unpackmd(struct obd_export *exp, struct lov_stripe_md **lsmp,
 129                         struct lov_mds_md *lmm, int lmm_bytes)
 130 {
 131         int lsm_size;
 132         ENTRY;
 133
 134         if (lmm != NULL) {
 135                 if (lmm_bytes < sizeof (*lmm)) {
 136                         CERROR("lov_mds_md too small: %d, need %d\n",
 137                                lmm_bytes, (int)sizeof(*lmm));
 138                         RETURN(-EINVAL);
 139                 }
 140                 /* XXX LOV_MAGIC etc check? */
 141
 142                 if (lmm->lmm_object_id == 0) {
 143                         CERROR("lov_mds_md: zero lmm_object_id\n");
 144                         RETURN(-EINVAL);
 145                 }
 146         }
 147
 148         lsm_size = lov_stripe_md_size(1);
 149         if (lsmp == NULL)
 150                 RETURN(lsm_size);
 151
 152         if (*lsmp != NULL && lmm == NULL) {
 153                 OBD_FREE(*lsmp, lsm_size);
 154                 *lsmp = NULL;
 155                 RETURN(0);
 156         }
 157
 158         if (*lsmp == NULL) {
 159                 OBD_ALLOC(*lsmp, lsm_size);
 160                 if (*lsmp == NULL)
 161                         RETURN(-ENOMEM);
 162                 loi_init((*lsmp)->lsm_oinfo);
 163         }
 164
 165         if (lmm != NULL) {
 166                 /* XXX zero *lsmp? */
 167                 (*lsmp)->lsm_object_id = le64_to_cpu (lmm->lmm_object_id);
 168                 LASSERT((*lsmp)->lsm_object_id);
 169         }
 170
 171         (*lsmp)->lsm_maxbytes = LUSTRE_STRIPE_MAXBYTES;
 172
 173         RETURN(lsm_size);
 174 }
 175
 176 static int osc_getattr_interpret(struct ptlrpc_request *req,
 177                                  struct osc_getattr_async_args *aa, int rc)
 178 {
 179         struct ost_body *body;
 180         ENTRY;
 181
 182         if (rc != 0)
 183                 RETURN(rc);
 184
 185         body = lustre_swab_repbuf(req, 0, sizeof(*body), lustre_swab_ost_body);
 186         if (body) {
 187                 CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
 188                 memcpy(aa->aa_oa, &body->oa, sizeof(*aa->aa_oa));
 189
 190                 /* This should really be sent by the OST */
 191                 aa->aa_oa->o_blksize = PTLRPC_MAX_BRW_SIZE;
 192                 aa->aa_oa->o_valid |= OBD_MD_FLBLKSZ;
 193         } else {
 194                 CERROR("can't unpack ost_body\n");
 195                 rc = -EPROTO;
 196                 aa->aa_oa->o_valid = 0;
 197         }
 198
 199         RETURN(rc);
 200 }
 201
 202 static int osc_getattr_async(struct obd_export *exp, struct obdo *oa,
 203                              struct lov_stripe_md *md,
 204                              struct ptlrpc_request_set *set)
 205 {
 206         struct ptlrpc_request *request;
 207         struct ost_body *body;
 208         int size = sizeof(*body);
 209         struct osc_getattr_async_args *aa;
 210         ENTRY;
 211
 212         request = ptlrpc_prep_req(class_exp2cliimp(exp), OST_GETATTR, 1,
 213                                   &size, NULL);
 214         if (!request)
 215                 RETURN(-ENOMEM);
 216
 217         body = lustre_msg_buf(request->rq_reqmsg, 0, sizeof (*body));
 218         memcpy(&body->oa, oa, sizeof(*oa));
 219
 220         request->rq_replen = lustre_msg_size(1, &size);
 221         request->rq_interpret_reply = osc_getattr_interpret;
 222
 223         LASSERT (sizeof (*aa) <= sizeof (request->rq_async_args));
 224         aa = (struct osc_getattr_async_args *)&request->rq_async_args;
 225         aa->aa_oa = oa;
 226
 227         ptlrpc_set_add_req (set, request);
 228         RETURN (0);
 229 }
 230
 231 static int osc_getattr(struct obd_export *exp, struct obdo *oa,
 232                        struct lov_stripe_md *md)
 233 {
 234         struct ptlrpc_request *request;
 235         struct ost_body *body;
 236         int rc, size = sizeof(*body);
 237         ENTRY;
 238
 239         request = ptlrpc_prep_req(class_exp2cliimp(exp), OST_GETATTR, 1,
 240                                   &size, NULL);
 241         if (!request)
 242                 RETURN(-ENOMEM);
 243
 244         body = lustre_msg_buf(request->rq_reqmsg, 0, sizeof (*body));
 245         memcpy(&body->oa, oa, sizeof(*oa));
 246
 247         request->rq_replen = lustre_msg_size(1, &size);
 248
 249         rc = ptlrpc_queue_wait(request);
 250         if (rc) {
 251                 CERROR("%s failed: rc = %d\n", __FUNCTION__, rc);
 252                 GOTO(out, rc);
 253         }
 254
 255         body = lustre_swab_repbuf(request, 0, sizeof (*body),
 256                                   lustre_swab_ost_body);
 257         if (body == NULL) {
 258                 CERROR ("can't unpack ost_body\n");
 259                 GOTO (out, rc = -EPROTO);
 260         }
 261
 262         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
 263         memcpy(oa, &body->oa, sizeof(*oa));
 264
 265         /* This should really be sent by the OST */
 266         oa->o_blksize = PTLRPC_MAX_BRW_SIZE;
 267         oa->o_valid |= OBD_MD_FLBLKSZ;
 268
 269         EXIT;
 270  out:
 271         ptlrpc_req_finished(request);
 272         return rc;
 273 }
 274
 275 static int osc_setattr(struct obd_export *exp, struct obdo *oa,
 276                        struct lov_stripe_md *md, struct obd_trans_info *oti)
 277 {
 278         struct ptlrpc_request *request;
 279         struct ost_body *body;
 280         int rc, size = sizeof(*body);
 281         ENTRY;
 282
 283         request = ptlrpc_prep_req(class_exp2cliimp(exp), OST_SETATTR, 1, &size,
 284                                   NULL);
 285         if (!request)
 286                 RETURN(-ENOMEM);
 287
 288         body = lustre_msg_buf(request->rq_reqmsg, 0, sizeof(*body));
 289         memcpy(&body->oa, oa, sizeof(*oa));
 290
 291         request->rq_replen = lustre_msg_size(1, &size);
 292
 293         rc = ptlrpc_queue_wait(request);
 294         if (rc)
 295                 GOTO(out, rc);
 296
 297         body = lustre_swab_repbuf(request, 0, sizeof(*body),
 298                                   lustre_swab_ost_body);
 299         if (body == NULL)
 300                 GOTO(out, rc = -EPROTO);
 301
 302         memcpy(oa, &body->oa, sizeof(*oa));
 303
 304         EXIT;
 305 out:
 306         ptlrpc_req_finished(request);
 307         RETURN(0);
 308 }
 309
 310 int osc_real_create(struct obd_export *exp, struct obdo *oa,
 311                     struct lov_stripe_md **ea, struct obd_trans_info *oti)
 312 {
 313         struct ptlrpc_request *request;
 314         struct ost_body *body;
 315         struct lov_stripe_md *lsm;
 316         int rc, size = sizeof(*body);
 317         ENTRY;
 318
 319         LASSERT(oa);
 320         LASSERT(ea);
 321
 322         lsm = *ea;
 323         if (!lsm) {
 324                 rc = obd_alloc_memmd(exp, &lsm);
 325                 if (rc < 0)
 326                         RETURN(rc);
 327         }
 328
 329         request = ptlrpc_prep_req(class_exp2cliimp(exp), OST_CREATE, 1, &size,
 330                                   NULL);
 331         if (!request)
 332                 GOTO(out, rc = -ENOMEM);
 333
 334         body = lustre_msg_buf(request->rq_reqmsg, 0, sizeof (*body));
 335         memcpy(&body->oa, oa, sizeof(body->oa));
 336
 337         request->rq_replen = lustre_msg_size(1, &size);
 338         if (oa->o_valid & OBD_MD_FLINLINE) {
 339                 LASSERT((oa->o_valid & OBD_MD_FLFLAGS) &&
 340                         oa->o_flags == OBD_FL_DELORPHAN);
 341                 DEBUG_REQ(D_HA, request,
 342                           "delorphan from OST integration");
 343         }
 344
 345         rc = ptlrpc_queue_wait(request);
 346         if (rc)
 347                 GOTO(out_req, rc);
 348
 349         body = lustre_swab_repbuf(request, 0, sizeof(*body),
 350                                   lustre_swab_ost_body);
 351         if (body == NULL) {
 352                 CERROR ("can't unpack ost_body\n");
 353                 GOTO (out_req, rc = -EPROTO);
 354         }
 355
 356         memcpy(oa, &body->oa, sizeof(*oa));
 357
 358         /* This should really be sent by the OST */
 359         oa->o_blksize = PTLRPC_MAX_BRW_SIZE;
 360         oa->o_valid |= OBD_MD_FLBLKSZ;
 361
 362         /* XXX LOV STACKING: the lsm that is passed to us from LOV does not
 363          * have valid lsm_oinfo data structs, so don't go touching that.
 364          * This needs to be fixed in a big way.
 365          */
 366         lsm->lsm_object_id = oa->o_id;
 367         *ea = lsm;
 368
 369         if (oti != NULL) {
 370                 oti->oti_transno = request->rq_repmsg->transno;
 371
 372                 if (oa->o_valid & OBD_MD_FLCOOKIE) {
 373                         if (!oti->oti_logcookies)
 374                                 oti_alloc_cookies(oti, 1);
 375                         memcpy(oti->oti_logcookies, obdo_logcookie(oa),
 376                                sizeof(oti->oti_onecookie));
 377                 }
 378         }
 379
 380         CDEBUG(D_HA, "transno: "LPD64"\n", request->rq_repmsg->transno);
 381         EXIT;
 382 out_req:
 383         ptlrpc_req_finished(request);
 384 out:
 385         if (rc && !*ea)
 386                 obd_free_memmd(exp, &lsm);
 387         return rc;
 388 }
 389
 390 static int osc_punch(struct obd_export *exp, struct obdo *oa,
 391                      struct lov_stripe_md *md, obd_size start,
 392                      obd_size end, struct obd_trans_info *oti)
 393 {
 394         struct ptlrpc_request *request;
 395         struct ost_body *body;
 396         int rc, size = sizeof(*body);
 397         ENTRY;
 398
 399         if (!oa) {
 400                 CERROR("oa NULL\n");
 401                 RETURN(-EINVAL);
 402         }
 403
 404         request = ptlrpc_prep_req(class_exp2cliimp(exp), OST_PUNCH, 1, &size,
 405                                   NULL);
 406         if (!request)
 407                 RETURN(-ENOMEM);
 408
 409         body = lustre_msg_buf(request->rq_reqmsg, 0, sizeof (*body));
 410         memcpy(&body->oa, oa, sizeof(*oa));
 411
 412         /* overload the size and blocks fields in the oa with start/end */
 413         body->oa.o_size = start;
 414         body->oa.o_blocks = end;
 415         body->oa.o_valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS);
 416
 417         request->rq_replen = lustre_msg_size(1, &size);
 418
 419         rc = ptlrpc_queue_wait(request);
 420         if (rc)
 421                 GOTO(out, rc);
 422
 423         body = lustre_swab_repbuf (request, 0, sizeof (*body),
 424                                    lustre_swab_ost_body);
 425         if (body == NULL) {
 426                 CERROR ("can't unpack ost_body\n");
 427                 GOTO (out, rc = -EPROTO);
 428         }
 429
 430         memcpy(oa, &body->oa, sizeof(*oa));
 431
 432         EXIT;
 433  out:
 434         ptlrpc_req_finished(request);
 435         return rc;
 436 }
 437
 438 static int osc_sync(struct obd_export *exp, struct obdo *oa,
 439                     struct lov_stripe_md *md, obd_size start, obd_size end)
 440 {
 441         struct ptlrpc_request *request;
 442         struct ost_body *body;
 443         int rc, size = sizeof(*body);
 444         ENTRY;
 445
 446         if (!oa) {
 447                 CERROR("oa NULL\n");
 448                 RETURN(-EINVAL);
 449         }
 450
 451         request = ptlrpc_prep_req(class_exp2cliimp(exp), OST_SYNC, 1, &size,
 452                                   NULL);
 453         if (!request)
 454                 RETURN(-ENOMEM);
 455
 456         body = lustre_msg_buf(request->rq_reqmsg, 0, sizeof (*body));
 457         memcpy(&body->oa, oa, sizeof(*oa));
 458
 459         /* overload the size and blocks fields in the oa with start/end */
 460         body->oa.o_size = start;
 461         body->oa.o_blocks = end;
 462         body->oa.o_valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS);
 463
 464         request->rq_replen = lustre_msg_size(1, &size);
 465
 466         rc = ptlrpc_queue_wait(request);
 467         if (rc)
 468                 GOTO(out, rc);
 469
 470         body = lustre_swab_repbuf(request, 0, sizeof(*body),
 471                                   lustre_swab_ost_body);
 472         if (body == NULL) {
 473                 CERROR ("can't unpack ost_body\n");
 474                 GOTO (out, rc = -EPROTO);
 475         }
 476
 477         memcpy(oa, &body->oa, sizeof(*oa));
 478
 479         EXIT;
 480  out:
 481         ptlrpc_req_finished(request);
 482         return rc;
 483 }
 484
 485 static int osc_destroy(struct obd_export *exp, struct obdo *oa,
 486                        struct lov_stripe_md *ea, struct obd_trans_info *oti)
 487 {
 488         struct ptlrpc_request *request;
 489         struct ost_body *body;
 490         int rc, size = sizeof(*body);
 491         ENTRY;
 492
 493         if (!oa) {
 494                 CERROR("oa NULL\n");
 495                 RETURN(-EINVAL);
 496         }
 497
 498         request = ptlrpc_prep_req(class_exp2cliimp(exp), OST_DESTROY, 1,
 499                                   &size, NULL);
 500         if (!request)
 501                 RETURN(-ENOMEM);
 502
 503         body = lustre_msg_buf(request->rq_reqmsg, 0, sizeof (*body));
 504
 505         if (oti != NULL && oa->o_valid & OBD_MD_FLCOOKIE) {
 506                 memcpy(obdo_logcookie(oa), oti->oti_logcookies,
 507                        sizeof(*oti->oti_logcookies));
 508                 oti->oti_logcookies++;
 509         }
 510
 511         memcpy(&body->oa, oa, sizeof(*oa));
 512         request->rq_replen = lustre_msg_size(1, &size);
 513
 514         rc = ptlrpc_queue_wait(request);
 515         if (rc)
 516                 GOTO(out, rc);
 517
 518         body = lustre_swab_repbuf(request, 0, sizeof(*body),
 519                                   lustre_swab_ost_body);
 520         if (body == NULL) {
 521                 CERROR ("Can't unpack body\n");
 522                 GOTO (out, rc = -EPROTO);
 523         }
 524
 525         memcpy(oa, &body->oa, sizeof(*oa));
 526
 527         EXIT;
 528  out:
 529         ptlrpc_req_finished(request);
 530         return rc;
 531 }
 532
 533 static void osc_announce_cached(struct client_obd *cli, struct obdo *oa,
 534                                 long writing_bytes)
 535 {
 536         obd_flag bits = OBD_MD_FLBLOCKS|OBD_MD_FLGRANT;
 537
 538         LASSERT(!(oa->o_valid & bits));
 539
 540         oa->o_valid |= bits;
 541         spin_lock(&cli->cl_loi_list_lock);
 542         oa->o_dirty = cli->cl_dirty;
 543         oa->o_undirty = cli->cl_dirty_max - oa->o_dirty;
 544         oa->o_grant = cli->cl_avail_grant;
 545         oa->o_dropped = cli->cl_lost_grant;
 546         cli->cl_lost_grant = 0;
 547         spin_unlock(&cli->cl_loi_list_lock);
 548         CDEBUG(D_CACHE,"dirty: "LPU64" undirty: %u dropped %u grant: "LPU64"\n",
 549                oa->o_dirty, oa->o_undirty, oa->o_dropped, oa->o_grant);
 550 }
 551
 552 /* caller must hold loi_list_lock */
 553 static void osc_consume_write_grant(struct client_obd *cli,
 554                                     struct osc_async_page *oap)
 555 {
 556         cli->cl_dirty += PAGE_SIZE;
 557         cli->cl_avail_grant -= PAGE_SIZE;
 558         oap->oap_brw_flags |= OBD_BRW_FROM_GRANT;
 559         CDEBUG(D_CACHE, "using %lu grant credits for oap %p\n", PAGE_SIZE, oap);
 560         LASSERT(cli->cl_avail_grant >= 0);
 561 }
 562
 563 /* caller must hold loi_list_lock */
 564 void osc_wake_cache_waiters(struct client_obd *cli)
 565 {
 566         struct list_head *l, *tmp;
 567         struct osc_cache_waiter *ocw;
 568
 569         list_for_each_safe(l, tmp, &cli->cl_cache_waiters) {
 570                 /* if we can't dirty more, we must wait until some is written */
 571                 if (cli->cl_dirty + PAGE_SIZE > cli->cl_dirty_max) {
 572                         CDEBUG(D_CACHE, "no dirty room: dirty: %ld max %ld\n",
 573                                cli->cl_dirty, cli->cl_dirty_max);
 574                         return;
 575                 }
 576
 577                 /* if still dirty cache but no grant wait for pending RPCs that
 578                  * may yet return us some grant before doing sync writes */
 579                 if (cli->cl_brw_in_flight && cli->cl_avail_grant < PAGE_SIZE) {
 580                         CDEBUG(D_CACHE, "%d BRWs in flight, no grant\n",
 581                                cli->cl_brw_in_flight);
 582                         return;
 583                 }
 584
 585                 ocw = list_entry(l, struct osc_cache_waiter, ocw_entry);
 586                 list_del_init(&ocw->ocw_entry);
 587                 if (cli->cl_avail_grant < PAGE_SIZE) {
 588                         /* no more RPCs in flight to return grant, do sync IO */
 589                         ocw->ocw_rc = -EDQUOT;
 590                         CDEBUG(D_INODE, "wake oap %p for sync\n", ocw->ocw_oap);
 591                 } else {
 592                         osc_consume_write_grant(cli, ocw->ocw_oap);
 593                 }
 594
 595                 wake_up(&ocw->ocw_waitq);
 596         }
 597
 598         EXIT;
 599 }
 600
 601 static void osc_update_grant(struct client_obd *cli, struct ost_body *body)
 602 {
 603         spin_lock(&cli->cl_loi_list_lock);
 604         CDEBUG(D_CACHE, "got "LPU64" extra grant\n", body->oa.o_grant);
 605         cli->cl_avail_grant += body->oa.o_grant;
 606         /* waiters are woken in brw_interpret_oap */
 607         spin_unlock(&cli->cl_loi_list_lock);
 608 }
 609
 610 /* We assume that the reason this OSC got a short read is because it read
 611  * beyond the end of a stripe file; i.e. lustre is reading a sparse file
 612  * via the LOV, and it _knows_ it's reading inside the file, it's just that
 613  * this stripe never got written at or beyond this stripe offset yet. */
 614 static void handle_short_read(int nob_read, obd_count page_count,
 615                               struct brw_page *pga)
 616 {
 617         char *ptr;
 618
 619         /* skip bytes read OK */
 620         while (nob_read > 0) {
 621                 LASSERT (page_count > 0);
 622
 623                 if (pga->count > nob_read) {
 624                         /* EOF inside this page */
 625                         ptr = kmap(pga->pg) + (pga->off & ~PAGE_MASK);
 626                         memset(ptr + nob_read, 0, pga->count - nob_read);
 627                         kunmap(pga->pg);
 628                         page_count--;
 629                         pga++;
 630                         break;
 631                 }
 632
 633                 nob_read -= pga->count;
 634                 page_count--;
 635                 pga++;
 636         }
 637
 638         /* zero remaining pages */
 639         while (page_count-- > 0) {
 640                 ptr = kmap(pga->pg) + (pga->off & ~PAGE_MASK);
 641                 memset(ptr, 0, pga->count);
 642                 kunmap(pga->pg);
 643                 pga++;
 644         }
 645 }
 646
 647 static int check_write_rcs(struct ptlrpc_request *request,
 648                            int requested_nob, int niocount,
 649                            obd_count page_count, struct brw_page *pga)
 650 {
 651         int    *remote_rcs, i;
 652
 653         /* return error if any niobuf was in error */
 654         remote_rcs = lustre_swab_repbuf(request, 1,
 655                                         sizeof(*remote_rcs) * niocount, NULL);
 656         if (remote_rcs == NULL) {
 657                 CERROR("Missing/short RC vector on BRW_WRITE reply\n");
 658                 return(-EPROTO);
 659         }
 660         if (lustre_msg_swabbed(request->rq_repmsg))
 661                 for (i = 0; i < niocount; i++)
 662                         __swab32s(&remote_rcs[i]);
 663
 664         for (i = 0; i < niocount; i++) {
 665                 if (remote_rcs[i] < 0)
 666                         return(remote_rcs[i]);
 667
 668                 if (remote_rcs[i] != 0) {
 669                         CERROR("rc[%d] invalid (%d) req %p\n",
 670                                 i, remote_rcs[i], request);
 671                         return(-EPROTO);
 672                 }
 673         }
 674
 675         if (request->rq_bulk->bd_nob_transferred != requested_nob) {
 676                 CERROR("Unexpected # bytes transferred: %d (requested %d)\n",
 677                        requested_nob, request->rq_bulk->bd_nob_transferred);
 678                 return(-EPROTO);
 679         }
 680
 681         return (0);
 682 }
 683
 684 static inline int can_merge_pages(struct brw_page *p1, struct brw_page *p2)
 685 {
 686         if (p1->flag != p2->flag) {
 687                 unsigned mask = ~OBD_BRW_FROM_GRANT;
 688
 689                 /* warn if we try to combine flags that we don't know to be
 690                  * safe to combine */
 691                 if ((p1->flag & mask) != (p2->flag & mask))
 692                         CERROR("is it ok to have flags 0x%x and 0x%x in the "
 693                                "same brw?\n", p1->flag, p2->flag);
 694                 return 0;
 695         }
 696
 697         return (p1->off + p1->count == p2->off);
 698 }
 699
 700 #if CHECKSUM_BULK
 701 static obd_count cksum_pages(int nob, obd_count page_count,
 702                              struct brw_page *pga)
 703 {
 704         obd_count cksum = 0;
 705         char *ptr;
 706
 707         while (nob > 0) {
 708                 LASSERT (page_count > 0);
 709
 710                 ptr = kmap(pga->pg);
 711                 ost_checksum(&cksum, ptr + (pga->off & (PAGE_SIZE - 1)),
 712                              pga->count > nob ? nob : pga->count);
 713                 kunmap(pga->pg);
 714
 715                 nob -= pga->count;
 716                 page_count--;
 717                 pga++;
 718         }
 719
 720         return (cksum);
 721 }
 722 #endif
 723
 724 static int osc_brw_prep_request(int cmd, struct obd_import *imp,struct obdo *oa,
 725                                 struct lov_stripe_md *lsm, obd_count page_count,
 726                                 struct brw_page *pga, int *requested_nobp,
 727                                 int *niocountp, struct ptlrpc_request **reqp)
 728 {
 729         struct ptlrpc_request   *req;
 730         struct ptlrpc_bulk_desc *desc;
 731         struct client_obd       *cli = &imp->imp_obd->u.cli;
 732         struct ost_body         *body;
 733         struct obd_ioobj        *ioobj;
 734         struct niobuf_remote    *niobuf;
 735         unsigned long            flags;
 736         int                      niocount;
 737         int                      size[3];
 738         int                      i;
 739         int                      requested_nob;
 740         int                      opc;
 741         int                      rc;
 742
 743         opc = ((cmd & OBD_BRW_WRITE) != 0) ? OST_WRITE : OST_READ;
 744
 745         for (niocount = i = 1; i < page_count; i++)
 746                 if (!can_merge_pages(&pga[i - 1], &pga[i]))
 747                         niocount++;
 748
 749         size[0] = sizeof(*body);
 750         size[1] = sizeof(*ioobj);
 751         size[2] = niocount * sizeof(*niobuf);
 752
 753         req = ptlrpc_prep_req(imp, opc, 3, size, NULL);
 754         if (req == NULL)
 755                 return (-ENOMEM);
 756
 757         if (opc == OST_WRITE)
 758                 desc = ptlrpc_prep_bulk_imp (req, page_count,
 759                                              BULK_GET_SOURCE, OST_BULK_PORTAL);
 760         else
 761                 desc = ptlrpc_prep_bulk_imp (req, page_count,
 762                                              BULK_PUT_SINK, OST_BULK_PORTAL);
 763         if (desc == NULL)
 764                 GOTO(out, rc = -ENOMEM);
 765         /* NB request now owns desc and will free it when it gets freed */
 766
 767         body = lustre_msg_buf(req->rq_reqmsg, 0, sizeof(*body));
 768         ioobj = lustre_msg_buf(req->rq_reqmsg, 1, sizeof(*ioobj));
 769         niobuf = lustre_msg_buf(req->rq_reqmsg, 2, niocount * sizeof(*niobuf));
 770
 771         memcpy(&body->oa, oa, sizeof(*oa));
 772
 773         obdo_to_ioobj(oa, ioobj);
 774         ioobj->ioo_bufcnt = niocount;
 775
 776         LASSERT (page_count > 0);
 777         for (requested_nob = i = 0; i < page_count; i++, niobuf++) {
 778                 struct brw_page *pg = &pga[i];
 779                 struct brw_page *pg_prev = pg - 1;
 780
 781                 LASSERT(pg->count > 0);
 782                 LASSERT((pg->off & ~PAGE_MASK) + pg->count <= PAGE_SIZE);
 783                 LASSERTF(i == 0 || pg->off > pg_prev->off,
 784                          "i %d p_c %u pg %p [pri %lu ind %lu] off "LPU64
 785                          " prev_pg %p [pri %lu ind %lu] off "LPU64"\n",
 786                          i, page_count,
 787                          pg->pg, pg->pg->private, pg->pg->index, pg->off,
 788                          pg_prev->pg, pg_prev->pg->private, pg_prev->pg->index,
 789                                  pg_prev->off);
 790
 791                 ptlrpc_prep_bulk_page(desc, pg->pg, pg->off & ~PAGE_MASK,
 792                                       pg->count);
 793                 requested_nob += pg->count;
 794
 795                 if (i > 0 && can_merge_pages(pg_prev, pg)) {
 796                         niobuf--;
 797                         niobuf->len += pg->count;
 798                 } else {
 799                         niobuf->offset = pg->off;
 800                         niobuf->len    = pg->count;
 801                         niobuf->flags  = pg->flag;
 802                 }
 803         }
 804
 805         LASSERT((void *)(niobuf - niocount) ==
 806                 lustre_msg_buf(req->rq_reqmsg, 2, niocount * sizeof(*niobuf)));
 807         osc_announce_cached(cli, &body->oa, opc == OST_WRITE ? requested_nob:0);
 808         spin_lock_irqsave(&req->rq_lock, flags);
 809         req->rq_no_resend = 1;
 810         spin_unlock_irqrestore(&req->rq_lock, flags);
 811
 812         /* size[0] still sizeof (*body) */
 813         if (opc == OST_WRITE) {
 814 #if CHECKSUM_BULK
 815                 body->oa.o_valid |= OBD_MD_FLCKSUM;
 816                 body->oa.o_cksum = cksum_pages(requested_nob, page_count, pga);
 817 #endif
 818                 /* 1 RC per niobuf */
 819                 size[1] = sizeof(__u32) * niocount;
 820                 req->rq_replen = lustre_msg_size(2, size);
 821         } else {
 822                 /* 1 RC for the whole I/O */
 823                 req->rq_replen = lustre_msg_size(1, size);
 824         }
 825
 826         *niocountp = niocount;
 827         *requested_nobp = requested_nob;
 828         *reqp = req;
 829         return (0);
 830
 831  out:
 832         ptlrpc_req_finished (req);
 833         return (rc);
 834 }
 835
 836 static int osc_brw_fini_request(struct ptlrpc_request *req, struct obdo *oa,
 837                                 int requested_nob, int niocount,
 838                                 obd_count page_count, struct brw_page *pga,
 839                                 int rc)
 840 {
 841         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
 842         struct ost_body *body;
 843         ENTRY;
 844
 845         if (rc < 0)
 846                 RETURN(rc);
 847
 848         body = lustre_swab_repbuf(req, 0, sizeof(*body), lustre_swab_ost_body);
 849         if (body == NULL) {
 850                 CERROR ("Can't unpack body\n");
 851                 RETURN(-EPROTO);
 852         }
 853
 854         osc_update_grant(cli, body);
 855
 856         if (req->rq_reqmsg->opc == OST_WRITE) {
 857                 if (rc > 0) {
 858                         CERROR ("Unexpected +ve rc %d\n", rc);
 859                         RETURN(-EPROTO);
 860                 }
 861                 LASSERT (req->rq_bulk->bd_nob == requested_nob);
 862
 863                 RETURN(check_write_rcs(req, requested_nob, niocount,
 864                                        page_count, pga));
 865         }
 866
 867         if (rc > requested_nob) {
 868                 CERROR("Unexpected rc %d (%d requested)\n", rc, requested_nob);
 869                 RETURN(-EPROTO);
 870         }
 871
 872         if (rc != req->rq_bulk->bd_nob_transferred) {
 873                 CERROR ("Unexpected rc %d (%d transferred)\n",
 874                         rc, req->rq_bulk->bd_nob_transferred);
 875                 return (-EPROTO);
 876         }
 877
 878         if (rc < requested_nob)
 879                 handle_short_read(rc, page_count, pga);
 880
 881         memcpy(oa, &body->oa, sizeof(*oa));
 882
 883 #if CHECKSUM_BULK
 884         if (oa->o_valid & OBD_MD_FLCKSUM) {
 885                 const struct ptlrpc_peer *peer =
 886                         &req->rq_import->imp_connection->c_peer;
 887                 static int cksum_counter;
 888                 obd_count server_cksum = oa->o_cksum;
 889                 obd_count cksum = cksum_pages(rc, page_count, pga);
 890                 char str[PTL_NALFMT_SIZE];
 891
 892                 portals_nid2str(peer->peer_ni->pni_number, peer->peer_nid, str);
 893
 894                 cksum_counter++;
 895                 if (server_cksum != cksum) {
 896                         CERROR("Bad checksum: server %x, client %x, server NID "
 897                                LPX64" (%s)\n", server_cksum, cksum,
 898                                peer->peer_nid, str);
 899                         cksum_counter = 0;
 900                         oa->o_cksum = cksum;
 901                 } else if ((cksum_counter & (-cksum_counter)) == cksum_counter){
 902                         CWARN("Checksum %u from "LPX64" (%s) OK: %x\n",
 903                               cksum_counter, peer->peer_nid, str, cksum);
 904                 }
 905         } else {
 906                 static int cksum_missed;
 907
 908                 cksum_missed++;
 909                 if ((cksum_missed & (-cksum_missed)) == cksum_missed)
 910                         CERROR("Request checksum %u from "LPX64", no reply\n",
 911                                cksum_missed,
 912                                req->rq_import->imp_connection->c_peer.peer_nid);
 913         }
 914 #endif
 915         RETURN(0);
 916 }
 917
 918 static int osc_brw_internal(int cmd, struct obd_export *exp,struct obdo *oa,
 919                             struct lov_stripe_md *lsm,
 920                             obd_count page_count, struct brw_page *pga)
 921 {
 922         int                    requested_nob;
 923         int                    niocount;
 924         struct ptlrpc_request *request;
 925         int                    rc;
 926         ENTRY;
 927
 928 restart_bulk:
 929         rc = osc_brw_prep_request(cmd, class_exp2cliimp(exp), oa, lsm,
 930                                   page_count, pga, &requested_nob, &niocount,
 931                                   &request);
 932         /* NB ^ sets rq_no_resend */
 933
 934         if (rc != 0)
 935                 return (rc);
 936
 937         rc = ptlrpc_queue_wait(request);
 938
 939         if (rc == -ETIMEDOUT && request->rq_resend) {
 940                 DEBUG_REQ(D_HA, request,  "BULK TIMEOUT");
 941                 ptlrpc_req_finished(request);
 942                 goto restart_bulk;
 943         }
 944
 945         rc = osc_brw_fini_request(request, oa, requested_nob, niocount,
 946                                   page_count, pga, rc);
 947
 948         ptlrpc_req_finished(request);
 949         RETURN (rc);
 950 }
 951
 952 static int brw_interpret(struct ptlrpc_request *request,
 953                          struct osc_brw_async_args *aa, int rc)
 954 {
 955         struct obdo *oa      = aa->aa_oa;
 956         int requested_nob    = aa->aa_requested_nob;
 957         int niocount         = aa->aa_nio_count;
 958         obd_count page_count = aa->aa_page_count;
 959         struct brw_page *pga = aa->aa_pga;
 960         ENTRY;
 961
 962         /* XXX bug 937 here */
 963         if (rc == -ETIMEDOUT && request->rq_resend) {
 964                 DEBUG_REQ(D_HA, request,  "BULK TIMEOUT");
 965                 LBUG(); /* re-send.  later. */
 966                 //goto restart_bulk;
 967         }
 968
 969         rc = osc_brw_fini_request(request, oa, requested_nob, niocount,
 970                                   page_count, pga, rc);
 971         RETURN (rc);
 972 }
 973
 974 static int async_internal(int cmd, struct obd_export *exp, struct obdo *oa,
 975                           struct lov_stripe_md *lsm, obd_count page_count,
 976                           struct brw_page *pga, struct ptlrpc_request_set *set)
 977 {
 978         struct ptlrpc_request     *request;
 979         int                        requested_nob;
 980         int                        nio_count;
 981         struct osc_brw_async_args *aa;
 982         int                        rc;
 983         ENTRY;
 984
 985         rc = osc_brw_prep_request(cmd, class_exp2cliimp(exp), oa, lsm,
 986                                   page_count, pga, &requested_nob, &nio_count,
 987                                   &request);
 988         /* NB ^ sets rq_no_resend */
 989
 990         if (rc == 0) {
 991                 LASSERT(sizeof(*aa) <= sizeof(request->rq_async_args));
 992                 aa = (struct osc_brw_async_args *)&request->rq_async_args;
 993                 aa->aa_oa = oa;
 994                 aa->aa_requested_nob = requested_nob;
 995                 aa->aa_nio_count = nio_count;
 996                 aa->aa_page_count = page_count;
 997                 aa->aa_pga = pga;
 998
 999                 request->rq_interpret_reply = brw_interpret;
1000                 ptlrpc_set_add_req(set, request);
1001         }
1002         RETURN (rc);
1003 }
1004
1005 #ifndef min_t
1006 #define min_t(type,x,y) \
1007         ({ type __x = (x); type __y = (y); __x < __y ? __x: __y; })
1008 #endif
1009
1010 /*
1011  * ugh, we want disk allocation on the target to happen in offset order.  we'll
1012  * follow sedgewicks advice and stick to the dead simple shellsort -- it'll do
1013  * fine for our small page arrays and doesn't require allocation.  its an
1014  * insertion sort that swaps elements that are strides apart, shrinking the
1015  * stride down until its '1' and the array is sorted.
1016  */
1017 static void sort_brw_pages(struct brw_page *array, int num)
1018 {
1019         int stride, i, j;
1020         struct brw_page tmp;
1021
1022         if (num == 1)
1023                 return;
1024         for (stride = 1; stride < num ; stride = (stride * 3) + 1)
1025                 ;
1026
1027         do {
1028                 stride /= 3;
1029                 for (i = stride ; i < num ; i++) {
1030                         tmp = array[i];
1031                         j = i;
1032                         while (j >= stride && array[j - stride].off > tmp.off) {
1033                                 array[j] = array[j - stride];
1034                                 j -= stride;
1035                         }
1036                         array[j] = tmp;
1037                 }
1038         } while (stride > 1);
1039 }
1040
1041 /* make sure we the regions we're passing to elan don't violate its '4
1042  * fragments' constraint.  portal headers are a fragment, all full
1043  * PAGE_SIZE long pages count as 1 fragment, and each partial page
1044  * counts as a fragment.  I think.  see bug 934. */
1045 static obd_count check_elan_limit(struct brw_page *pg, obd_count pages)
1046 {
1047         int frags_left = 3;
1048         int saw_whole_frag = 0;
1049         int i;
1050
1051         for (i = 0 ; frags_left && i < pages ; pg++, i++) {
1052                 if (pg->count == PAGE_SIZE) {
1053                         if (!saw_whole_frag) {
1054                                 saw_whole_frag = 1;
1055                                 frags_left--;
1056                         }
1057                 } else {
1058                         frags_left--;
1059                 }
1060         }
1061         return i;
1062 }
1063
1064 static int osc_brw(int cmd, struct obd_export *exp, struct obdo *oa,
1065                    struct lov_stripe_md *md, obd_count page_count,
1066                    struct brw_page *pga, struct obd_trans_info *oti)
1067 {
1068         ENTRY;
1069
1070         if (cmd == OBD_BRW_CHECK) {
1071                 /* The caller just wants to know if there's a chance that this
1072                  * I/O can succeed */
1073                 struct obd_import *imp = class_exp2cliimp(exp);
1074
1075                 if (imp == NULL || imp->imp_invalid)
1076                         RETURN(-EIO);
1077                 RETURN(0);
1078         }
1079
1080         while (page_count) {
1081                 obd_count pages_per_brw;
1082                 int rc;
1083
1084                 if (page_count > PTLRPC_MAX_BRW_PAGES)
1085                         pages_per_brw = PTLRPC_MAX_BRW_PAGES;
1086                 else
1087                         pages_per_brw = page_count;
1088
1089                 sort_brw_pages(pga, pages_per_brw);
1090                 pages_per_brw = check_elan_limit(pga, pages_per_brw);
1091
1092                 rc = osc_brw_internal(cmd, exp, oa, md, pages_per_brw, pga);
1093
1094                 if (rc != 0)
1095                         RETURN(rc);
1096
1097                 page_count -= pages_per_brw;
1098                 pga += pages_per_brw;
1099         }
1100         RETURN(0);
1101 }
1102
1103 static int osc_brw_async(int cmd, struct obd_export *exp, struct obdo *oa,
1104                          struct lov_stripe_md *md, obd_count page_count,
1105                          struct brw_page *pga, struct ptlrpc_request_set *set,
1106                          struct obd_trans_info *oti)
1107 {
1108         ENTRY;
1109
1110         if (cmd == OBD_BRW_CHECK) {
1111                 /* The caller just wants to know if there's a chance that this
1112                  * I/O can succeed */
1113                 struct obd_import *imp = class_exp2cliimp(exp);
1114
1115                 if (imp == NULL || imp->imp_invalid)
1116                         RETURN(-EIO);
1117                 RETURN(0);
1118         }
1119
1120         while (page_count) {
1121                 obd_count pages_per_brw;
1122                 int rc;
1123
1124                 if (page_count > PTLRPC_MAX_BRW_PAGES)
1125                         pages_per_brw = PTLRPC_MAX_BRW_PAGES;
1126                 else
1127                         pages_per_brw = page_count;
1128
1129                 sort_brw_pages(pga, pages_per_brw);
1130                 pages_per_brw = check_elan_limit(pga, pages_per_brw);
1131
1132                 rc = async_internal(cmd, exp, oa, md, pages_per_brw, pga, set);
1133
1134                 if (rc != 0)
1135                         RETURN(rc);
1136
1137                 page_count -= pages_per_brw;
1138                 pga += pages_per_brw;
1139         }
1140         RETURN(0);
1141 }
1142
1143 static void osc_check_rpcs(struct client_obd *cli);
1144 static void osc_exit_cache(struct client_obd *cli, struct osc_async_page *oap,
1145                            int sent);
1146 static void loi_list_maint(struct client_obd *cli, struct lov_oinfo *loi);
1147 static void lop_update_pending(struct client_obd *cli,
1148                                struct loi_oap_pages *lop, int cmd, int delta);
1149
1150 /* this is called when a sync waiter receives an interruption.  Its job is to
1151  * get the caller woken as soon as possible.  If its page hasn't been put in an
1152  * rpc yet it can dequeue immediately.  Otherwise it has to mark the rpc as
1153  * desiring interruption which will forcefully complete the rpc once the rpc
1154  * has timed out */
1155 static void osc_occ_interrupted(struct oig_callback_context *occ)
1156 {
1157         struct osc_async_page *oap;
1158         struct loi_oap_pages *lop;
1159         struct lov_oinfo *loi;
1160         ENTRY;
1161
1162         /* XXX member_of() */
1163         oap = list_entry(occ, struct osc_async_page, oap_occ);
1164
1165         spin_lock(&oap->oap_cli->cl_loi_list_lock);
1166
1167         oap->oap_interrupted = 1;
1168
1169         /* ok, it's been put in an rpc. */
1170         if (oap->oap_request != NULL) {
1171                 ptlrpc_mark_interrupted(oap->oap_request);
1172                 ptlrpcd_wake(oap->oap_request);
1173                 GOTO(unlock, 0);
1174         }
1175
1176         /* we don't get interruption callbacks until osc_trigger_sync_io()
1177          * has been called and put the sync oaps in the pending/urgent lists.*/
1178         if (!list_empty(&oap->oap_pending_item)) {
1179                 list_del_init(&oap->oap_pending_item);
1180                 if (oap->oap_async_flags & ASYNC_URGENT)
1181                         list_del_init(&oap->oap_urgent_item);
1182
1183                 loi = oap->oap_loi;
1184                 lop = (oap->oap_cmd == OBD_BRW_WRITE) ?
1185                         &loi->loi_write_lop : &loi->loi_read_lop;
1186                 lop_update_pending(oap->oap_cli, lop, oap->oap_cmd, -1);
1187                 loi_list_maint(oap->oap_cli, oap->oap_loi);
1188
1189                 oig_complete_one(oap->oap_oig, &oap->oap_occ, 0);
1190                 oap->oap_oig = NULL;
1191         }
1192
1193 unlock:
1194         spin_unlock(&oap->oap_cli->cl_loi_list_lock);
1195 }
1196
1197 /* this must be called holding the loi list lock to give coverage to exit_cache,
1198  * async_flag maintenance, and oap_request */
1199 static void osc_complete_oap(struct client_obd *cli,
1200                              struct osc_async_page *oap, int sent, int rc)
1201 {
1202         osc_exit_cache(cli, oap, sent);
1203         oap->oap_async_flags = 0;
1204         oap->oap_interrupted = 0;
1205
1206         if (oap->oap_request != NULL) {
1207                 ptlrpc_req_finished(oap->oap_request);
1208                 oap->oap_request = NULL;
1209         }
1210
1211         if (oap->oap_oig) {
1212                 oig_complete_one(oap->oap_oig, &oap->oap_occ, rc);
1213                 oap->oap_oig = NULL;
1214                 EXIT;
1215                 return;
1216         }
1217
1218         oap->oap_caller_ops->ap_completion(oap->oap_caller_data, oap->oap_cmd,
1219                                            rc);
1220 }
1221
1222 static int brw_interpret_oap(struct ptlrpc_request *request,
1223                              struct osc_brw_async_args *aa, int rc)
1224 {
1225         struct osc_async_page *oap;
1226         struct client_obd *cli;
1227         struct list_head *pos, *n;
1228         ENTRY;
1229
1230
1231         rc = osc_brw_fini_request(request, aa->aa_oa, aa->aa_requested_nob,
1232                                   aa->aa_nio_count, aa->aa_page_count,
1233                                   aa->aa_pga, rc);
1234
1235         CDEBUG(D_INODE, "request %p aa %p rc %d\n", request, aa, rc);
1236
1237         cli = aa->aa_cli;
1238         /* in failout recovery we ignore writeback failure and want
1239          * to just tell llite to unlock the page and continue */
1240         if (request->rq_reqmsg->opc == OST_WRITE &&
1241             (cli->cl_import == NULL || cli->cl_import->imp_invalid)) {
1242                 CDEBUG(D_INODE, "flipping to rc 0 imp %p inv %d\n",
1243                        cli->cl_import,
1244                        cli->cl_import ? cli->cl_import->imp_invalid : -1);
1245                 rc = 0;
1246         }
1247
1248         spin_lock(&cli->cl_loi_list_lock);
1249
1250         /* We need to decrement before osc_complete_oap->osc_wake_cache_waiters
1251          * is called so we know whether to go to sync BRWs or wait for more
1252          * RPCs to complete */
1253         cli->cl_brw_in_flight--;
1254
1255         /* the caller may re-use the oap after the completion call so
1256          * we need to clean it up a little */
1257         list_for_each_safe(pos, n, &aa->aa_oaps) {
1258                 oap = list_entry(pos, struct osc_async_page, oap_rpc_item);
1259
1260                 //CDEBUG(D_INODE, "page %p index %lu oap %p\n",
1261                        //oap->oap_page, oap->oap_page->index, oap);
1262
1263                 list_del_init(&oap->oap_rpc_item);
1264                 osc_complete_oap(cli, oap, 1, rc);
1265         }
1266
1267         osc_wake_cache_waiters(cli);
1268         osc_check_rpcs(cli);
1269
1270         spin_unlock(&cli->cl_loi_list_lock);
1271
1272         obdo_free(aa->aa_oa);
1273         OBD_FREE(aa->aa_pga, aa->aa_page_count * sizeof(struct brw_page));
1274
1275         RETURN(0);
1276 }
1277
1278 static struct ptlrpc_request *osc_build_req(struct client_obd *cli,
1279                                             struct list_head *rpc_list,
1280                                             int page_count, int cmd)
1281 {
1282         struct ptlrpc_request *req;
1283         struct brw_page *pga = NULL;
1284         int requested_nob, nio_count;
1285         struct osc_brw_async_args *aa;
1286         struct obdo *oa = NULL;
1287         struct obd_async_page_ops *ops = NULL;
1288         void *caller_data = NULL;
1289         struct list_head *pos;
1290         int i, rc;
1291
1292         LASSERT(!list_empty(rpc_list));
1293
1294         OBD_ALLOC(pga, sizeof(*pga) * page_count);
1295         if (pga == NULL)
1296                 RETURN(ERR_PTR(-ENOMEM));
1297
1298         oa = obdo_alloc();
1299         if (oa == NULL)
1300                 GOTO(out, req = ERR_PTR(-ENOMEM));
1301
1302         i = 0;
1303         list_for_each(pos, rpc_list) {
1304                 struct osc_async_page *oap;
1305
1306                 oap = list_entry(pos, struct osc_async_page, oap_rpc_item);
1307                 if (ops == NULL) {
1308                         ops = oap->oap_caller_ops;
1309                         caller_data = oap->oap_caller_data;
1310                 }
1311                 pga[i].off = oap->oap_obj_off + oap->oap_page_off;
1312                 pga[i].pg = oap->oap_page;
1313                 pga[i].count = oap->oap_count;
1314                 pga[i].flag = oap->oap_brw_flags;
1315                 CDEBUG(0, "put page %p index %lu oap %p flg %x to pga\n",
1316                        pga[i].pg, oap->oap_page->index, oap, pga[i].flag);
1317                 i++;
1318         }
1319
1320         /* always get the data for the obdo for the rpc */
1321         LASSERT(ops != NULL);
1322         ops->ap_fill_obdo(caller_data, cmd, oa);
1323
1324         sort_brw_pages(pga, page_count);
1325         rc = osc_brw_prep_request(cmd, cli->cl_import, oa, NULL, page_count,
1326                                   pga, &requested_nob, &nio_count, &req);
1327         if (rc != 0) {
1328                 CERROR("prep_req failed: %d\n", rc);
1329                 GOTO(out, req = ERR_PTR(rc));
1330         }
1331
1332         LASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
1333         aa = (struct osc_brw_async_args *)&req->rq_async_args;
1334         aa->aa_oa = oa;
1335         aa->aa_requested_nob = requested_nob;
1336         aa->aa_nio_count = nio_count;
1337         aa->aa_page_count = page_count;
1338         aa->aa_pga = pga;
1339         aa->aa_cli = cli;
1340
1341 out:
1342         if (IS_ERR(req)) {
1343                 if (oa)
1344                         obdo_free(oa);
1345                 if (pga)
1346                         OBD_FREE(pga, sizeof(*pga) * page_count);
1347         }
1348         RETURN(req);
1349 }
1350
1351 static void lop_update_pending(struct client_obd *cli,
1352                                struct loi_oap_pages *lop, int cmd, int delta)
1353 {
1354         lop->lop_num_pending += delta;
1355         if (cmd == OBD_BRW_WRITE)
1356                 cli->cl_pending_w_pages += delta;
1357         else
1358                 cli->cl_pending_r_pages += delta;
1359 }
1360
1361 /* the loi lock is held across this function but it's allowed to release
1362  * and reacquire it during its work */
1363 static int osc_send_oap_rpc(struct client_obd *cli, struct lov_oinfo *loi,
1364                             int cmd, struct loi_oap_pages *lop)
1365 {
1366         struct ptlrpc_request *request;
1367         obd_count page_count = 0;
1368         struct list_head *tmp, *pos;
1369         struct osc_async_page *oap = NULL;
1370         struct osc_brw_async_args *aa;
1371         struct obd_async_page_ops *ops;
1372         LIST_HEAD(rpc_list);
1373         ENTRY;
1374
1375         /* first we find the pages we're allowed to work with */
1376         list_for_each_safe(pos, tmp, &lop->lop_pending) {
1377                 oap = list_entry(pos, struct osc_async_page, oap_pending_item);
1378                 ops = oap->oap_caller_ops;
1379
1380                 LASSERT(oap->oap_magic == OAP_MAGIC);
1381
1382                 /* in llite being 'ready' equates to the page being locked
1383                  * until completion unlocks it.  commit_write submits a page
1384                  * as not ready because its unlock will happen unconditionally
1385                  * as the call returns.  if we race with commit_write giving
1386                  * us that page we dont' want to create a hole in the page
1387                  * stream, so we stop and leave the rpc to be fired by
1388                  * another dirtier or kupdated interval (the not ready page
1389                  * will still be on the dirty list).  we could call in
1390                  * at the end of ll_file_write to process the queue again. */
1391                 if (!(oap->oap_async_flags & ASYNC_READY)) {
1392                         int rc = ops->ap_make_ready(oap->oap_caller_data, cmd);
1393                         if (rc < 0)
1394                                 CDEBUG(D_INODE, "oap %p page %p returned %d "
1395                                                 "instead of ready\n", oap,
1396                                                 oap->oap_page, rc);
1397                         switch (rc) {
1398                         case -EAGAIN:
1399                                 /* llite is telling us that the page is still
1400                                  * in commit_write and that we should try
1401                                  * and put it in an rpc again later.  we
1402                                  * break out of the loop so we don't create
1403                                  * a hole in the sequence of pages in the rpc
1404                                  * stream.*/
1405                                 pos = NULL;
1406                                 break;
1407                         case -EINTR:
1408                                 /* the io isn't needed.. tell the checks
1409                                  * below to complete the rpc with EINTR */
1410                                 oap->oap_async_flags |= ASYNC_COUNT_STABLE;
1411                                 oap->oap_count = -EINTR;
1412                                 break;
1413                         case 0:
1414                                 oap->oap_async_flags |= ASYNC_READY;
1415                                 break;
1416                         default:
1417                                 LASSERTF(0, "oap %p page %p returned %d "
1418                                             "from make_ready\n", oap,
1419                                             oap->oap_page, rc);
1420                                 break;
1421                         }
1422                 }
1423                 if (pos == NULL)
1424                         break;
1425
1426                 /* take the page out of our book-keeping */
1427                 list_del_init(&oap->oap_pending_item);
1428                 lop_update_pending(cli, lop, cmd, -1);
1429                 if (!list_empty(&oap->oap_urgent_item))
1430                         list_del_init(&oap->oap_urgent_item);
1431
1432                 /* ask the caller for the size of the io as the rpc leaves. */
1433                 if (!(oap->oap_async_flags & ASYNC_COUNT_STABLE))
1434                         oap->oap_count =
1435                                 ops->ap_refresh_count(oap->oap_caller_data,cmd);
1436                 if (oap->oap_count <= 0) {
1437                         CDEBUG(D_CACHE, "oap %p count %d, completing\n", oap,
1438                                oap->oap_count);
1439                         osc_complete_oap(cli, oap, 0, oap->oap_count);
1440                         continue;
1441                 }
1442
1443                 /* now put the page back in our accounting */
1444                 list_add_tail(&oap->oap_rpc_item, &rpc_list);
1445                 if (++page_count >= cli->cl_max_pages_per_rpc)
1446                         break;
1447         }
1448
1449         osc_wake_cache_waiters(cli);
1450
1451         if (page_count == 0)
1452                 RETURN(0);
1453
1454         loi_list_maint(cli, loi);
1455         spin_unlock(&cli->cl_loi_list_lock);
1456
1457         request = osc_build_req(cli, &rpc_list, page_count, cmd);
1458         if (IS_ERR(request)) {
1459                 /* this should happen rarely and is pretty bad, it makes the
1460                  * pending list not follow the dirty order */
1461                 spin_lock(&cli->cl_loi_list_lock);
1462                 list_for_each_safe(pos, tmp, &rpc_list) {
1463                         oap = list_entry(pos, struct osc_async_page,
1464                                          oap_rpc_item);
1465                         list_del_init(&oap->oap_rpc_item);
1466
1467                         /* queued sync pages can be torn down while the pages
1468                          * were between the pending list and the rpc */
1469                         if (oap->oap_interrupted) {
1470                                 CDEBUG(D_INODE, "oap %p interrupted\n", oap);
1471                                 osc_complete_oap(cli, oap, 0, oap->oap_count);
1472                                 continue;
1473                         }
1474
1475                         /* put the page back in the loi/lop lists */
1476                         list_add_tail(&oap->oap_pending_item,
1477                                       &lop->lop_pending);
1478                         lop_update_pending(cli, lop, cmd, 1);
1479                         if (oap->oap_async_flags & ASYNC_URGENT)
1480                                 list_add(&oap->oap_urgent_item,
1481                                          &lop->lop_urgent);
1482                 }
1483                 loi_list_maint(cli, loi);
1484                 RETURN(PTR_ERR(request));
1485         }
1486
1487         LASSERT(sizeof(*aa) <= sizeof(request->rq_async_args));
1488         aa = (struct osc_brw_async_args *)&request->rq_async_args;
1489         INIT_LIST_HEAD(&aa->aa_oaps);
1490         list_splice(&rpc_list, &aa->aa_oaps);
1491         INIT_LIST_HEAD(&rpc_list);
1492
1493 #ifdef __KERNEL__
1494         if (cmd == OBD_BRW_READ) {
1495                 lprocfs_oh_tally_log2(&cli->cl_read_page_hist, page_count);
1496                 lprocfs_oh_tally(&cli->cl_read_rpc_hist, cli->cl_brw_in_flight);
1497         } else {
1498                 lprocfs_oh_tally_log2(&cli->cl_write_page_hist, page_count);
1499                 lprocfs_oh_tally(&cli->cl_write_rpc_hist,
1500                                  cli->cl_brw_in_flight);
1501         }
1502 #endif
1503
1504         spin_lock(&cli->cl_loi_list_lock);
1505
1506         cli->cl_brw_in_flight++;
1507         /* queued sync pages can be torn down while the pages
1508          * were between the pending list and the rpc */
1509         list_for_each(pos, &aa->aa_oaps) {
1510                 oap = list_entry(pos, struct osc_async_page, oap_rpc_item);
1511                 if (oap->oap_interrupted) {
1512                         CDEBUG(D_INODE, "oap %p in req %p interrupted\n",
1513                                oap, request);
1514                         ptlrpc_mark_interrupted(request);
1515                         break;
1516                 }
1517         }
1518
1519         CDEBUG(D_INODE, "req %p: %d pages, aa %p.  now %d in flight\n", request,
1520                page_count, aa, cli->cl_brw_in_flight);
1521
1522         oap->oap_request = ptlrpc_request_addref(request);
1523         request->rq_interpret_reply = brw_interpret_oap;
1524         ptlrpcd_add_req(request);
1525         RETURN(1);
1526 }
1527
1528 static int lop_makes_rpc(struct client_obd *cli, struct loi_oap_pages *lop,
1529                          int cmd)
1530 {
1531         int optimal;
1532         ENTRY;
1533
1534         if (lop->lop_num_pending == 0)
1535                 RETURN(0);
1536
1537         /* if we have an invalid import we want to drain the queued pages
1538          * by forcing them through rpcs that immediately fail and complete
1539          * the pages.  recovery relies on this to empty the queued pages
1540          * before canceling the locks and evicting down the llite pages */
1541         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
1542                 RETURN(1);
1543
1544         /* stream rpcs in queue order as long as as there is an urgent page
1545          * queued.  this is our cheap solution for good batching in the case
1546          * where writepage marks some random page in the middle of the file as
1547          * urgent because of, say, memory pressure */
1548         if (!list_empty(&lop->lop_urgent))
1549                 RETURN(1);
1550
1551         /* fire off rpcs when we have 'optimal' rpcs as tuned for the wire. */
1552         optimal = cli->cl_max_pages_per_rpc;
1553         if (cmd == OBD_BRW_WRITE) {
1554                 /* trigger a write rpc stream as long as there are dirtiers
1555                  * waiting for space.  as they're waiting, they're not going to
1556                  * create more pages to coallesce with what's waiting.. */
1557                 if (!list_empty(&cli->cl_cache_waiters))
1558                         RETURN(1);
1559
1560                 /* *2 to avoid triggering rpcs that would want to include pages
1561                  * that are being queued but which can't be made ready until
1562                  * the queuer finishes with the page. this is a wart for
1563                  * llite::commit_write() */
1564                 optimal += 16;
1565         }
1566         if (lop->lop_num_pending >= optimal)
1567                 RETURN(1);
1568
1569         RETURN(0);
1570 }
1571
1572 static void on_list(struct list_head *item, struct list_head *list,
1573                     int should_be_on)
1574 {
1575         if (list_empty(item) && should_be_on)
1576                 list_add_tail(item, list);
1577         else if (!list_empty(item) && !should_be_on)
1578                 list_del_init(item);
1579 }
1580
1581 /* maintain the loi's cli list membership invariants so that osc_send_oap_rpc
1582  * can find pages to build into rpcs quickly */
1583 static void loi_list_maint(struct client_obd *cli, struct lov_oinfo *loi)
1584 {
1585         on_list(&loi->loi_cli_item, &cli->cl_loi_ready_list,
1586                 lop_makes_rpc(cli, &loi->loi_write_lop, OBD_BRW_WRITE) ||
1587                 lop_makes_rpc(cli, &loi->loi_read_lop, OBD_BRW_READ));
1588
1589         on_list(&loi->loi_write_item, &cli->cl_loi_write_list,
1590                 loi->loi_write_lop.lop_num_pending);
1591
1592         on_list(&loi->loi_read_item, &cli->cl_loi_read_list,
1593                 loi->loi_read_lop.lop_num_pending);
1594 }
1595
1596 #define LOI_DEBUG(LOI, STR, args...)                                     \
1597         CDEBUG(D_INODE, "loi ready %d wr %d:%d rd %d:%d " STR,           \
1598                !list_empty(&(LOI)->loi_cli_item),                        \
1599                (LOI)->loi_write_lop.lop_num_pending,                     \
1600                !list_empty(&(LOI)->loi_write_lop.lop_urgent),            \
1601                (LOI)->loi_read_lop.lop_num_pending,                      \
1602                !list_empty(&(LOI)->loi_read_lop.lop_urgent),             \
1603                args)                                                     \
1604
1605 struct lov_oinfo *osc_next_loi(struct client_obd *cli)
1606 {
1607         ENTRY;
1608         /* first return all objects which we already know to have
1609          * pages ready to be stuffed into rpcs */
1610         if (!list_empty(&cli->cl_loi_ready_list))
1611                 RETURN(list_entry(cli->cl_loi_ready_list.next,
1612                                   struct lov_oinfo, loi_cli_item));
1613
1614         /* then if we have cache waiters, return all objects with queued
1615          * writes.  This is especially important when many small files
1616          * have filled up the cache and not been fired into rpcs because
1617          * they don't pass the nr_pending/object threshhold */
1618         if (!list_empty(&cli->cl_cache_waiters) &&
1619             !list_empty(&cli->cl_loi_write_list))
1620                 RETURN(list_entry(cli->cl_loi_write_list.next,
1621                                   struct lov_oinfo, loi_write_item));
1622
1623         /* then return all queued objects when we have an invalid import
1624          * so that they get flushed */
1625         if (cli->cl_import == NULL || cli->cl_import->imp_invalid) {
1626                 if (!list_empty(&cli->cl_loi_write_list))
1627                         RETURN(list_entry(cli->cl_loi_write_list.next,
1628                                           struct lov_oinfo, loi_write_item));
1629                 if (!list_empty(&cli->cl_loi_read_list))
1630                         RETURN(list_entry(cli->cl_loi_read_list.next,
1631                                           struct lov_oinfo, loi_read_item));
1632         }
1633         RETURN(NULL);
1634 }
1635
1636 /* called with the loi list lock held */
1637 static void osc_check_rpcs(struct client_obd *cli)
1638 {
1639         struct lov_oinfo *loi;
1640         int rc = 0, race_counter = 0;
1641         ENTRY;
1642
1643         while ((loi = osc_next_loi(cli)) != NULL) {
1644                 LOI_DEBUG(loi, "%d in flight\n", cli->cl_brw_in_flight);
1645
1646                 if (cli->cl_brw_in_flight >= cli->cl_max_rpcs_in_flight)
1647                         break;
1648
1649                 /* attempt some read/write balancing by alternating between
1650                  * reads and writes in an object.  The makes_rpc checks here
1651                  * would be redundant if we were getting read/write work items
1652                  * instead of objects.  we don't want send_oap_rpc to drain a
1653                  * partial read pending queue when we're given this object to
1654                  * do io on writes while there are cache waiters */
1655                 if (lop_makes_rpc(cli, &loi->loi_write_lop, OBD_BRW_WRITE)) {
1656                         rc = osc_send_oap_rpc(cli, loi, OBD_BRW_WRITE,
1657                                               &loi->loi_write_lop);
1658                         if (rc < 0)
1659                                 break;
1660                         if (rc > 0)
1661                                 race_counter = 0;
1662                         else
1663                                 race_counter++;
1664                 }
1665                 if (lop_makes_rpc(cli, &loi->loi_read_lop, OBD_BRW_READ)) {
1666                         rc = osc_send_oap_rpc(cli, loi, OBD_BRW_READ,
1667                                               &loi->loi_read_lop);
1668                         if (rc < 0)
1669                                 break;
1670                         if (rc > 0)
1671                                 race_counter = 0;
1672                         else
1673                                 race_counter++;
1674                 }
1675
1676                 /* attempt some inter-object balancing by issueing rpcs
1677                  * for each object in turn */
1678                 if (!list_empty(&loi->loi_cli_item))
1679                         list_del_init(&loi->loi_cli_item);
1680                 if (!list_empty(&loi->loi_write_item))
1681                         list_del_init(&loi->loi_write_item);
1682                 if (!list_empty(&loi->loi_read_item))
1683                         list_del_init(&loi->loi_read_item);
1684
1685                 loi_list_maint(cli, loi);
1686
1687                 /* send_oap_rpc fails with 0 when make_ready tells it to
1688                  * back off.  llite's make_ready does this when it tries
1689                  * to lock a page queued for write that is already locked.
1690                  * we want to try sending rpcs from many objects, but we
1691                  * don't want to spin failing with 0.  */
1692                 if (race_counter == 10)
1693                         break;
1694         }
1695         EXIT;
1696 }
1697
1698 /* we're trying to queue a page in the osc so we're subject to the
1699  * 'cl_dirty_max' limit on the number of pages that can be queued in the osc.
1700  * If the osc's queued pages are already at that limit, then we want to sleep
1701  * until there is space in the osc's queue for us.  We also may be waiting for
1702  * write credits from the OST if there are RPCs in flight that may return some
1703  * before we fall back to sync writes.
1704  *
1705  * We need this know our allocation was granted in the presence of signals */
1706 static int ocw_granted(struct client_obd *cli, struct osc_cache_waiter *ocw)
1707 {
1708         int rc;
1709         ENTRY;
1710         spin_lock(&cli->cl_loi_list_lock);
1711         rc = list_empty(&ocw->ocw_entry) || cli->cl_brw_in_flight == 0;
1712         spin_unlock(&cli->cl_loi_list_lock);
1713         RETURN(rc);
1714 };
1715
1716 /* Caller must hold loi_list_lock - we drop/regain it if we need to wait for
1717  * grant or cache space. */
1718 static int osc_enter_cache(struct client_obd *cli, struct lov_oinfo *loi,
1719                            struct osc_async_page *oap)
1720 {
1721         struct osc_cache_waiter ocw;
1722         struct l_wait_info lwi = { 0 };
1723
1724         CDEBUG(D_CACHE, "dirty: %ld dirty_max: %ld dropped: %lu grant: %lu\n",
1725                cli->cl_dirty, cli->cl_dirty_max, cli->cl_lost_grant,
1726                cli->cl_avail_grant);
1727
1728         if (cli->cl_dirty_max < PAGE_SIZE)
1729                 return(-EDQUOT);
1730
1731         /* Hopefully normal case - cache space and write credits available */
1732         if (cli->cl_dirty + PAGE_SIZE <= cli->cl_dirty_max &&
1733             cli->cl_avail_grant >= PAGE_SIZE) {
1734                 /* account for ourselves */
1735                 osc_consume_write_grant(cli, oap);
1736                 return(0);
1737         }
1738
1739         /* Make sure that there are write rpcs in flight to wait for.  This
1740          * is a little silly as this object may not have any pending but
1741          * other objects sure might. */
1742         if (cli->cl_brw_in_flight) {
1743                 list_add_tail(&ocw.ocw_entry, &cli->cl_cache_waiters);
1744                 init_waitqueue_head(&ocw.ocw_waitq);
1745                 ocw.ocw_oap = oap;
1746                 ocw.ocw_rc = 0;
1747
1748                 loi_list_maint(cli, loi);
1749                 osc_check_rpcs(cli);
1750                 spin_unlock(&cli->cl_loi_list_lock);
1751
1752                 CDEBUG(0, "sleeping for cache space\n");
1753                 l_wait_event(ocw.ocw_waitq, ocw_granted(cli, &ocw), &lwi);
1754
1755                 spin_lock(&cli->cl_loi_list_lock);
1756                 if (!list_empty(&ocw.ocw_entry)) {
1757                         list_del(&ocw.ocw_entry);
1758                         RETURN(-EINTR);
1759                 }
1760                 RETURN(ocw.ocw_rc);
1761         }
1762
1763         RETURN(-EDQUOT);
1764 }
1765
1766 /* the companion to enter_cache, called when an oap is no longer part of the
1767  * dirty accounting.. so writeback completes or truncate happens before writing
1768  * starts.  must be called with the loi lock held. */
1769 static void osc_exit_cache(struct client_obd *cli, struct osc_async_page *oap,
1770                            int sent)
1771 {
1772         ENTRY;
1773
1774         if (!(oap->oap_brw_flags & OBD_BRW_FROM_GRANT)) {
1775                 EXIT;
1776                 return;
1777         }
1778
1779         oap->oap_brw_flags &= ~OBD_BRW_FROM_GRANT;
1780         cli->cl_dirty -= PAGE_SIZE;
1781         if (!sent) {
1782                 cli->cl_lost_grant += PAGE_SIZE;
1783                 CDEBUG(D_CACHE, "lost grant: %lu avail grant: %lu dirty: %lu\n",
1784                        cli->cl_lost_grant, cli->cl_avail_grant, cli->cl_dirty);
1785         }
1786
1787         EXIT;
1788 }
1789
1790 int osc_prep_async_page(struct obd_export *exp, struct lov_stripe_md *lsm,
1791                         struct lov_oinfo *loi, struct page *page,
1792                         obd_off offset, struct obd_async_page_ops *ops,
1793                         void *data, void **res)
1794 {
1795         struct osc_async_page *oap;
1796         ENTRY;
1797
1798         OBD_ALLOC(oap, sizeof(*oap));
1799         if (oap == NULL)
1800                 return -ENOMEM;
1801
1802         oap->oap_magic = OAP_MAGIC;
1803         oap->oap_cli = &exp->exp_obd->u.cli;
1804         oap->oap_loi = loi;
1805
1806         oap->oap_caller_ops = ops;
1807         oap->oap_caller_data = data;
1808
1809         oap->oap_page = page;
1810         oap->oap_obj_off = offset;
1811
1812         INIT_LIST_HEAD(&oap->oap_pending_item);
1813         INIT_LIST_HEAD(&oap->oap_urgent_item);
1814         INIT_LIST_HEAD(&oap->oap_rpc_item);
1815
1816         oap->oap_occ.occ_interrupted = osc_occ_interrupted;
1817
1818         CDEBUG(D_CACHE, "oap %p page %p obj off "LPU64"\n", oap, page, offset);
1819         *res = oap;
1820         RETURN(0);
1821 }
1822
1823 struct osc_async_page *oap_from_cookie(void *cookie)
1824 {
1825         struct osc_async_page *oap = cookie;
1826         if (oap->oap_magic != OAP_MAGIC)
1827                 return ERR_PTR(-EINVAL);
1828         return oap;
1829 };
1830
1831 static int osc_queue_async_io(struct obd_export *exp, struct lov_stripe_md *lsm,
1832                               struct lov_oinfo *loi, void *cookie,
1833                               int cmd, obd_off off, int count,
1834                               obd_flag brw_flags, enum async_flags async_flags)
1835 {
1836         struct client_obd *cli = &exp->exp_obd->u.cli;
1837         struct osc_async_page *oap;
1838         struct loi_oap_pages *lop;
1839         int rc;
1840         ENTRY;
1841
1842         oap = oap_from_cookie(cookie);
1843         if (IS_ERR(oap))
1844                 RETURN(PTR_ERR(oap));
1845
1846         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
1847                 RETURN(-EIO);
1848
1849         if (!list_empty(&oap->oap_pending_item) ||
1850             !list_empty(&oap->oap_urgent_item) ||
1851             !list_empty(&oap->oap_rpc_item))
1852                 RETURN(-EBUSY);
1853
1854         if (loi == NULL)
1855                 loi = &lsm->lsm_oinfo[0];
1856
1857         spin_lock(&cli->cl_loi_list_lock);
1858
1859         oap->oap_cmd = cmd;
1860         oap->oap_async_flags = async_flags;
1861         oap->oap_page_off = off;
1862         oap->oap_count = count;
1863         oap->oap_brw_flags = brw_flags;
1864
1865         if (cmd == OBD_BRW_WRITE) {
1866                 rc = osc_enter_cache(cli, loi, oap);
1867                 if (rc) {
1868                         spin_unlock(&cli->cl_loi_list_lock);
1869                         RETURN(rc);
1870                 }
1871                 lop = &loi->loi_write_lop;
1872         } else {
1873                 lop = &loi->loi_read_lop;
1874         }
1875
1876         if (oap->oap_async_flags & ASYNC_URGENT)
1877                 list_add(&oap->oap_urgent_item, &lop->lop_urgent);
1878         list_add_tail(&oap->oap_pending_item, &lop->lop_pending);
1879         lop_update_pending(cli, lop, cmd, 1);
1880
1881         loi_list_maint(cli, loi);
1882
1883         LOI_DEBUG(loi, "oap %p page %p added for cmd %d\n", oap, oap->oap_page,
1884                   cmd);
1885
1886         osc_check_rpcs(cli);
1887         spin_unlock(&cli->cl_loi_list_lock);
1888
1889         RETURN(0);
1890 }
1891
1892 /* aka (~was & now & flag), but this is more clear :) */
1893 #define SETTING(was, now, flag) (!(was & flag) && (now & flag))
1894
1895 static int osc_set_async_flags(struct obd_export *exp,
1896                                struct lov_stripe_md *lsm,
1897                                struct lov_oinfo *loi, void *cookie,
1898                                obd_flag async_flags)
1899 {
1900         struct client_obd *cli = &exp->exp_obd->u.cli;
1901         struct loi_oap_pages *lop;
1902         struct osc_async_page *oap;
1903         int rc = 0;
1904         ENTRY;
1905
1906         oap = oap_from_cookie(cookie);
1907         if (IS_ERR(oap))
1908                 RETURN(PTR_ERR(oap));
1909
1910         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
1911                 RETURN(-EIO);
1912
1913         if (loi == NULL)
1914                 loi = &lsm->lsm_oinfo[0];
1915
1916         if (oap->oap_cmd == OBD_BRW_WRITE) {
1917                 lop = &loi->loi_write_lop;
1918         } else {
1919                 lop = &loi->loi_read_lop;
1920         }
1921
1922         spin_lock(&cli->cl_loi_list_lock);
1923
1924         if (list_empty(&oap->oap_pending_item))
1925                 GOTO(out, rc = -EINVAL);
1926
1927         if ((oap->oap_async_flags & async_flags) == async_flags)
1928                 GOTO(out, rc = 0);
1929
1930         if (SETTING(oap->oap_async_flags, async_flags, ASYNC_READY))
1931                 oap->oap_async_flags |= ASYNC_READY;
1932
1933         if (SETTING(oap->oap_async_flags, async_flags, ASYNC_URGENT)) {
1934                 if (list_empty(&oap->oap_rpc_item)) {
1935                         list_add(&oap->oap_urgent_item, &lop->lop_urgent);
1936                         loi_list_maint(cli, loi);
1937                 }
1938         }
1939
1940         LOI_DEBUG(loi, "oap %p page %p has flags %x\n", oap, oap->oap_page,
1941                         oap->oap_async_flags);
1942 out:
1943         osc_check_rpcs(cli);
1944         spin_unlock(&cli->cl_loi_list_lock);
1945         RETURN(rc);
1946 }
1947
1948 static int osc_queue_group_io(struct obd_export *exp, struct lov_stripe_md *lsm,
1949                              struct lov_oinfo *loi,
1950                              struct obd_io_group *oig, void *cookie,
1951                              int cmd, obd_off off, int count,
1952                              obd_flag brw_flags,
1953                              obd_flag async_flags)
1954 {
1955         struct client_obd *cli = &exp->exp_obd->u.cli;
1956         struct osc_async_page *oap;
1957         struct loi_oap_pages *lop;
1958         ENTRY;
1959
1960         oap = oap_from_cookie(cookie);
1961         if (IS_ERR(oap))
1962                 RETURN(PTR_ERR(oap));
1963
1964         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
1965                 RETURN(-EIO);
1966
1967         if (!list_empty(&oap->oap_pending_item) ||
1968             !list_empty(&oap->oap_urgent_item) ||
1969             !list_empty(&oap->oap_rpc_item))
1970                 RETURN(-EBUSY);
1971
1972         if (loi == NULL)
1973                 loi = &lsm->lsm_oinfo[0];
1974
1975         spin_lock(&cli->cl_loi_list_lock);
1976
1977         oap->oap_cmd = cmd;
1978         oap->oap_page_off = off;
1979         oap->oap_count = count;
1980         oap->oap_brw_flags = brw_flags;
1981         oap->oap_async_flags = async_flags;
1982
1983         if (cmd == OBD_BRW_WRITE)
1984                 lop = &loi->loi_write_lop;
1985         else
1986                 lop = &loi->loi_read_lop;
1987
1988         list_add_tail(&oap->oap_pending_item, &lop->lop_pending_group);
1989         if (oap->oap_async_flags & ASYNC_GROUP_SYNC) {
1990                 oap->oap_oig = oig;
1991                 oig_add_one(oig, &oap->oap_occ);
1992         }
1993
1994         LOI_DEBUG(loi, "oap %p page %p on group pending\n", oap, oap->oap_page);
1995
1996         spin_unlock(&cli->cl_loi_list_lock);
1997
1998         RETURN(0);
1999 }
2000
2001 static void osc_group_to_pending(struct client_obd *cli, struct lov_oinfo *loi,
2002                                  struct loi_oap_pages *lop, int cmd)
2003 {
2004         struct list_head *pos, *tmp;
2005         struct osc_async_page *oap;
2006
2007         list_for_each_safe(pos, tmp, &lop->lop_pending_group) {
2008                 oap = list_entry(pos, struct osc_async_page, oap_pending_item);
2009                 list_del(&oap->oap_pending_item);
2010                 list_add_tail(&oap->oap_pending_item, &lop->lop_pending);
2011                 list_add(&oap->oap_urgent_item, &lop->lop_urgent);
2012                 lop_update_pending(cli, lop, cmd, 1);
2013         }
2014         loi_list_maint(cli, loi);
2015 }
2016
2017 static int osc_trigger_group_io(struct obd_export *exp,
2018                                 struct lov_stripe_md *lsm,
2019                                 struct lov_oinfo *loi,
2020                                 struct obd_io_group *oig)
2021 {
2022         struct client_obd *cli = &exp->exp_obd->u.cli;
2023         ENTRY;
2024
2025         if (loi == NULL)
2026                 loi = &lsm->lsm_oinfo[0];
2027
2028         spin_lock(&cli->cl_loi_list_lock);
2029
2030         osc_group_to_pending(cli, loi, &loi->loi_write_lop, OBD_BRW_WRITE);
2031         osc_group_to_pending(cli, loi, &loi->loi_read_lop, OBD_BRW_READ);
2032
2033         osc_check_rpcs(cli);
2034         spin_unlock(&cli->cl_loi_list_lock);
2035
2036         RETURN(0);
2037 }
2038
2039 static int osc_teardown_async_page(struct obd_export *exp,
2040                                    struct lov_stripe_md *lsm,
2041                                    struct lov_oinfo *loi, void *cookie)
2042 {
2043         struct client_obd *cli = &exp->exp_obd->u.cli;
2044         struct loi_oap_pages *lop;
2045         struct osc_async_page *oap;
2046         int rc = 0;
2047         ENTRY;
2048
2049         oap = oap_from_cookie(cookie);
2050         if (IS_ERR(oap))
2051                 RETURN(PTR_ERR(oap));
2052
2053         if (loi == NULL)
2054                 loi = &lsm->lsm_oinfo[0];
2055
2056         if (oap->oap_cmd == OBD_BRW_WRITE) {
2057                 lop = &loi->loi_write_lop;
2058         } else {
2059                 lop = &loi->loi_read_lop;
2060         }
2061
2062         spin_lock(&cli->cl_loi_list_lock);
2063
2064         if (!list_empty(&oap->oap_rpc_item))
2065                 GOTO(out, rc = -EBUSY);
2066
2067         osc_exit_cache(cli, oap, 0);
2068         osc_wake_cache_waiters(cli);
2069
2070         if (!list_empty(&oap->oap_urgent_item)) {
2071                 list_del_init(&oap->oap_urgent_item);
2072                 oap->oap_async_flags &= ~ASYNC_URGENT;
2073         }
2074         if (!list_empty(&oap->oap_pending_item)) {
2075                 list_del_init(&oap->oap_pending_item);
2076                 lop_update_pending(cli, lop, oap->oap_cmd, -1);
2077         }
2078         loi_list_maint(cli, loi);
2079
2080         LOI_DEBUG(loi, "oap %p page %p torn down\n", oap, oap->oap_page);
2081 out:
2082         spin_unlock(&cli->cl_loi_list_lock);
2083         if (rc == 0)
2084                 OBD_FREE(oap, sizeof(*oap));
2085         RETURN(rc);
2086 }
2087
2088 #ifdef __KERNEL__
2089 /* Note: caller will lock/unlock, and set uptodate on the pages */
2090 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
2091 static int sanosc_brw_read(struct obd_export *exp, struct obdo *oa,
2092                            struct lov_stripe_md *lsm, obd_count page_count,
2093                            struct brw_page *pga)
2094 {
2095         struct ptlrpc_request *request = NULL;
2096         struct ost_body *body;
2097         struct niobuf_remote *nioptr;
2098         struct obd_ioobj *iooptr;
2099         int rc, size[3] = {sizeof(*body)}, mapped = 0;
2100         int swab;
2101         ENTRY;
2102
2103         /* XXX does not handle 'new' brw protocol */
2104
2105         size[1] = sizeof(struct obd_ioobj);
2106         size[2] = page_count * sizeof(*nioptr);
2107
2108         request = ptlrpc_prep_req(class_exp2cliimp(exp), OST_SAN_READ, 3,
2109                                   size, NULL);
2110         if (!request)
2111                 RETURN(-ENOMEM);
2112
2113         body = lustre_msg_buf(request->rq_reqmsg, 0, sizeof(*body));
2114         iooptr = lustre_msg_buf(request->rq_reqmsg, 1, sizeof(*iooptr));
2115         nioptr = lustre_msg_buf(request->rq_reqmsg, 2,
2116                                 sizeof(*nioptr) * page_count);
2117
2118         memcpy(&body->oa, oa, sizeof(body->oa));
2119
2120         obdo_to_ioobj(oa, iooptr);
2121         iooptr->ioo_bufcnt = page_count;
2122
2123         for (mapped = 0; mapped < page_count; mapped++, nioptr++) {
2124                 LASSERT(PageLocked(pga[mapped].pg));
2125                 LASSERT(mapped == 0 || pga[mapped].off > pga[mapped - 1].off);
2126
2127                 nioptr->offset = pga[mapped].off;
2128                 nioptr->len    = pga[mapped].count;
2129                 nioptr->flags  = pga[mapped].flag;
2130         }
2131
2132         size[1] = page_count * sizeof(*nioptr);
2133         request->rq_replen = lustre_msg_size(2, size);
2134
2135         rc = ptlrpc_queue_wait(request);
2136         if (rc)
2137                 GOTO(out_req, rc);
2138
2139         body = lustre_swab_repbuf(request, 0, sizeof(*body),
2140                                   lustre_swab_ost_body);
2141         if (body == NULL) {
2142                 CERROR("Can't unpack body\n");
2143                 GOTO(out_req, rc = -EPROTO);
2144         }
2145
2146         memcpy(oa, &body->oa, sizeof(*oa));
2147
2148         swab = lustre_msg_swabbed(request->rq_repmsg);
2149         LASSERT_REPSWAB(request, 1);
2150         nioptr = lustre_msg_buf(request->rq_repmsg, 1, size[1]);
2151         if (!nioptr) {
2152                 /* nioptr missing or short */
2153                 GOTO(out_req, rc = -EPROTO);
2154         }
2155
2156         /* actual read */
2157         for (mapped = 0; mapped < page_count; mapped++, nioptr++) {
2158                 struct page *page = pga[mapped].pg;
2159                 struct buffer_head *bh;
2160                 kdev_t dev;
2161
2162                 if (swab)
2163                         lustre_swab_niobuf_remote (nioptr);
2164
2165                 /* got san device associated */
2166                 LASSERT(exp->exp_obd != NULL);
2167                 dev = exp->exp_obd->u.cli.cl_sandev;
2168
2169                 /* hole */
2170                 if (!nioptr->offset) {
2171                         CDEBUG(D_PAGE, "hole at ino %lu; index %ld\n",
2172                                         page->mapping->host->i_ino,
2173                                         page->index);
2174                         memset(page_address(page), 0, PAGE_SIZE);
2175                         continue;
2176                 }
2177
2178                 if (!page->buffers) {
2179                         create_empty_buffers(page, dev, PAGE_SIZE);
2180                         bh = page->buffers;
2181
2182                         clear_bit(BH_New, &bh->b_state);
2183                         set_bit(BH_Mapped, &bh->b_state);
2184                         bh->b_blocknr = (unsigned long)nioptr->offset;
2185
2186                         clear_bit(BH_Uptodate, &bh->b_state);
2187
2188                         ll_rw_block(READ, 1, &bh);
2189                 } else {
2190                         bh = page->buffers;
2191
2192                         /* if buffer already existed, it must be the
2193                          * one we mapped before, check it */
2194                         LASSERT(!test_bit(BH_New, &bh->b_state));
2195                         LASSERT(test_bit(BH_Mapped, &bh->b_state));
2196                         LASSERT(bh->b_blocknr == (unsigned long)nioptr->offset);
2197
2198                         /* wait it's io completion */
2199                         if (test_bit(BH_Lock, &bh->b_state))
2200                                 wait_on_buffer(bh);
2201
2202                         if (!test_bit(BH_Uptodate, &bh->b_state))
2203                                 ll_rw_block(READ, 1, &bh);
2204                 }
2205
2206
2207                 /* must do syncronous write here */
2208                 wait_on_buffer(bh);
2209                 if (!buffer_uptodate(bh)) {
2210                         /* I/O error */
2211                         rc = -EIO;
2212                         goto out_req;
2213                 }
2214         }
2215
2216 out_req:
2217         ptlrpc_req_finished(request);
2218         RETURN(rc);
2219 }
2220
2221 static int sanosc_brw_write(struct obd_export *exp, struct obdo *oa,
2222                             struct lov_stripe_md *lsm, obd_count page_count,
2223                             struct brw_page *pga)
2224 {
2225         struct ptlrpc_request *request = NULL;
2226         struct ost_body *body;
2227         struct niobuf_remote *nioptr;
2228         struct obd_ioobj *iooptr;
2229         int rc, size[3] = {sizeof(*body)}, mapped = 0;
2230         int swab;
2231         ENTRY;
2232
2233         size[1] = sizeof(struct obd_ioobj);
2234         size[2] = page_count * sizeof(*nioptr);
2235
2236         request = ptlrpc_prep_req(class_exp2cliimp(exp), OST_SAN_WRITE,
2237                                   3, size, NULL);
2238         if (!request)
2239                 RETURN(-ENOMEM);
2240
2241         body = lustre_msg_buf(request->rq_reqmsg, 0, sizeof (*body));
2242         iooptr = lustre_msg_buf(request->rq_reqmsg, 1, sizeof (*iooptr));
2243         nioptr = lustre_msg_buf(request->rq_reqmsg, 2,
2244                                 sizeof (*nioptr) * page_count);
2245
2246         memcpy(&body->oa, oa, sizeof(body->oa));
2247
2248         obdo_to_ioobj(oa, iooptr);
2249         iooptr->ioo_bufcnt = page_count;
2250
2251         /* pack request */
2252         for (mapped = 0; mapped < page_count; mapped++, nioptr++) {
2253                 LASSERT(PageLocked(pga[mapped].pg));
2254                 LASSERT(mapped == 0 || pga[mapped].off > pga[mapped - 1].off);
2255
2256                 nioptr->offset = pga[mapped].off;
2257                 nioptr->len    = pga[mapped].count;
2258                 nioptr->flags  = pga[mapped].flag;
2259         }
2260
2261         size[1] = page_count * sizeof(*nioptr);
2262         request->rq_replen = lustre_msg_size(2, size);
2263
2264         rc = ptlrpc_queue_wait(request);
2265         if (rc)
2266                 GOTO(out_req, rc);
2267
2268         swab = lustre_msg_swabbed (request->rq_repmsg);
2269         LASSERT_REPSWAB (request, 1);
2270         nioptr = lustre_msg_buf(request->rq_repmsg, 1, size[1]);
2271         if (!nioptr) {
2272                 CERROR("absent/short niobuf array\n");
2273                 GOTO(out_req, rc = -EPROTO);
2274         }
2275
2276         /* actual write */
2277         for (mapped = 0; mapped < page_count; mapped++, nioptr++) {
2278                 struct page *page = pga[mapped].pg;
2279                 struct buffer_head *bh;
2280                 kdev_t dev;
2281
2282                 if (swab)
2283                         lustre_swab_niobuf_remote (nioptr);
2284
2285                 /* got san device associated */
2286                 LASSERT(exp->exp_obd != NULL);
2287                 dev = exp->exp_obd->u.cli.cl_sandev;
2288
2289                 if (!page->buffers) {
2290                         create_empty_buffers(page, dev, PAGE_SIZE);
2291                 } else {
2292                         /* checking */
2293                         LASSERT(!test_bit(BH_New, &page->buffers->b_state));
2294                         LASSERT(test_bit(BH_Mapped, &page->buffers->b_state));
2295                         LASSERT(page->buffers->b_blocknr ==
2296                                 (unsigned long)nioptr->offset);
2297                 }
2298                 bh = page->buffers;
2299
2300                 LASSERT(bh);
2301
2302                 /* if buffer locked, wait it's io completion */
2303                 if (test_bit(BH_Lock, &bh->b_state))
2304                         wait_on_buffer(bh);
2305
2306                 clear_bit(BH_New, &bh->b_state);
2307                 set_bit(BH_Mapped, &bh->b_state);
2308
2309                 /* override the block nr */
2310                 bh->b_blocknr = (unsigned long)nioptr->offset;
2311
2312                 /* we are about to write it, so set it
2313                  * uptodate/dirty
2314                  * page lock should garentee no race condition here */
2315                 set_bit(BH_Uptodate, &bh->b_state);
2316                 set_bit(BH_Dirty, &bh->b_state);
2317
2318                 ll_rw_block(WRITE, 1, &bh);
2319
2320                 /* must do syncronous write here */
2321                 wait_on_buffer(bh);
2322                 if (!buffer_uptodate(bh) || test_bit(BH_Dirty, &bh->b_state)) {
2323                         /* I/O error */
2324                         rc = -EIO;
2325                         goto out_req;
2326                 }
2327         }
2328
2329 out_req:
2330         ptlrpc_req_finished(request);
2331         RETURN(rc);
2332 }
2333
2334 static int sanosc_brw(int cmd, struct obd_export *exp, struct obdo *oa,
2335                       struct lov_stripe_md *lsm, obd_count page_count,
2336                       struct brw_page *pga, struct obd_trans_info *oti)
2337 {
2338         ENTRY;
2339
2340         while (page_count) {
2341                 obd_count pages_per_brw;
2342                 int rc;
2343
2344                 if (page_count > PTLRPC_MAX_BRW_PAGES)
2345                         pages_per_brw = PTLRPC_MAX_BRW_PAGES;
2346                 else
2347                         pages_per_brw = page_count;
2348
2349                 if (cmd & OBD_BRW_WRITE)
2350                         rc = sanosc_brw_write(exp, oa, lsm, pages_per_brw,pga);
2351                 else
2352                         rc = sanosc_brw_read(exp, oa, lsm, pages_per_brw, pga);
2353
2354                 if (rc != 0)
2355                         RETURN(rc);
2356
2357                 page_count -= pages_per_brw;
2358                 pga += pages_per_brw;
2359         }
2360         RETURN(0);
2361 }
2362 #endif
2363 #endif
2364
2365 static void osc_set_data_with_check(struct lustre_handle *lockh, void *data)
2366 {
2367         struct ldlm_lock *lock = ldlm_handle2lock(lockh);
2368
2369         LASSERT(lock != NULL);
2370         l_lock(&lock->l_resource->lr_namespace->ns_lock);
2371 #ifdef __KERNEL__
2372         if (lock->l_ast_data && lock->l_ast_data != data) {
2373                 struct inode *new_inode = data;
2374                 struct inode *old_inode = lock->l_ast_data;
2375                 LASSERTF(old_inode->i_state & I_FREEING,
2376                          "Found existing inode %p/%lu/%u state %lu in lock: "
2377                          "setting data to %p/%lu/%u\n", old_inode,
2378                          old_inode->i_ino, old_inode->i_generation,
2379                          old_inode->i_state,
2380                          new_inode, new_inode->i_ino, new_inode->i_generation);
2381         }
2382 #endif
2383         lock->l_ast_data = data;
2384         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
2385         LDLM_LOCK_PUT(lock);
2386 }
2387
2388 static int osc_change_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
2389                              ldlm_iterator_t replace, void *data)
2390 {
2391         struct ldlm_res_id res_id = { .name = {lsm->lsm_object_id} };
2392         struct obd_device *obd = class_exp2obd(exp);
2393
2394         ldlm_change_cbdata(obd->obd_namespace, &res_id, replace, data);
2395         return 0;
2396 }
2397
2398 static int osc_enqueue(struct obd_export *exp, struct lov_stripe_md *lsm,
2399                        __u32 type, ldlm_policy_data_t *policy, __u32 mode,
2400                        int *flags, void *bl_cb, void *cp_cb, void *gl_cb,
2401                        void *data, __u32 lvb_len, void *lvb_swabber,
2402                        struct lustre_handle *lockh)
2403 {
2404         struct ldlm_res_id res_id = { .name = {lsm->lsm_object_id} };
2405         struct obd_device *obd = exp->exp_obd;
2406         struct ost_lvb lvb;
2407         int rc;
2408         ENTRY;
2409
2410         /* Filesystem lock extents are extended to page boundaries so that
2411          * dealing with the page cache is a little smoother.  */
2412         policy->l_extent.start -= policy->l_extent.start & ~PAGE_MASK;
2413         policy->l_extent.end |= ~PAGE_MASK;
2414
2415         if (lsm->lsm_oinfo->loi_kms_valid == 0)
2416                 goto no_match;
2417
2418         /* Next, search for already existing extent locks that will cover us */
2419         rc = ldlm_lock_match(obd->obd_namespace, 0, &res_id, type, policy, mode,
2420                              lockh);
2421         if (rc == 1) {
2422                 osc_set_data_with_check(lockh, data);
2423                 if (*flags & LDLM_FL_HAS_INTENT) {
2424                         /* I would like to be able to ASSERT here that rss <=
2425                          * kms, but I can't, for reasons which are explained in
2426                          * lov_enqueue() */
2427                 }
2428                 /* We already have a lock, and it's referenced */
2429                 RETURN(ELDLM_OK);
2430         }
2431
2432         /* If we're trying to read, we also search for an existing PW lock.  The
2433          * VFS and page cache already protect us locally, so lots of readers/
2434          * writers can share a single PW lock.
2435          *
2436          * There are problems with conversion deadlocks, so instead of
2437          * converting a read lock to a write lock, we'll just enqueue a new
2438          * one.
2439          *
2440          * At some point we should cancel the read lock instead of making them
2441          * send us a blocking callback, but there are problems with canceling
2442          * locks out from other users right now, too. */
2443
2444         if (mode == LCK_PR) {
2445                 rc = ldlm_lock_match(obd->obd_namespace, 0, &res_id, type,
2446                                      policy, LCK_PW, lockh);
2447                 if (rc == 1) {
2448                         /* FIXME: This is not incredibly elegant, but it might
2449                          * be more elegant than adding another parameter to
2450                          * lock_match.  I want a second opinion. */
2451                         ldlm_lock_addref(lockh, LCK_PR);
2452                         ldlm_lock_decref(lockh, LCK_PW);
2453                         osc_set_data_with_check(lockh, data);
2454                         RETURN(ELDLM_OK);
2455                 }
2456         }
2457
2458  no_match:
2459         rc = ldlm_cli_enqueue(exp, NULL, obd->obd_namespace, res_id, type,
2460                               policy, mode, flags, bl_cb, cp_cb, gl_cb, data,
2461                               &lvb, sizeof(lvb), lustre_swab_ost_lvb, lockh);
2462
2463         if ((*flags & LDLM_FL_HAS_INTENT && rc == ELDLM_LOCK_ABORTED) || !rc) {
2464                 CDEBUG(D_INODE, "received kms == "LPU64"\n", lvb.lvb_size);
2465                 lsm->lsm_oinfo->loi_rss = lvb.lvb_size;
2466         }
2467
2468         RETURN(rc);
2469 }
2470
2471 static int osc_match(struct obd_export *exp, struct lov_stripe_md *lsm,
2472                      __u32 type, ldlm_policy_data_t *policy, __u32 mode,
2473                      int *flags, void *data, struct lustre_handle *lockh)
2474 {
2475         struct ldlm_res_id res_id = { .name = {lsm->lsm_object_id} };
2476         struct obd_device *obd = exp->exp_obd;
2477         int rc;
2478         ENTRY;
2479
2480         OBD_FAIL_RETURN(OBD_FAIL_OSC_MATCH, -EIO);
2481
2482         /* Filesystem lock extents are extended to page boundaries so that
2483          * dealing with the page cache is a little smoother */
2484         policy->l_extent.start -= policy->l_extent.start & ~PAGE_MASK;
2485         policy->l_extent.end |= ~PAGE_MASK;
2486
2487         /* Next, search for already existing extent locks that will cover us */
2488         rc = ldlm_lock_match(obd->obd_namespace, *flags, &res_id, type,
2489                              policy, mode, lockh);
2490         if (rc) {
2491                 osc_set_data_with_check(lockh, data);
2492                 RETURN(rc);
2493         }
2494         /* If we're trying to read, we also search for an existing PW lock.  The
2495          * VFS and page cache already protect us locally, so lots of readers/
2496          * writers can share a single PW lock. */
2497         if (mode == LCK_PR) {
2498                 rc = ldlm_lock_match(obd->obd_namespace, *flags, &res_id, type,
2499                                      policy, LCK_PW, lockh);
2500                 if (rc == 1 && !(*flags & LDLM_FL_TEST_LOCK)) {
2501                         /* FIXME: This is not incredibly elegant, but it might
2502                          * be more elegant than adding another parameter to
2503                          * lock_match.  I want a second opinion. */
2504                         osc_set_data_with_check(lockh, data);
2505                         ldlm_lock_addref(lockh, LCK_PR);
2506                         ldlm_lock_decref(lockh, LCK_PW);
2507                 }
2508         }
2509         RETURN(rc);
2510 }
2511
2512 static int osc_cancel(struct obd_export *exp, struct lov_stripe_md *md,
2513                       __u32 mode, struct lustre_handle *lockh)
2514 {
2515         ENTRY;
2516
2517         if (mode == LCK_GROUP)
2518                 ldlm_lock_decref_and_cancel(lockh, mode);
2519         else
2520                 ldlm_lock_decref(lockh, mode);
2521
2522         RETURN(0);
2523 }
2524
2525 static int osc_cancel_unused(struct obd_export *exp,
2526                              struct lov_stripe_md *lsm, int flags, void *opaque)
2527 {
2528         struct obd_device *obd = class_exp2obd(exp);
2529         struct ldlm_res_id res_id = { .name = {lsm->lsm_object_id} };
2530
2531         return ldlm_cli_cancel_unused(obd->obd_namespace, &res_id, flags,
2532                                       opaque);
2533 }
2534
2535 static int osc_statfs(struct obd_device *obd, struct obd_statfs *osfs,
2536                       unsigned long max_age)
2537 {
2538         struct obd_statfs *msfs;
2539         struct ptlrpc_request *request;
2540         int rc, size = sizeof(*osfs);
2541         ENTRY;
2542
2543         /* We could possibly pass max_age in the request (as an absolute
2544          * timestamp or a "seconds.usec ago") so the target can avoid doing
2545          * extra calls into the filesystem if that isn't necessary (e.g.
2546          * during mount that would help a bit).  Having relative timestamps
2547          * is not so great if request processing is slow, while absolute
2548          * timestamps are not ideal because they need time synchronization. */
2549         request = ptlrpc_prep_req(obd->u.cli.cl_import, OST_STATFS,0,NULL,NULL);
2550         if (!request)
2551                 RETURN(-ENOMEM);
2552
2553         request->rq_replen = lustre_msg_size(1, &size);
2554         request->rq_request_portal = OST_CREATE_PORTAL; //XXX FIXME bug 249
2555
2556         rc = ptlrpc_queue_wait(request);
2557         if (rc)
2558                 GOTO(out, rc);
2559
2560         msfs = lustre_swab_repbuf(request, 0, sizeof(*msfs),
2561                                   lustre_swab_obd_statfs);
2562         if (msfs == NULL) {
2563                 CERROR("Can't unpack obd_statfs\n");
2564                 GOTO(out, rc = -EPROTO);
2565         }
2566
2567         memcpy(osfs, msfs, sizeof(*osfs));
2568
2569         EXIT;
2570  out:
2571         ptlrpc_req_finished(request);
2572         return rc;
2573 }
2574
2575 /* Retrieve object striping information.
2576  *
2577  * @lmmu is a pointer to an in-core struct with lmm_ost_count indicating
2578  * the maximum number of OST indices which will fit in the user buffer.
2579  * lmm_magic must be LOV_MAGIC (we only use 1 slot here).
2580  */
2581 static int osc_getstripe(struct lov_stripe_md *lsm, struct lov_user_md *lump)
2582 {
2583         struct lov_user_md lum, *lumk;
2584         int rc, lum_size;
2585         ENTRY;
2586
2587         if (!lsm)
2588                 RETURN(-ENODATA);
2589
2590         rc = copy_from_user(&lum, lump, sizeof(lum));
2591         if (rc)
2592                 RETURN(-EFAULT);
2593
2594         if (lum.lmm_magic != LOV_USER_MAGIC)
2595                 RETURN(-EINVAL);
2596
2597         if (lum.lmm_stripe_count > 0) {
2598                 lum_size = sizeof(lum) + sizeof(lum.lmm_objects[0]);
2599                 OBD_ALLOC(lumk, lum_size);
2600                 if (!lumk)
2601                         RETURN(-ENOMEM);
2602
2603                 lumk->lmm_objects[0].l_object_id = lsm->lsm_object_id;
2604         } else {
2605                 lum_size = sizeof(lum);
2606                 lumk = &lum;
2607         }
2608
2609         lumk->lmm_object_id = lsm->lsm_object_id;
2610         lumk->lmm_stripe_count = 1;
2611
2612         if (copy_to_user(lump, lumk, lum_size))
2613                 rc = -EFAULT;
2614
2615         if (lumk != &lum)
2616                 OBD_FREE(lumk, lum_size);
2617
2618         RETURN(rc);
2619 }
2620
2621 static int osc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
2622                          void *karg, void *uarg)
2623 {
2624         struct obd_device *obd = exp->exp_obd;
2625         struct obd_ioctl_data *data = karg;
2626         int err = 0;
2627         ENTRY;
2628
2629         MOD_INC_USE_COUNT;
2630
2631         switch (cmd) {
2632         case OBD_IOC_LOV_GET_CONFIG: {
2633                 char *buf;
2634                 struct lov_desc *desc;
2635                 struct obd_uuid uuid;
2636
2637                 buf = NULL;
2638                 len = 0;
2639                 if (obd_ioctl_getdata(&buf, &len, (void *)uarg))
2640                         GOTO(out, err = -EINVAL);
2641
2642                 data = (struct obd_ioctl_data *)buf;
2643
2644                 if (sizeof(*desc) > data->ioc_inllen1) {
2645                         OBD_FREE(buf, len);
2646                         GOTO(out, err = -EINVAL);
2647                 }
2648
2649                 if (data->ioc_inllen2 < sizeof(uuid)) {
2650                         OBD_FREE(buf, len);
2651                         GOTO(out, err = -EINVAL);
2652                 }
2653
2654                 desc = (struct lov_desc *)data->ioc_inlbuf1;
2655                 desc->ld_tgt_count = 1;
2656                 desc->ld_active_tgt_count = 1;
2657                 desc->ld_default_stripe_count = 1;
2658                 desc->ld_default_stripe_size = 0;
2659                 desc->ld_default_stripe_offset = 0;
2660                 desc->ld_pattern = 0;
2661                 memcpy(&desc->ld_uuid, &obd->obd_uuid, sizeof(uuid));
2662
2663                 memcpy(data->ioc_inlbuf2, &obd->obd_uuid, sizeof(uuid));
2664
2665                 err = copy_to_user((void *)uarg, buf, len);
2666                 if (err)
2667                         err = -EFAULT;
2668                 obd_ioctl_freedata(buf, len);
2669                 GOTO(out, err);
2670         }
2671         case LL_IOC_LOV_SETSTRIPE:
2672                 err = obd_alloc_memmd(exp, karg);
2673                 if (err > 0)
2674                         err = 0;
2675                 GOTO(out, err);
2676         case LL_IOC_LOV_GETSTRIPE:
2677                 err = osc_getstripe(karg, uarg);
2678                 GOTO(out, err);
2679         case OBD_IOC_CLIENT_RECOVER:
2680                 err = ptlrpc_recover_import(obd->u.cli.cl_import,
2681                                             data->ioc_inlbuf1);
2682                 if (err > 0)
2683                         err = 0;
2684                 GOTO(out, err);
2685         case IOC_OSC_SET_ACTIVE:
2686                 err = ptlrpc_set_import_active(obd->u.cli.cl_import,
2687                                                data->ioc_offset);
2688                 GOTO(out, err);
2689         default:
2690                 CDEBUG(D_INODE, "unrecognised ioctl %#x by %s\n", cmd, current->comm);
2691                 GOTO(out, err = -ENOTTY);
2692         }
2693 out:
2694         MOD_DEC_USE_COUNT;
2695         return err;
2696 }
2697
2698 static int osc_get_info(struct obd_export *exp, obd_count keylen,
2699                         void *key, __u32 *vallen, void *val)
2700 {
2701         ENTRY;
2702         if (!vallen || !val)
2703                 RETURN(-EFAULT);
2704
2705         if (keylen > strlen("lock_to_stripe") &&
2706             strcmp(key, "lock_to_stripe") == 0) {
2707                 __u32 *stripe = val;
2708                 *vallen = sizeof(*stripe);
2709                 *stripe = 0;
2710                 RETURN(0);
2711         } else if (keylen >= strlen("last_id") && strcmp(key, "last_id") == 0) {
2712                 struct ptlrpc_request *req;
2713                 obd_id *reply;
2714                 char *bufs[1] = {key};
2715                 int rc;
2716                 req = ptlrpc_prep_req(class_exp2cliimp(exp), OST_GET_INFO, 1,
2717                                       &keylen, bufs);
2718                 if (req == NULL)
2719                         RETURN(-ENOMEM);
2720
2721                 req->rq_replen = lustre_msg_size(1, vallen);
2722                 rc = ptlrpc_queue_wait(req);
2723                 if (rc)
2724                         GOTO(out, rc);
2725
2726                 reply = lustre_swab_repbuf(req, 0, sizeof(*reply),
2727                                            lustre_swab_ost_last_id);
2728                 if (reply == NULL) {
2729                         CERROR("Can't unpack OST last ID\n");
2730                         GOTO(out, rc = -EPROTO);
2731                 }
2732                 *((obd_id *)val) = *reply;
2733         out:
2734                 ptlrpc_req_finished(req);
2735                 RETURN(rc);
2736         }
2737         RETURN(-EINVAL);
2738 }
2739
2740 static int osc_set_info(struct obd_export *exp, obd_count keylen,
2741                         void *key, obd_count vallen, void *val)
2742 {
2743         struct ptlrpc_request *req;
2744         struct obd_device  *obd = exp->exp_obd;
2745         struct obd_import *imp = class_exp2cliimp(exp);
2746         struct llog_ctxt *ctxt;
2747         int rc, size = keylen;
2748         char *bufs[1] = {key};
2749         ENTRY;
2750
2751         if (keylen == strlen("next_id") &&
2752             memcmp(key, "next_id", strlen("next_id")) == 0) {
2753                 if (vallen != sizeof(obd_id))
2754                         RETURN(-EINVAL);
2755                 obd->u.cli.cl_oscc.oscc_next_id = *((obd_id*)val) + 1;
2756                 CDEBUG(D_INODE, "%s: set oscc_next_id = "LPU64"\n",
2757                        exp->exp_obd->obd_name,
2758                        obd->u.cli.cl_oscc.oscc_next_id);
2759
2760                 RETURN(0);
2761         }
2762
2763         if (keylen == strlen("growth_count") &&
2764             memcmp(key, "growth_count", strlen("growth_count")) == 0) {
2765                 if (vallen != sizeof(int))
2766                         RETURN(-EINVAL);
2767                 obd->u.cli.cl_oscc.oscc_grow_count = *((int*)val);
2768                 RETURN(0);
2769         }
2770
2771         if (keylen == strlen("unlinked") &&
2772             memcmp(key, "unlinked", keylen) == 0) {
2773                 struct osc_creator *oscc = &obd->u.cli.cl_oscc;
2774                 spin_lock(&oscc->oscc_lock);
2775                 oscc->oscc_flags &= ~OSCC_FLAG_NOSPC;
2776                 spin_unlock(&oscc->oscc_lock);
2777                 RETURN(0);
2778         }
2779
2780
2781         if (keylen == strlen("initial_recov") &&
2782             memcmp(key, "initial_recov", strlen("initial_recov")) == 0) {
2783                 struct obd_import *imp = exp->exp_obd->u.cli.cl_import;
2784                 if (vallen != sizeof(int))
2785                         RETURN(-EINVAL);
2786                 imp->imp_initial_recov = *(int *)val;
2787                 CDEBUG(D_HA, "%s: set imp_no_init_recov = %d\n",
2788                        exp->exp_obd->obd_name,
2789                        imp->imp_initial_recov);
2790                 RETURN(0);
2791         }
2792
2793         if (keylen < strlen("mds_conn") ||
2794             memcmp(key, "mds_conn", strlen("mds_conn")) != 0)
2795                 RETURN(-EINVAL);
2796
2797
2798         req = ptlrpc_prep_req(imp, OST_SET_INFO, 1, &size, bufs);
2799         if (req == NULL)
2800                 RETURN(-ENOMEM);
2801
2802         req->rq_replen = lustre_msg_size(0, NULL);
2803         rc = ptlrpc_queue_wait(req);
2804         ptlrpc_req_finished(req);
2805
2806         ctxt = llog_get_context(exp->exp_obd, LLOG_UNLINK_ORIG_CTXT);
2807         if (ctxt) {
2808                 rc = llog_initiator_connect(ctxt);
2809                 if (rc)
2810                         RETURN(rc);
2811         }
2812
2813         imp->imp_server_timeout = 1;
2814         CDEBUG(D_HA, "pinging OST %s\n", imp->imp_target_uuid.uuid);
2815         imp->imp_pingable = 1;
2816
2817         RETURN(rc);
2818 }
2819
2820
2821 static struct llog_operations osc_size_repl_logops = {
2822         lop_cancel: llog_obd_repl_cancel
2823 };
2824
2825 static struct llog_operations osc_unlink_orig_logops;
2826 static int osc_llog_init(struct obd_device *obd, struct obd_device *tgt,
2827                         int count, struct llog_catid *catid)
2828 {
2829         int rc;
2830         ENTRY;
2831
2832         osc_unlink_orig_logops = llog_lvfs_ops;
2833         osc_unlink_orig_logops.lop_setup = llog_obd_origin_setup;
2834         osc_unlink_orig_logops.lop_cleanup = llog_obd_origin_cleanup;
2835         osc_unlink_orig_logops.lop_add = llog_obd_origin_add;
2836         osc_unlink_orig_logops.lop_connect = llog_origin_connect;
2837
2838         rc = llog_setup(obd, LLOG_UNLINK_ORIG_CTXT, tgt, count,
2839                         &catid->lci_logid, &osc_unlink_orig_logops);
2840         if (rc)
2841                 RETURN(rc);
2842
2843         rc = llog_setup(obd, LLOG_SIZE_REPL_CTXT, tgt, count, NULL,
2844                         &osc_size_repl_logops);
2845         RETURN(rc);
2846 }
2847
2848 static int osc_llog_finish(struct obd_device *obd, int count)
2849 {
2850         int rc;
2851         ENTRY;
2852
2853         rc = llog_cleanup(llog_get_context(obd, LLOG_UNLINK_ORIG_CTXT));
2854         if (rc)
2855                 RETURN(rc);
2856
2857         rc = llog_cleanup(llog_get_context(obd, LLOG_SIZE_REPL_CTXT));
2858         RETURN(rc);
2859 }
2860
2861
2862 static int osc_connect(struct lustre_handle *exph,
2863                        struct obd_device *obd, struct obd_uuid *cluuid)
2864 {
2865         int rc;
2866
2867         rc = client_connect_import(exph, obd, cluuid);
2868
2869         return rc;
2870 }
2871
2872 static int osc_disconnect(struct obd_export *exp, int flags)
2873 {
2874         struct obd_device *obd = class_exp2obd(exp);
2875         struct llog_ctxt *ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
2876         int rc;
2877
2878         if (obd->u.cli.cl_conn_count == 1)
2879                 /* flush any remaining cancel messages out to the target */
2880                 llog_sync(ctxt, exp);
2881
2882         rc = client_disconnect_export(exp, flags);
2883         return rc;
2884 }
2885
2886 static int osc_import_event(struct obd_device *obd,
2887                             struct obd_import *imp,
2888                             enum obd_import_event event)
2889 {
2890         struct client_obd *cli;
2891         int rc = 0;
2892
2893         LASSERT(imp->imp_obd == obd);
2894
2895         switch (event) {
2896         case IMP_EVENT_DISCON: {
2897                 /* Only do this on the MDS OSC's */
2898                 if (imp->imp_server_timeout) {
2899                         struct osc_creator *oscc = &obd->u.cli.cl_oscc;
2900
2901                         spin_lock(&oscc->oscc_lock);
2902                         oscc->oscc_flags |= OSCC_FLAG_RECOVERING;
2903                         spin_unlock(&oscc->oscc_lock);
2904                 }
2905                 break;
2906         }
2907         case IMP_EVENT_INACTIVE: {
2908                 if (obd->obd_observer)
2909                         rc = obd_notify(obd->obd_observer, obd, 0);
2910                 break;
2911         }
2912         case IMP_EVENT_INVALIDATE: {
2913                 struct ldlm_namespace *ns = obd->obd_namespace;
2914
2915                 /* Reset grants */
2916                 cli = &obd->u.cli;
2917                 spin_lock(&cli->cl_loi_list_lock);
2918                 cli->cl_avail_grant = 0;
2919                 cli->cl_lost_grant = 0;
2920                 /* all pages go to failing rpcs due to the invalid import */
2921                 osc_check_rpcs(cli);
2922                 spin_unlock(&cli->cl_loi_list_lock);
2923
2924                 ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
2925
2926                 break;
2927         }
2928         case IMP_EVENT_ACTIVE: {
2929                 if (obd->obd_observer)
2930                         rc = obd_notify(obd->obd_observer, obd, 1);
2931                 break;
2932         }
2933         default:
2934                 CERROR("Unknown import event %d\n", event);
2935                 LBUG();
2936         }
2937         RETURN(rc);
2938 }
2939
2940 int osc_setup(struct obd_device *obd, obd_count len, void *buf)
2941 {
2942         int rc;
2943
2944         rc = ptlrpcd_addref();
2945         if (rc)
2946                 return rc;
2947
2948         rc = client_obd_setup(obd, len, buf);
2949         if (rc)
2950                 ptlrpcd_decref();
2951         else
2952                 oscc_init(obd);
2953
2954         RETURN(rc);
2955 }
2956
2957 int osc_cleanup(struct obd_device *obd, int flags)
2958 {
2959         int rc;
2960
2961         rc = client_obd_cleanup(obd, flags);
2962         ptlrpcd_decref();
2963         RETURN(rc);
2964 }
2965
2966
2967 struct obd_ops osc_obd_ops = {
2968         o_owner:        THIS_MODULE,
2969         o_attach:       osc_attach,
2970         o_detach:       osc_detach,
2971         o_setup:        osc_setup,
2972         o_cleanup:      osc_cleanup,
2973         o_connect:      osc_connect,
2974         o_disconnect:   osc_disconnect,
2975         o_statfs:       osc_statfs,
2976         o_packmd:       osc_packmd,
2977         o_unpackmd:     osc_unpackmd,
2978         o_create:       osc_create,
2979         o_destroy:      osc_destroy,
2980         o_getattr:      osc_getattr,
2981         o_getattr_async:osc_getattr_async,
2982         o_setattr:      osc_setattr,
2983         o_brw:          osc_brw,
2984         o_brw_async:    osc_brw_async,
2985         .o_prep_async_page =            osc_prep_async_page,
2986         .o_queue_async_io =             osc_queue_async_io,
2987         .o_set_async_flags =            osc_set_async_flags,
2988         .o_queue_group_io =             osc_queue_group_io,
2989         .o_trigger_group_io =           osc_trigger_group_io,
2990         .o_teardown_async_page =        osc_teardown_async_page,
2991         o_punch:        osc_punch,
2992         o_sync:         osc_sync,
2993         o_enqueue:      osc_enqueue,
2994         o_match:        osc_match,
2995         o_change_cbdata:osc_change_cbdata,
2996         o_cancel:       osc_cancel,
2997         o_cancel_unused:osc_cancel_unused,
2998         o_iocontrol:    osc_iocontrol,
2999         o_get_info:     osc_get_info,
3000         o_set_info:     osc_set_info,
3001         o_import_event: osc_import_event,
3002         o_llog_init:    osc_llog_init,
3003         o_llog_finish:  osc_llog_finish,
3004 };
3005
3006 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
3007 struct obd_ops sanosc_obd_ops = {
3008         o_owner:        THIS_MODULE,
3009         o_attach:       osc_attach,
3010         o_detach:       osc_detach,
3011         o_cleanup:      client_obd_cleanup,
3012         o_connect:      osc_connect,
3013         o_disconnect:   client_disconnect_export,
3014         o_statfs:       osc_statfs,
3015         o_packmd:       osc_packmd,
3016         o_unpackmd:     osc_unpackmd,
3017         o_create:       osc_real_create,
3018         o_destroy:      osc_destroy,
3019         o_getattr:      osc_getattr,
3020         o_getattr_async:osc_getattr_async,
3021         o_setattr:      osc_setattr,
3022         o_setup:        client_sanobd_setup,
3023         o_brw:          sanosc_brw,
3024         o_punch:        osc_punch,
3025         o_sync:         osc_sync,
3026         o_enqueue:      osc_enqueue,
3027         o_match:        osc_match,
3028         o_change_cbdata:osc_change_cbdata,
3029         o_cancel:       osc_cancel,
3030         o_cancel_unused:osc_cancel_unused,
3031         o_iocontrol:    osc_iocontrol,
3032         o_import_event: osc_import_event,
3033         o_llog_init:    osc_llog_init,
3034         o_llog_finish:  osc_llog_finish,
3035 };
3036 #endif
3037
3038 int __init osc_init(void)
3039 {
3040         struct lprocfs_static_vars lvars, sanlvars;
3041         int rc;
3042         ENTRY;
3043
3044         lprocfs_init_vars(osc, &lvars);
3045         lprocfs_init_vars(osc, &sanlvars);
3046
3047         rc = class_register_type(&osc_obd_ops, lvars.module_vars,
3048                                  LUSTRE_OSC_NAME);
3049         if (rc)
3050                 RETURN(rc);
3051
3052 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
3053         rc = class_register_type(&sanosc_obd_ops, sanlvars.module_vars,
3054                                  LUSTRE_SANOSC_NAME);
3055         if (rc)
3056                 class_unregister_type(LUSTRE_OSC_NAME);
3057 #endif
3058
3059         RETURN(rc);
3060 }
3061
3062 static void /*__exit*/ osc_exit(void)
3063 {
3064 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
3065         class_unregister_type(LUSTRE_SANOSC_NAME);
3066 #endif
3067         class_unregister_type(LUSTRE_OSC_NAME);
3068 }
3069
3070 #ifdef __KERNEL__
3071 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
3072 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC)");
3073 MODULE_LICENSE("GPL");
3074
3075 module_init(osc_init);
3076 module_exit(osc_exit);
3077 #endif