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         memcpy(oa, &body->oa, sizeof(*oa));
 856
 857         if (req->rq_reqmsg->opc == OST_WRITE) {
 858                 if (rc > 0) {
 859                         CERROR ("Unexpected +ve rc %d\n", rc);
 860                         RETURN(-EPROTO);
 861                 }
 862                 LASSERT (req->rq_bulk->bd_nob == requested_nob);
 863
 864                 RETURN(check_write_rcs(req, requested_nob, niocount,
 865                                        page_count, pga));
 866         }
 867
 868         if (rc > requested_nob) {
 869                 CERROR("Unexpected rc %d (%d requested)\n", rc, requested_nob);
 870                 RETURN(-EPROTO);
 871         }
 872
 873         if (rc != req->rq_bulk->bd_nob_transferred) {
 874                 CERROR ("Unexpected rc %d (%d transferred)\n",
 875                         rc, req->rq_bulk->bd_nob_transferred);
 876                 return (-EPROTO);
 877         }
 878
 879         if (rc < requested_nob)
 880                 handle_short_read(rc, page_count, pga);
 881
 882 #if CHECKSUM_BULK
 883         if (oa->o_valid & OBD_MD_FLCKSUM) {
 884                 const struct ptlrpc_peer *peer =
 885                         &req->rq_import->imp_connection->c_peer;
 886                 static int cksum_counter;
 887                 obd_count server_cksum = oa->o_cksum;
 888                 obd_count cksum = cksum_pages(rc, page_count, pga);
 889                 char str[PTL_NALFMT_SIZE];
 890
 891                 portals_nid2str(peer->peer_ni->pni_number, peer->peer_nid, str);
 892
 893                 cksum_counter++;
 894                 if (server_cksum != cksum) {
 895                         CERROR("Bad checksum: server %x, client %x, server NID "
 896                                LPX64" (%s)\n", server_cksum, cksum,
 897                                peer->peer_nid, str);
 898                         cksum_counter = 0;
 899                         oa->o_cksum = cksum;
 900                 } else if ((cksum_counter & (-cksum_counter)) == cksum_counter){
 901                         CWARN("Checksum %u from "LPX64" (%s) OK: %x\n",
 902                               cksum_counter, peer->peer_nid, str, cksum);
 903                 }
 904         } else {
 905                 static int cksum_missed;
 906
 907                 cksum_missed++;
 908                 if ((cksum_missed & (-cksum_missed)) == cksum_missed)
 909                         CERROR("Request checksum %u from "LPX64", no reply\n",
 910                                cksum_missed,
 911                                req->rq_import->imp_connection->c_peer.peer_nid);
 912         }
 913 #endif
 914         RETURN(0);
 915 }
 916
 917 static int osc_brw_internal(int cmd, struct obd_export *exp,struct obdo *oa,
 918                             struct lov_stripe_md *lsm,
 919                             obd_count page_count, struct brw_page *pga)
 920 {
 921         int                    requested_nob;
 922         int                    niocount;
 923         struct ptlrpc_request *request;
 924         int                    rc;
 925         ENTRY;
 926
 927 restart_bulk:
 928         rc = osc_brw_prep_request(cmd, class_exp2cliimp(exp), oa, lsm,
 929                                   page_count, pga, &requested_nob, &niocount,
 930                                   &request);
 931         /* NB ^ sets rq_no_resend */
 932
 933         if (rc != 0)
 934                 return (rc);
 935
 936         rc = ptlrpc_queue_wait(request);
 937
 938         if (rc == -ETIMEDOUT && request->rq_resend) {
 939                 DEBUG_REQ(D_HA, request,  "BULK TIMEOUT");
 940                 ptlrpc_req_finished(request);
 941                 goto restart_bulk;
 942         }
 943
 944         rc = osc_brw_fini_request(request, oa, requested_nob, niocount,
 945                                   page_count, pga, rc);
 946
 947         ptlrpc_req_finished(request);
 948         RETURN (rc);
 949 }
 950
 951 static int brw_interpret(struct ptlrpc_request *request,
 952                          struct osc_brw_async_args *aa, int rc)
 953 {
 954         struct obdo *oa      = aa->aa_oa;
 955         int requested_nob    = aa->aa_requested_nob;
 956         int niocount         = aa->aa_nio_count;
 957         obd_count page_count = aa->aa_page_count;
 958         struct brw_page *pga = aa->aa_pga;
 959         ENTRY;
 960
 961         /* XXX bug 937 here */
 962         if (rc == -ETIMEDOUT && request->rq_resend) {
 963                 DEBUG_REQ(D_HA, request,  "BULK TIMEOUT");
 964                 LBUG(); /* re-send.  later. */
 965                 //goto restart_bulk;
 966         }
 967
 968         rc = osc_brw_fini_request(request, oa, requested_nob, niocount,
 969                                   page_count, pga, rc);
 970         RETURN (rc);
 971 }
 972
 973 static int async_internal(int cmd, struct obd_export *exp, struct obdo *oa,
 974                           struct lov_stripe_md *lsm, obd_count page_count,
 975                           struct brw_page *pga, struct ptlrpc_request_set *set)
 976 {
 977         struct ptlrpc_request     *request;
 978         int                        requested_nob;
 979         int                        nio_count;
 980         struct osc_brw_async_args *aa;
 981         int                        rc;
 982         ENTRY;
 983
 984         rc = osc_brw_prep_request(cmd, class_exp2cliimp(exp), oa, lsm,
 985                                   page_count, pga, &requested_nob, &nio_count,
 986                                   &request);
 987         /* NB ^ sets rq_no_resend */
 988
 989         if (rc == 0) {
 990                 LASSERT(sizeof(*aa) <= sizeof(request->rq_async_args));
 991                 aa = (struct osc_brw_async_args *)&request->rq_async_args;
 992                 aa->aa_oa = oa;
 993                 aa->aa_requested_nob = requested_nob;
 994                 aa->aa_nio_count = nio_count;
 995                 aa->aa_page_count = page_count;
 996                 aa->aa_pga = pga;
 997
 998                 request->rq_interpret_reply = brw_interpret;
 999                 ptlrpc_set_add_req(set, request);
1000         }
1001         RETURN (rc);
1002 }
1003
1004 #ifndef min_t
1005 #define min_t(type,x,y) \
1006         ({ type __x = (x); type __y = (y); __x < __y ? __x: __y; })
1007 #endif
1008
1009 /*
1010  * ugh, we want disk allocation on the target to happen in offset order.  we'll
1011  * follow sedgewicks advice and stick to the dead simple shellsort -- it'll do
1012  * fine for our small page arrays and doesn't require allocation.  its an
1013  * insertion sort that swaps elements that are strides apart, shrinking the
1014  * stride down until its '1' and the array is sorted.
1015  */
1016 static void sort_brw_pages(struct brw_page *array, int num)
1017 {
1018         int stride, i, j;
1019         struct brw_page tmp;
1020
1021         if (num == 1)
1022                 return;
1023         for (stride = 1; stride < num ; stride = (stride * 3) + 1)
1024                 ;
1025
1026         do {
1027                 stride /= 3;
1028                 for (i = stride ; i < num ; i++) {
1029                         tmp = array[i];
1030                         j = i;
1031                         while (j >= stride && array[j - stride].off > tmp.off) {
1032                                 array[j] = array[j - stride];
1033                                 j -= stride;
1034                         }
1035                         array[j] = tmp;
1036                 }
1037         } while (stride > 1);
1038 }
1039
1040 /* make sure we the regions we're passing to elan don't violate its '4
1041  * fragments' constraint.  portal headers are a fragment, all full
1042  * PAGE_SIZE long pages count as 1 fragment, and each partial page
1043  * counts as a fragment.  I think.  see bug 934. */
1044 static obd_count check_elan_limit(struct brw_page *pg, obd_count pages)
1045 {
1046         int frags_left = 3;
1047         int saw_whole_frag = 0;
1048         int i;
1049
1050         for (i = 0 ; frags_left && i < pages ; pg++, i++) {
1051                 if (pg->count == PAGE_SIZE) {
1052                         if (!saw_whole_frag) {
1053                                 saw_whole_frag = 1;
1054                                 frags_left--;
1055                         }
1056                 } else {
1057                         frags_left--;
1058                 }
1059         }
1060         return i;
1061 }
1062
1063 static int osc_brw(int cmd, struct obd_export *exp, struct obdo *oa,
1064                    struct lov_stripe_md *md, obd_count page_count,
1065                    struct brw_page *pga, struct obd_trans_info *oti)
1066 {
1067         ENTRY;
1068
1069         if (cmd == OBD_BRW_CHECK) {
1070                 /* The caller just wants to know if there's a chance that this
1071                  * I/O can succeed */
1072                 struct obd_import *imp = class_exp2cliimp(exp);
1073
1074                 if (imp == NULL || imp->imp_invalid)
1075                         RETURN(-EIO);
1076                 RETURN(0);
1077         }
1078
1079         while (page_count) {
1080                 obd_count pages_per_brw;
1081                 int rc;
1082
1083                 if (page_count > PTLRPC_MAX_BRW_PAGES)
1084                         pages_per_brw = PTLRPC_MAX_BRW_PAGES;
1085                 else
1086                         pages_per_brw = page_count;
1087
1088                 sort_brw_pages(pga, pages_per_brw);
1089                 pages_per_brw = check_elan_limit(pga, pages_per_brw);
1090
1091                 rc = osc_brw_internal(cmd, exp, oa, md, pages_per_brw, pga);
1092
1093                 if (rc != 0)
1094                         RETURN(rc);
1095
1096                 page_count -= pages_per_brw;
1097                 pga += pages_per_brw;
1098         }
1099         RETURN(0);
1100 }
1101
1102 static int osc_brw_async(int cmd, struct obd_export *exp, struct obdo *oa,
1103                          struct lov_stripe_md *md, obd_count page_count,
1104                          struct brw_page *pga, struct ptlrpc_request_set *set,
1105                          struct obd_trans_info *oti)
1106 {
1107         ENTRY;
1108
1109         if (cmd == OBD_BRW_CHECK) {
1110                 /* The caller just wants to know if there's a chance that this
1111                  * I/O can succeed */
1112                 struct obd_import *imp = class_exp2cliimp(exp);
1113
1114                 if (imp == NULL || imp->imp_invalid)
1115                         RETURN(-EIO);
1116                 RETURN(0);
1117         }
1118
1119         while (page_count) {
1120                 obd_count pages_per_brw;
1121                 int rc;
1122
1123                 if (page_count > PTLRPC_MAX_BRW_PAGES)
1124                         pages_per_brw = PTLRPC_MAX_BRW_PAGES;
1125                 else
1126                         pages_per_brw = page_count;
1127
1128                 sort_brw_pages(pga, pages_per_brw);
1129                 pages_per_brw = check_elan_limit(pga, pages_per_brw);
1130
1131                 rc = async_internal(cmd, exp, oa, md, pages_per_brw, pga, set);
1132
1133                 if (rc != 0)
1134                         RETURN(rc);
1135
1136                 page_count -= pages_per_brw;
1137                 pga += pages_per_brw;
1138         }
1139         RETURN(0);
1140 }
1141
1142 static void osc_check_rpcs(struct client_obd *cli);
1143 static void osc_exit_cache(struct client_obd *cli, struct osc_async_page *oap,
1144                            int sent);
1145 static void loi_list_maint(struct client_obd *cli, struct lov_oinfo *loi);
1146 static void lop_update_pending(struct client_obd *cli,
1147                                struct loi_oap_pages *lop, int cmd, int delta);
1148
1149 /* this is called when a sync waiter receives an interruption.  Its job is to
1150  * get the caller woken as soon as possible.  If its page hasn't been put in an
1151  * rpc yet it can dequeue immediately.  Otherwise it has to mark the rpc as
1152  * desiring interruption which will forcefully complete the rpc once the rpc
1153  * has timed out */
1154 static void osc_occ_interrupted(struct oig_callback_context *occ)
1155 {
1156         struct osc_async_page *oap;
1157         struct loi_oap_pages *lop;
1158         struct lov_oinfo *loi;
1159         ENTRY;
1160
1161         /* XXX member_of() */
1162         oap = list_entry(occ, struct osc_async_page, oap_occ);
1163
1164         spin_lock(&oap->oap_cli->cl_loi_list_lock);
1165
1166         oap->oap_interrupted = 1;
1167
1168         /* ok, it's been put in an rpc. */
1169         if (oap->oap_request != NULL) {
1170                 ptlrpc_mark_interrupted(oap->oap_request);
1171                 ptlrpcd_wake(oap->oap_request);
1172                 GOTO(unlock, 0);
1173         }
1174
1175         /* we don't get interruption callbacks until osc_trigger_sync_io()
1176          * has been called and put the sync oaps in the pending/urgent lists.*/
1177         if (!list_empty(&oap->oap_pending_item)) {
1178                 list_del_init(&oap->oap_pending_item);
1179                 if (oap->oap_async_flags & ASYNC_URGENT)
1180                         list_del_init(&oap->oap_urgent_item);
1181
1182                 loi = oap->oap_loi;
1183                 lop = (oap->oap_cmd == OBD_BRW_WRITE) ?
1184                         &loi->loi_write_lop : &loi->loi_read_lop;
1185                 lop_update_pending(oap->oap_cli, lop, oap->oap_cmd, -1);
1186                 loi_list_maint(oap->oap_cli, oap->oap_loi);
1187
1188                 oig_complete_one(oap->oap_oig, &oap->oap_occ, 0);
1189                 oap->oap_oig = NULL;
1190         }
1191
1192 unlock:
1193         spin_unlock(&oap->oap_cli->cl_loi_list_lock);
1194 }
1195
1196 /* this must be called holding the loi list lock to give coverage to exit_cache,
1197  * async_flag maintenance, and oap_request */
1198 static void osc_ap_completion(struct client_obd *cli, struct obdo *oa,
1199                               struct osc_async_page *oap, int sent, int rc)
1200 {
1201         osc_exit_cache(cli, oap, sent);
1202         oap->oap_async_flags = 0;
1203         oap->oap_interrupted = 0;
1204
1205         if (oap->oap_request != NULL) {
1206                 ptlrpc_req_finished(oap->oap_request);
1207                 oap->oap_request = NULL;
1208         }
1209
1210         if (rc == 0 && oa != NULL)
1211                 oap->oap_loi->loi_blocks = oa->o_blocks;
1212
1213         if (oap->oap_oig) {
1214                 oig_complete_one(oap->oap_oig, &oap->oap_occ, rc);
1215                 oap->oap_oig = NULL;
1216                 EXIT;
1217                 return;
1218         }
1219
1220         oap->oap_caller_ops->ap_completion(oap->oap_caller_data, oap->oap_cmd,
1221                                            oa, rc);
1222 }
1223
1224 static int brw_interpret_oap(struct ptlrpc_request *request,
1225                              struct osc_brw_async_args *aa, int rc)
1226 {
1227         struct osc_async_page *oap;
1228         struct client_obd *cli;
1229         struct list_head *pos, *n;
1230         ENTRY;
1231
1232
1233         rc = osc_brw_fini_request(request, aa->aa_oa, aa->aa_requested_nob,
1234                                   aa->aa_nio_count, aa->aa_page_count,
1235                                   aa->aa_pga, rc);
1236
1237         CDEBUG(D_INODE, "request %p aa %p rc %d\n", request, aa, rc);
1238
1239         cli = aa->aa_cli;
1240         /* in failout recovery we ignore writeback failure and want
1241          * to just tell llite to unlock the page and continue */
1242         if (request->rq_reqmsg->opc == OST_WRITE &&
1243             (cli->cl_import == NULL || cli->cl_import->imp_invalid)) {
1244                 CDEBUG(D_INODE, "flipping to rc 0 imp %p inv %d\n",
1245                        cli->cl_import,
1246                        cli->cl_import ? cli->cl_import->imp_invalid : -1);
1247                 rc = 0;
1248         }
1249
1250         spin_lock(&cli->cl_loi_list_lock);
1251
1252         /* We need to decrement before osc_ap_completion->osc_wake_cache_waiters
1253          * is called so we know whether to go to sync BRWs or wait for more
1254          * RPCs to complete */
1255         cli->cl_brw_in_flight--;
1256
1257         /* the caller may re-use the oap after the completion call so
1258          * we need to clean it up a little */
1259         list_for_each_safe(pos, n, &aa->aa_oaps) {
1260                 oap = list_entry(pos, struct osc_async_page, oap_rpc_item);
1261
1262                 //CDEBUG(D_INODE, "page %p index %lu oap %p\n",
1263                        //oap->oap_page, oap->oap_page->index, oap);
1264
1265                 list_del_init(&oap->oap_rpc_item);
1266                 osc_ap_completion(cli, aa->aa_oa, oap, 1, rc);
1267         }
1268
1269         osc_wake_cache_waiters(cli);
1270         osc_check_rpcs(cli);
1271
1272         spin_unlock(&cli->cl_loi_list_lock);
1273
1274         obdo_free(aa->aa_oa);
1275         OBD_FREE(aa->aa_pga, aa->aa_page_count * sizeof(struct brw_page));
1276
1277         RETURN(0);
1278 }
1279
1280 static struct ptlrpc_request *osc_build_req(struct client_obd *cli,
1281                                             struct list_head *rpc_list,
1282                                             int page_count, int cmd)
1283 {
1284         struct ptlrpc_request *req;
1285         struct brw_page *pga = NULL;
1286         int requested_nob, nio_count;
1287         struct osc_brw_async_args *aa;
1288         struct obdo *oa = NULL;
1289         struct obd_async_page_ops *ops = NULL;
1290         void *caller_data = NULL;
1291         struct list_head *pos;
1292         int i, rc;
1293
1294         LASSERT(!list_empty(rpc_list));
1295
1296         OBD_ALLOC(pga, sizeof(*pga) * page_count);
1297         if (pga == NULL)
1298                 RETURN(ERR_PTR(-ENOMEM));
1299
1300         oa = obdo_alloc();
1301         if (oa == NULL)
1302                 GOTO(out, req = ERR_PTR(-ENOMEM));
1303
1304         i = 0;
1305         list_for_each(pos, rpc_list) {
1306                 struct osc_async_page *oap;
1307
1308                 oap = list_entry(pos, struct osc_async_page, oap_rpc_item);
1309                 if (ops == NULL) {
1310                         ops = oap->oap_caller_ops;
1311                         caller_data = oap->oap_caller_data;
1312                 }
1313                 pga[i].off = oap->oap_obj_off + oap->oap_page_off;
1314                 pga[i].pg = oap->oap_page;
1315                 pga[i].count = oap->oap_count;
1316                 pga[i].flag = oap->oap_brw_flags;
1317                 CDEBUG(0, "put page %p index %lu oap %p flg %x to pga\n",
1318                        pga[i].pg, oap->oap_page->index, oap, pga[i].flag);
1319                 i++;
1320         }
1321
1322         /* always get the data for the obdo for the rpc */
1323         LASSERT(ops != NULL);
1324         ops->ap_fill_obdo(caller_data, cmd, oa);
1325
1326         sort_brw_pages(pga, page_count);
1327         rc = osc_brw_prep_request(cmd, cli->cl_import, oa, NULL, page_count,
1328                                   pga, &requested_nob, &nio_count, &req);
1329         if (rc != 0) {
1330                 CERROR("prep_req failed: %d\n", rc);
1331                 GOTO(out, req = ERR_PTR(rc));
1332         }
1333
1334         LASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
1335         aa = (struct osc_brw_async_args *)&req->rq_async_args;
1336         aa->aa_oa = oa;
1337         aa->aa_requested_nob = requested_nob;
1338         aa->aa_nio_count = nio_count;
1339         aa->aa_page_count = page_count;
1340         aa->aa_pga = pga;
1341         aa->aa_cli = cli;
1342
1343 out:
1344         if (IS_ERR(req)) {
1345                 if (oa)
1346                         obdo_free(oa);
1347                 if (pga)
1348                         OBD_FREE(pga, sizeof(*pga) * page_count);
1349         }
1350         RETURN(req);
1351 }
1352
1353 static void lop_update_pending(struct client_obd *cli,
1354                                struct loi_oap_pages *lop, int cmd, int delta)
1355 {
1356         lop->lop_num_pending += delta;
1357         if (cmd == OBD_BRW_WRITE)
1358                 cli->cl_pending_w_pages += delta;
1359         else
1360                 cli->cl_pending_r_pages += delta;
1361 }
1362
1363 /* the loi lock is held across this function but it's allowed to release
1364  * and reacquire it during its work */
1365 static int osc_send_oap_rpc(struct client_obd *cli, struct lov_oinfo *loi,
1366                             int cmd, struct loi_oap_pages *lop)
1367 {
1368         struct ptlrpc_request *request;
1369         obd_count page_count = 0;
1370         struct list_head *tmp, *pos;
1371         struct osc_async_page *oap = NULL;
1372         struct osc_brw_async_args *aa;
1373         struct obd_async_page_ops *ops;
1374         LIST_HEAD(rpc_list);
1375         ENTRY;
1376
1377         /* first we find the pages we're allowed to work with */
1378         list_for_each_safe(pos, tmp, &lop->lop_pending) {
1379                 oap = list_entry(pos, struct osc_async_page, oap_pending_item);
1380                 ops = oap->oap_caller_ops;
1381
1382                 LASSERT(oap->oap_magic == OAP_MAGIC);
1383
1384                 /* in llite being 'ready' equates to the page being locked
1385                  * until completion unlocks it.  commit_write submits a page
1386                  * as not ready because its unlock will happen unconditionally
1387                  * as the call returns.  if we race with commit_write giving
1388                  * us that page we dont' want to create a hole in the page
1389                  * stream, so we stop and leave the rpc to be fired by
1390                  * another dirtier or kupdated interval (the not ready page
1391                  * will still be on the dirty list).  we could call in
1392                  * at the end of ll_file_write to process the queue again. */
1393                 if (!(oap->oap_async_flags & ASYNC_READY)) {
1394                         int rc = ops->ap_make_ready(oap->oap_caller_data, cmd);
1395                         if (rc < 0)
1396                                 CDEBUG(D_INODE, "oap %p page %p returned %d "
1397                                                 "instead of ready\n", oap,
1398                                                 oap->oap_page, rc);
1399                         switch (rc) {
1400                         case -EAGAIN:
1401                                 /* llite is telling us that the page is still
1402                                  * in commit_write and that we should try
1403                                  * and put it in an rpc again later.  we
1404                                  * break out of the loop so we don't create
1405                                  * a hole in the sequence of pages in the rpc
1406                                  * stream.*/
1407                                 pos = NULL;
1408                                 break;
1409                         case -EINTR:
1410                                 /* the io isn't needed.. tell the checks
1411                                  * below to complete the rpc with EINTR */
1412                                 oap->oap_async_flags |= ASYNC_COUNT_STABLE;
1413                                 oap->oap_count = -EINTR;
1414                                 break;
1415                         case 0:
1416                                 oap->oap_async_flags |= ASYNC_READY;
1417                                 break;
1418                         default:
1419                                 LASSERTF(0, "oap %p page %p returned %d "
1420                                             "from make_ready\n", oap,
1421                                             oap->oap_page, rc);
1422                                 break;
1423                         }
1424                 }
1425                 if (pos == NULL)
1426                         break;
1427
1428                 /* take the page out of our book-keeping */
1429                 list_del_init(&oap->oap_pending_item);
1430                 lop_update_pending(cli, lop, cmd, -1);
1431                 if (!list_empty(&oap->oap_urgent_item))
1432                         list_del_init(&oap->oap_urgent_item);
1433
1434                 /* ask the caller for the size of the io as the rpc leaves. */
1435                 if (!(oap->oap_async_flags & ASYNC_COUNT_STABLE))
1436                         oap->oap_count =
1437                                 ops->ap_refresh_count(oap->oap_caller_data,cmd);
1438                 if (oap->oap_count <= 0) {
1439                         CDEBUG(D_CACHE, "oap %p count %d, completing\n", oap,
1440                                oap->oap_count);
1441                         osc_ap_completion(cli, NULL, oap, 0, oap->oap_count);
1442                         continue;
1443                 }
1444
1445                 /* now put the page back in our accounting */
1446                 list_add_tail(&oap->oap_rpc_item, &rpc_list);
1447                 if (++page_count >= cli->cl_max_pages_per_rpc)
1448                         break;
1449         }
1450
1451         osc_wake_cache_waiters(cli);
1452
1453         if (page_count == 0)
1454                 RETURN(0);
1455
1456         loi_list_maint(cli, loi);
1457         spin_unlock(&cli->cl_loi_list_lock);
1458
1459         request = osc_build_req(cli, &rpc_list, page_count, cmd);
1460         if (IS_ERR(request)) {
1461                 /* this should happen rarely and is pretty bad, it makes the
1462                  * pending list not follow the dirty order */
1463                 spin_lock(&cli->cl_loi_list_lock);
1464                 list_for_each_safe(pos, tmp, &rpc_list) {
1465                         oap = list_entry(pos, struct osc_async_page,
1466                                          oap_rpc_item);
1467                         list_del_init(&oap->oap_rpc_item);
1468
1469                         /* queued sync pages can be torn down while the pages
1470                          * were between the pending list and the rpc */
1471                         if (oap->oap_interrupted) {
1472                                 CDEBUG(D_INODE, "oap %p interrupted\n", oap);
1473                                 osc_ap_completion(cli, NULL, oap, 0,
1474                                                   oap->oap_count);
1475                                 continue;
1476                         }
1477
1478                         /* put the page back in the loi/lop lists */
1479                         list_add_tail(&oap->oap_pending_item,
1480                                       &lop->lop_pending);
1481                         lop_update_pending(cli, lop, cmd, 1);
1482                         if (oap->oap_async_flags & ASYNC_URGENT)
1483                                 list_add(&oap->oap_urgent_item,
1484                                          &lop->lop_urgent);
1485                 }
1486                 loi_list_maint(cli, loi);
1487                 RETURN(PTR_ERR(request));
1488         }
1489
1490         LASSERT(sizeof(*aa) <= sizeof(request->rq_async_args));
1491         aa = (struct osc_brw_async_args *)&request->rq_async_args;
1492         INIT_LIST_HEAD(&aa->aa_oaps);
1493         list_splice(&rpc_list, &aa->aa_oaps);
1494         INIT_LIST_HEAD(&rpc_list);
1495
1496 #ifdef __KERNEL__
1497         if (cmd == OBD_BRW_READ) {
1498                 lprocfs_oh_tally_log2(&cli->cl_read_page_hist, page_count);
1499                 lprocfs_oh_tally(&cli->cl_read_rpc_hist, cli->cl_brw_in_flight);
1500         } else {
1501                 lprocfs_oh_tally_log2(&cli->cl_write_page_hist, page_count);
1502                 lprocfs_oh_tally(&cli->cl_write_rpc_hist,
1503                                  cli->cl_brw_in_flight);
1504         }
1505 #endif
1506
1507         spin_lock(&cli->cl_loi_list_lock);
1508
1509         cli->cl_brw_in_flight++;
1510         /* queued sync pages can be torn down while the pages
1511          * were between the pending list and the rpc */
1512         list_for_each(pos, &aa->aa_oaps) {
1513                 oap = list_entry(pos, struct osc_async_page, oap_rpc_item);
1514                 if (oap->oap_interrupted) {
1515                         CDEBUG(D_INODE, "oap %p in req %p interrupted\n",
1516                                oap, request);
1517                         ptlrpc_mark_interrupted(request);
1518                         break;
1519                 }
1520         }
1521
1522         CDEBUG(D_INODE, "req %p: %d pages, aa %p.  now %d in flight\n", request,
1523                page_count, aa, cli->cl_brw_in_flight);
1524
1525         oap->oap_request = ptlrpc_request_addref(request);
1526         request->rq_interpret_reply = brw_interpret_oap;
1527         ptlrpcd_add_req(request);
1528         RETURN(1);
1529 }
1530
1531 static int lop_makes_rpc(struct client_obd *cli, struct loi_oap_pages *lop,
1532                          int cmd)
1533 {
1534         int optimal;
1535         ENTRY;
1536
1537         if (lop->lop_num_pending == 0)
1538                 RETURN(0);
1539
1540         /* if we have an invalid import we want to drain the queued pages
1541          * by forcing them through rpcs that immediately fail and complete
1542          * the pages.  recovery relies on this to empty the queued pages
1543          * before canceling the locks and evicting down the llite pages */
1544         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
1545                 RETURN(1);
1546
1547         /* stream rpcs in queue order as long as as there is an urgent page
1548          * queued.  this is our cheap solution for good batching in the case
1549          * where writepage marks some random page in the middle of the file as
1550          * urgent because of, say, memory pressure */
1551         if (!list_empty(&lop->lop_urgent))
1552                 RETURN(1);
1553
1554         /* fire off rpcs when we have 'optimal' rpcs as tuned for the wire. */
1555         optimal = cli->cl_max_pages_per_rpc;
1556         if (cmd == OBD_BRW_WRITE) {
1557                 /* trigger a write rpc stream as long as there are dirtiers
1558                  * waiting for space.  as they're waiting, they're not going to
1559                  * create more pages to coallesce with what's waiting.. */
1560                 if (!list_empty(&cli->cl_cache_waiters))
1561                         RETURN(1);
1562
1563                 /* *2 to avoid triggering rpcs that would want to include pages
1564                  * that are being queued but which can't be made ready until
1565                  * the queuer finishes with the page. this is a wart for
1566                  * llite::commit_write() */
1567                 optimal += 16;
1568         }
1569         if (lop->lop_num_pending >= optimal)
1570                 RETURN(1);
1571
1572         RETURN(0);
1573 }
1574
1575 static void on_list(struct list_head *item, struct list_head *list,
1576                     int should_be_on)
1577 {
1578         if (list_empty(item) && should_be_on)
1579                 list_add_tail(item, list);
1580         else if (!list_empty(item) && !should_be_on)
1581                 list_del_init(item);
1582 }
1583
1584 /* maintain the loi's cli list membership invariants so that osc_send_oap_rpc
1585  * can find pages to build into rpcs quickly */
1586 static void loi_list_maint(struct client_obd *cli, struct lov_oinfo *loi)
1587 {
1588         on_list(&loi->loi_cli_item, &cli->cl_loi_ready_list,
1589                 lop_makes_rpc(cli, &loi->loi_write_lop, OBD_BRW_WRITE) ||
1590                 lop_makes_rpc(cli, &loi->loi_read_lop, OBD_BRW_READ));
1591
1592         on_list(&loi->loi_write_item, &cli->cl_loi_write_list,
1593                 loi->loi_write_lop.lop_num_pending);
1594
1595         on_list(&loi->loi_read_item, &cli->cl_loi_read_list,
1596                 loi->loi_read_lop.lop_num_pending);
1597 }
1598
1599 #define LOI_DEBUG(LOI, STR, args...)                                     \
1600         CDEBUG(D_INODE, "loi ready %d wr %d:%d rd %d:%d " STR,           \
1601                !list_empty(&(LOI)->loi_cli_item),                        \
1602                (LOI)->loi_write_lop.lop_num_pending,                     \
1603                !list_empty(&(LOI)->loi_write_lop.lop_urgent),            \
1604                (LOI)->loi_read_lop.lop_num_pending,                      \
1605                !list_empty(&(LOI)->loi_read_lop.lop_urgent),             \
1606                args)                                                     \
1607
1608 struct lov_oinfo *osc_next_loi(struct client_obd *cli)
1609 {
1610         ENTRY;
1611         /* first return all objects which we already know to have
1612          * pages ready to be stuffed into rpcs */
1613         if (!list_empty(&cli->cl_loi_ready_list))
1614                 RETURN(list_entry(cli->cl_loi_ready_list.next,
1615                                   struct lov_oinfo, loi_cli_item));
1616
1617         /* then if we have cache waiters, return all objects with queued
1618          * writes.  This is especially important when many small files
1619          * have filled up the cache and not been fired into rpcs because
1620          * they don't pass the nr_pending/object threshhold */
1621         if (!list_empty(&cli->cl_cache_waiters) &&
1622             !list_empty(&cli->cl_loi_write_list))
1623                 RETURN(list_entry(cli->cl_loi_write_list.next,
1624                                   struct lov_oinfo, loi_write_item));
1625
1626         /* then return all queued objects when we have an invalid import
1627          * so that they get flushed */
1628         if (cli->cl_import == NULL || cli->cl_import->imp_invalid) {
1629                 if (!list_empty(&cli->cl_loi_write_list))
1630                         RETURN(list_entry(cli->cl_loi_write_list.next,
1631                                           struct lov_oinfo, loi_write_item));
1632                 if (!list_empty(&cli->cl_loi_read_list))
1633                         RETURN(list_entry(cli->cl_loi_read_list.next,
1634                                           struct lov_oinfo, loi_read_item));
1635         }
1636         RETURN(NULL);
1637 }
1638
1639 /* called with the loi list lock held */
1640 static void osc_check_rpcs(struct client_obd *cli)
1641 {
1642         struct lov_oinfo *loi;
1643         int rc = 0, race_counter = 0;
1644         ENTRY;
1645
1646         while ((loi = osc_next_loi(cli)) != NULL) {
1647                 LOI_DEBUG(loi, "%d in flight\n", cli->cl_brw_in_flight);
1648
1649                 if (cli->cl_brw_in_flight >= cli->cl_max_rpcs_in_flight)
1650                         break;
1651
1652                 /* attempt some read/write balancing by alternating between
1653                  * reads and writes in an object.  The makes_rpc checks here
1654                  * would be redundant if we were getting read/write work items
1655                  * instead of objects.  we don't want send_oap_rpc to drain a
1656                  * partial read pending queue when we're given this object to
1657                  * do io on writes while there are cache waiters */
1658                 if (lop_makes_rpc(cli, &loi->loi_write_lop, OBD_BRW_WRITE)) {
1659                         rc = osc_send_oap_rpc(cli, loi, OBD_BRW_WRITE,
1660                                               &loi->loi_write_lop);
1661                         if (rc < 0)
1662                                 break;
1663                         if (rc > 0)
1664                                 race_counter = 0;
1665                         else
1666                                 race_counter++;
1667                 }
1668                 if (lop_makes_rpc(cli, &loi->loi_read_lop, OBD_BRW_READ)) {
1669                         rc = osc_send_oap_rpc(cli, loi, OBD_BRW_READ,
1670                                               &loi->loi_read_lop);
1671                         if (rc < 0)
1672                                 break;
1673                         if (rc > 0)
1674                                 race_counter = 0;
1675                         else
1676                                 race_counter++;
1677                 }
1678
1679                 /* attempt some inter-object balancing by issueing rpcs
1680                  * for each object in turn */
1681                 if (!list_empty(&loi->loi_cli_item))
1682                         list_del_init(&loi->loi_cli_item);
1683                 if (!list_empty(&loi->loi_write_item))
1684                         list_del_init(&loi->loi_write_item);
1685                 if (!list_empty(&loi->loi_read_item))
1686                         list_del_init(&loi->loi_read_item);
1687
1688                 loi_list_maint(cli, loi);
1689
1690                 /* send_oap_rpc fails with 0 when make_ready tells it to
1691                  * back off.  llite's make_ready does this when it tries
1692                  * to lock a page queued for write that is already locked.
1693                  * we want to try sending rpcs from many objects, but we
1694                  * don't want to spin failing with 0.  */
1695                 if (race_counter == 10)
1696                         break;
1697         }
1698         EXIT;
1699 }
1700
1701 /* we're trying to queue a page in the osc so we're subject to the
1702  * 'cl_dirty_max' limit on the number of pages that can be queued in the osc.
1703  * If the osc's queued pages are already at that limit, then we want to sleep
1704  * until there is space in the osc's queue for us.  We also may be waiting for
1705  * write credits from the OST if there are RPCs in flight that may return some
1706  * before we fall back to sync writes.
1707  *
1708  * We need this know our allocation was granted in the presence of signals */
1709 static int ocw_granted(struct client_obd *cli, struct osc_cache_waiter *ocw)
1710 {
1711         int rc;
1712         ENTRY;
1713         spin_lock(&cli->cl_loi_list_lock);
1714         rc = list_empty(&ocw->ocw_entry) || cli->cl_brw_in_flight == 0;
1715         spin_unlock(&cli->cl_loi_list_lock);
1716         RETURN(rc);
1717 };
1718
1719 /* Caller must hold loi_list_lock - we drop/regain it if we need to wait for
1720  * grant or cache space. */
1721 static int osc_enter_cache(struct client_obd *cli, struct lov_oinfo *loi,
1722                            struct osc_async_page *oap)
1723 {
1724         struct osc_cache_waiter ocw;
1725         struct l_wait_info lwi = { 0 };
1726
1727         CDEBUG(D_CACHE, "dirty: %ld dirty_max: %ld dropped: %lu grant: %lu\n",
1728                cli->cl_dirty, cli->cl_dirty_max, cli->cl_lost_grant,
1729                cli->cl_avail_grant);
1730
1731         if (cli->cl_dirty_max < PAGE_SIZE)
1732                 return(-EDQUOT);
1733
1734         /* Hopefully normal case - cache space and write credits available */
1735         if (cli->cl_dirty + PAGE_SIZE <= cli->cl_dirty_max &&
1736             cli->cl_avail_grant >= PAGE_SIZE) {
1737                 /* account for ourselves */
1738                 osc_consume_write_grant(cli, oap);
1739                 return(0);
1740         }
1741
1742         /* Make sure that there are write rpcs in flight to wait for.  This
1743          * is a little silly as this object may not have any pending but
1744          * other objects sure might. */
1745         if (cli->cl_brw_in_flight) {
1746                 list_add_tail(&ocw.ocw_entry, &cli->cl_cache_waiters);
1747                 init_waitqueue_head(&ocw.ocw_waitq);
1748                 ocw.ocw_oap = oap;
1749                 ocw.ocw_rc = 0;
1750
1751                 loi_list_maint(cli, loi);
1752                 osc_check_rpcs(cli);
1753                 spin_unlock(&cli->cl_loi_list_lock);
1754
1755                 CDEBUG(0, "sleeping for cache space\n");
1756                 l_wait_event(ocw.ocw_waitq, ocw_granted(cli, &ocw), &lwi);
1757
1758                 spin_lock(&cli->cl_loi_list_lock);
1759                 if (!list_empty(&ocw.ocw_entry)) {
1760                         list_del(&ocw.ocw_entry);
1761                         RETURN(-EINTR);
1762                 }
1763                 RETURN(ocw.ocw_rc);
1764         }
1765
1766         RETURN(-EDQUOT);
1767 }
1768
1769 /* the companion to enter_cache, called when an oap is no longer part of the
1770  * dirty accounting.. so writeback completes or truncate happens before writing
1771  * starts.  must be called with the loi lock held. */
1772 static void osc_exit_cache(struct client_obd *cli, struct osc_async_page *oap,
1773                            int sent)
1774 {
1775         ENTRY;
1776
1777         if (!(oap->oap_brw_flags & OBD_BRW_FROM_GRANT)) {
1778                 EXIT;
1779                 return;
1780         }
1781
1782         oap->oap_brw_flags &= ~OBD_BRW_FROM_GRANT;
1783         cli->cl_dirty -= PAGE_SIZE;
1784         if (!sent) {
1785                 cli->cl_lost_grant += PAGE_SIZE;
1786                 CDEBUG(D_CACHE, "lost grant: %lu avail grant: %lu dirty: %lu\n",
1787                        cli->cl_lost_grant, cli->cl_avail_grant, cli->cl_dirty);
1788         }
1789
1790         EXIT;
1791 }
1792
1793 int osc_prep_async_page(struct obd_export *exp, struct lov_stripe_md *lsm,
1794                         struct lov_oinfo *loi, struct page *page,
1795                         obd_off offset, struct obd_async_page_ops *ops,
1796                         void *data, void **res)
1797 {
1798         struct osc_async_page *oap;
1799         ENTRY;
1800
1801         OBD_ALLOC(oap, sizeof(*oap));
1802         if (oap == NULL)
1803                 return -ENOMEM;
1804
1805         oap->oap_magic = OAP_MAGIC;
1806         oap->oap_cli = &exp->exp_obd->u.cli;
1807         oap->oap_loi = loi;
1808
1809         oap->oap_caller_ops = ops;
1810         oap->oap_caller_data = data;
1811
1812         oap->oap_page = page;
1813         oap->oap_obj_off = offset;
1814
1815         INIT_LIST_HEAD(&oap->oap_pending_item);
1816         INIT_LIST_HEAD(&oap->oap_urgent_item);
1817         INIT_LIST_HEAD(&oap->oap_rpc_item);
1818
1819         oap->oap_occ.occ_interrupted = osc_occ_interrupted;
1820
1821         CDEBUG(D_CACHE, "oap %p page %p obj off "LPU64"\n", oap, page, offset);
1822         *res = oap;
1823         RETURN(0);
1824 }
1825
1826 struct osc_async_page *oap_from_cookie(void *cookie)
1827 {
1828         struct osc_async_page *oap = cookie;
1829         if (oap->oap_magic != OAP_MAGIC)
1830                 return ERR_PTR(-EINVAL);
1831         return oap;
1832 };
1833
1834 static int osc_queue_async_io(struct obd_export *exp, struct lov_stripe_md *lsm,
1835                               struct lov_oinfo *loi, void *cookie,
1836                               int cmd, obd_off off, int count,
1837                               obd_flag brw_flags, enum async_flags async_flags)
1838 {
1839         struct client_obd *cli = &exp->exp_obd->u.cli;
1840         struct osc_async_page *oap;
1841         struct loi_oap_pages *lop;
1842         int rc;
1843         ENTRY;
1844
1845         oap = oap_from_cookie(cookie);
1846         if (IS_ERR(oap))
1847                 RETURN(PTR_ERR(oap));
1848
1849         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
1850                 RETURN(-EIO);
1851
1852         if (!list_empty(&oap->oap_pending_item) ||
1853             !list_empty(&oap->oap_urgent_item) ||
1854             !list_empty(&oap->oap_rpc_item))
1855                 RETURN(-EBUSY);
1856
1857         if (loi == NULL)
1858                 loi = &lsm->lsm_oinfo[0];
1859
1860         spin_lock(&cli->cl_loi_list_lock);
1861
1862         oap->oap_cmd = cmd;
1863         oap->oap_async_flags = async_flags;
1864         oap->oap_page_off = off;
1865         oap->oap_count = count;
1866         oap->oap_brw_flags = brw_flags;
1867
1868         if (cmd == OBD_BRW_WRITE) {
1869                 rc = osc_enter_cache(cli, loi, oap);
1870                 if (rc) {
1871                         spin_unlock(&cli->cl_loi_list_lock);
1872                         RETURN(rc);
1873                 }
1874                 lop = &loi->loi_write_lop;
1875         } else {
1876                 lop = &loi->loi_read_lop;
1877         }
1878
1879         if (oap->oap_async_flags & ASYNC_URGENT)
1880                 list_add(&oap->oap_urgent_item, &lop->lop_urgent);
1881         list_add_tail(&oap->oap_pending_item, &lop->lop_pending);
1882         lop_update_pending(cli, lop, cmd, 1);
1883
1884         loi_list_maint(cli, loi);
1885
1886         LOI_DEBUG(loi, "oap %p page %p added for cmd %d\n", oap, oap->oap_page,
1887                   cmd);
1888
1889         osc_check_rpcs(cli);
1890         spin_unlock(&cli->cl_loi_list_lock);
1891
1892         RETURN(0);
1893 }
1894
1895 /* aka (~was & now & flag), but this is more clear :) */
1896 #define SETTING(was, now, flag) (!(was & flag) && (now & flag))
1897
1898 static int osc_set_async_flags(struct obd_export *exp,
1899                                struct lov_stripe_md *lsm,
1900                                struct lov_oinfo *loi, void *cookie,
1901                                obd_flag async_flags)
1902 {
1903         struct client_obd *cli = &exp->exp_obd->u.cli;
1904         struct loi_oap_pages *lop;
1905         struct osc_async_page *oap;
1906         int rc = 0;
1907         ENTRY;
1908
1909         oap = oap_from_cookie(cookie);
1910         if (IS_ERR(oap))
1911                 RETURN(PTR_ERR(oap));
1912
1913         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
1914                 RETURN(-EIO);
1915
1916         if (loi == NULL)
1917                 loi = &lsm->lsm_oinfo[0];
1918
1919         if (oap->oap_cmd == OBD_BRW_WRITE) {
1920                 lop = &loi->loi_write_lop;
1921         } else {
1922                 lop = &loi->loi_read_lop;
1923         }
1924
1925         spin_lock(&cli->cl_loi_list_lock);
1926
1927         if (list_empty(&oap->oap_pending_item))
1928                 GOTO(out, rc = -EINVAL);
1929
1930         if ((oap->oap_async_flags & async_flags) == async_flags)
1931                 GOTO(out, rc = 0);
1932
1933         if (SETTING(oap->oap_async_flags, async_flags, ASYNC_READY))
1934                 oap->oap_async_flags |= ASYNC_READY;
1935
1936         if (SETTING(oap->oap_async_flags, async_flags, ASYNC_URGENT)) {
1937                 if (list_empty(&oap->oap_rpc_item)) {
1938                         list_add(&oap->oap_urgent_item, &lop->lop_urgent);
1939                         loi_list_maint(cli, loi);
1940                 }
1941         }
1942
1943         LOI_DEBUG(loi, "oap %p page %p has flags %x\n", oap, oap->oap_page,
1944                         oap->oap_async_flags);
1945 out:
1946         osc_check_rpcs(cli);
1947         spin_unlock(&cli->cl_loi_list_lock);
1948         RETURN(rc);
1949 }
1950
1951 static int osc_queue_group_io(struct obd_export *exp, struct lov_stripe_md *lsm,
1952                              struct lov_oinfo *loi,
1953                              struct obd_io_group *oig, void *cookie,
1954                              int cmd, obd_off off, int count,
1955                              obd_flag brw_flags,
1956                              obd_flag async_flags)
1957 {
1958         struct client_obd *cli = &exp->exp_obd->u.cli;
1959         struct osc_async_page *oap;
1960         struct loi_oap_pages *lop;
1961         ENTRY;
1962
1963         oap = oap_from_cookie(cookie);
1964         if (IS_ERR(oap))
1965                 RETURN(PTR_ERR(oap));
1966
1967         if (cli->cl_import == NULL || cli->cl_import->imp_invalid)
1968                 RETURN(-EIO);
1969
1970         if (!list_empty(&oap->oap_pending_item) ||
1971             !list_empty(&oap->oap_urgent_item) ||
1972             !list_empty(&oap->oap_rpc_item))
1973                 RETURN(-EBUSY);
1974
1975         if (loi == NULL)
1976                 loi = &lsm->lsm_oinfo[0];
1977
1978         spin_lock(&cli->cl_loi_list_lock);
1979
1980         oap->oap_cmd = cmd;
1981         oap->oap_page_off = off;
1982         oap->oap_count = count;
1983         oap->oap_brw_flags = brw_flags;
1984         oap->oap_async_flags = async_flags;
1985
1986         if (cmd == OBD_BRW_WRITE)
1987                 lop = &loi->loi_write_lop;
1988         else
1989                 lop = &loi->loi_read_lop;
1990
1991         list_add_tail(&oap->oap_pending_item, &lop->lop_pending_group);
1992         if (oap->oap_async_flags & ASYNC_GROUP_SYNC) {
1993                 oap->oap_oig = oig;
1994                 oig_add_one(oig, &oap->oap_occ);
1995         }
1996
1997         LOI_DEBUG(loi, "oap %p page %p on group pending\n", oap, oap->oap_page);
1998
1999         spin_unlock(&cli->cl_loi_list_lock);
2000
2001         RETURN(0);
2002 }
2003
2004 static void osc_group_to_pending(struct client_obd *cli, struct lov_oinfo *loi,
2005                                  struct loi_oap_pages *lop, int cmd)
2006 {
2007         struct list_head *pos, *tmp;
2008         struct osc_async_page *oap;
2009
2010         list_for_each_safe(pos, tmp, &lop->lop_pending_group) {
2011                 oap = list_entry(pos, struct osc_async_page, oap_pending_item);
2012                 list_del(&oap->oap_pending_item);
2013                 list_add_tail(&oap->oap_pending_item, &lop->lop_pending);
2014                 list_add(&oap->oap_urgent_item, &lop->lop_urgent);
2015                 lop_update_pending(cli, lop, cmd, 1);
2016         }
2017         loi_list_maint(cli, loi);
2018 }
2019
2020 static int osc_trigger_group_io(struct obd_export *exp,
2021                                 struct lov_stripe_md *lsm,
2022                                 struct lov_oinfo *loi,
2023                                 struct obd_io_group *oig)
2024 {
2025         struct client_obd *cli = &exp->exp_obd->u.cli;
2026         ENTRY;
2027
2028         if (loi == NULL)
2029                 loi = &lsm->lsm_oinfo[0];
2030
2031         spin_lock(&cli->cl_loi_list_lock);
2032
2033         osc_group_to_pending(cli, loi, &loi->loi_write_lop, OBD_BRW_WRITE);
2034         osc_group_to_pending(cli, loi, &loi->loi_read_lop, OBD_BRW_READ);
2035
2036         osc_check_rpcs(cli);
2037         spin_unlock(&cli->cl_loi_list_lock);
2038
2039         RETURN(0);
2040 }
2041
2042 static int osc_teardown_async_page(struct obd_export *exp,
2043                                    struct lov_stripe_md *lsm,
2044                                    struct lov_oinfo *loi, void *cookie)
2045 {
2046         struct client_obd *cli = &exp->exp_obd->u.cli;
2047         struct loi_oap_pages *lop;
2048         struct osc_async_page *oap;
2049         int rc = 0;
2050         ENTRY;
2051
2052         oap = oap_from_cookie(cookie);
2053         if (IS_ERR(oap))
2054                 RETURN(PTR_ERR(oap));
2055
2056         if (loi == NULL)
2057                 loi = &lsm->lsm_oinfo[0];
2058
2059         if (oap->oap_cmd == OBD_BRW_WRITE) {
2060                 lop = &loi->loi_write_lop;
2061         } else {
2062                 lop = &loi->loi_read_lop;
2063         }
2064
2065         spin_lock(&cli->cl_loi_list_lock);
2066
2067         if (!list_empty(&oap->oap_rpc_item))
2068                 GOTO(out, rc = -EBUSY);
2069
2070         osc_exit_cache(cli, oap, 0);
2071         osc_wake_cache_waiters(cli);
2072
2073         if (!list_empty(&oap->oap_urgent_item)) {
2074                 list_del_init(&oap->oap_urgent_item);
2075                 oap->oap_async_flags &= ~ASYNC_URGENT;
2076         }
2077         if (!list_empty(&oap->oap_pending_item)) {
2078                 list_del_init(&oap->oap_pending_item);
2079                 lop_update_pending(cli, lop, oap->oap_cmd, -1);
2080         }
2081         loi_list_maint(cli, loi);
2082
2083         LOI_DEBUG(loi, "oap %p page %p torn down\n", oap, oap->oap_page);
2084 out:
2085         spin_unlock(&cli->cl_loi_list_lock);
2086         if (rc == 0)
2087                 OBD_FREE(oap, sizeof(*oap));
2088         RETURN(rc);
2089 }
2090
2091 #ifdef __KERNEL__
2092 /* Note: caller will lock/unlock, and set uptodate on the pages */
2093 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
2094 static int sanosc_brw_read(struct obd_export *exp, struct obdo *oa,
2095                            struct lov_stripe_md *lsm, obd_count page_count,
2096                            struct brw_page *pga)
2097 {
2098         struct ptlrpc_request *request = NULL;
2099         struct ost_body *body;
2100         struct niobuf_remote *nioptr;
2101         struct obd_ioobj *iooptr;
2102         int rc, size[3] = {sizeof(*body)}, mapped = 0;
2103         int swab;
2104         ENTRY;
2105
2106         /* XXX does not handle 'new' brw protocol */
2107
2108         size[1] = sizeof(struct obd_ioobj);
2109         size[2] = page_count * sizeof(*nioptr);
2110
2111         request = ptlrpc_prep_req(class_exp2cliimp(exp), OST_SAN_READ, 3,
2112                                   size, NULL);
2113         if (!request)
2114                 RETURN(-ENOMEM);
2115
2116         body = lustre_msg_buf(request->rq_reqmsg, 0, sizeof(*body));
2117         iooptr = lustre_msg_buf(request->rq_reqmsg, 1, sizeof(*iooptr));
2118         nioptr = lustre_msg_buf(request->rq_reqmsg, 2,
2119                                 sizeof(*nioptr) * page_count);
2120
2121         memcpy(&body->oa, oa, sizeof(body->oa));
2122
2123         obdo_to_ioobj(oa, iooptr);
2124         iooptr->ioo_bufcnt = page_count;
2125
2126         for (mapped = 0; mapped < page_count; mapped++, nioptr++) {
2127                 LASSERT(PageLocked(pga[mapped].pg));
2128                 LASSERT(mapped == 0 || pga[mapped].off > pga[mapped - 1].off);
2129
2130                 nioptr->offset = pga[mapped].off;
2131                 nioptr->len    = pga[mapped].count;
2132                 nioptr->flags  = pga[mapped].flag;
2133         }
2134
2135         size[1] = page_count * sizeof(*nioptr);
2136         request->rq_replen = lustre_msg_size(2, size);
2137
2138         rc = ptlrpc_queue_wait(request);
2139         if (rc)
2140                 GOTO(out_req, rc);
2141
2142         body = lustre_swab_repbuf(request, 0, sizeof(*body),
2143                                   lustre_swab_ost_body);
2144         if (body == NULL) {
2145                 CERROR("Can't unpack body\n");
2146                 GOTO(out_req, rc = -EPROTO);
2147         }
2148
2149         memcpy(oa, &body->oa, sizeof(*oa));
2150
2151         swab = lustre_msg_swabbed(request->rq_repmsg);
2152         LASSERT_REPSWAB(request, 1);
2153         nioptr = lustre_msg_buf(request->rq_repmsg, 1, size[1]);
2154         if (!nioptr) {
2155                 /* nioptr missing or short */
2156                 GOTO(out_req, rc = -EPROTO);
2157         }
2158
2159         /* actual read */
2160         for (mapped = 0; mapped < page_count; mapped++, nioptr++) {
2161                 struct page *page = pga[mapped].pg;
2162                 struct buffer_head *bh;
2163                 kdev_t dev;
2164
2165                 if (swab)
2166                         lustre_swab_niobuf_remote (nioptr);
2167
2168                 /* got san device associated */
2169                 LASSERT(exp->exp_obd != NULL);
2170                 dev = exp->exp_obd->u.cli.cl_sandev;
2171
2172                 /* hole */
2173                 if (!nioptr->offset) {
2174                         CDEBUG(D_PAGE, "hole at ino %lu; index %ld\n",
2175                                         page->mapping->host->i_ino,
2176                                         page->index);
2177                         memset(page_address(page), 0, PAGE_SIZE);
2178                         continue;
2179                 }
2180
2181                 if (!page->buffers) {
2182                         create_empty_buffers(page, dev, PAGE_SIZE);
2183                         bh = page->buffers;
2184
2185                         clear_bit(BH_New, &bh->b_state);
2186                         set_bit(BH_Mapped, &bh->b_state);
2187                         bh->b_blocknr = (unsigned long)nioptr->offset;
2188
2189                         clear_bit(BH_Uptodate, &bh->b_state);
2190
2191                         ll_rw_block(READ, 1, &bh);
2192                 } else {
2193                         bh = page->buffers;
2194
2195                         /* if buffer already existed, it must be the
2196                          * one we mapped before, check it */
2197                         LASSERT(!test_bit(BH_New, &bh->b_state));
2198                         LASSERT(test_bit(BH_Mapped, &bh->b_state));
2199                         LASSERT(bh->b_blocknr == (unsigned long)nioptr->offset);
2200
2201                         /* wait it's io completion */
2202                         if (test_bit(BH_Lock, &bh->b_state))
2203                                 wait_on_buffer(bh);
2204
2205                         if (!test_bit(BH_Uptodate, &bh->b_state))
2206                                 ll_rw_block(READ, 1, &bh);
2207                 }
2208
2209
2210                 /* must do syncronous write here */
2211                 wait_on_buffer(bh);
2212                 if (!buffer_uptodate(bh)) {
2213                         /* I/O error */
2214                         rc = -EIO;
2215                         goto out_req;
2216                 }
2217         }
2218
2219 out_req:
2220         ptlrpc_req_finished(request);
2221         RETURN(rc);
2222 }
2223
2224 static int sanosc_brw_write(struct obd_export *exp, struct obdo *oa,
2225                             struct lov_stripe_md *lsm, obd_count page_count,
2226                             struct brw_page *pga)
2227 {
2228         struct ptlrpc_request *request = NULL;
2229         struct ost_body *body;
2230         struct niobuf_remote *nioptr;
2231         struct obd_ioobj *iooptr;
2232         int rc, size[3] = {sizeof(*body)}, mapped = 0;
2233         int swab;
2234         ENTRY;
2235
2236         size[1] = sizeof(struct obd_ioobj);
2237         size[2] = page_count * sizeof(*nioptr);
2238
2239         request = ptlrpc_prep_req(class_exp2cliimp(exp), OST_SAN_WRITE,
2240                                   3, size, NULL);
2241         if (!request)
2242                 RETURN(-ENOMEM);
2243
2244         body = lustre_msg_buf(request->rq_reqmsg, 0, sizeof (*body));
2245         iooptr = lustre_msg_buf(request->rq_reqmsg, 1, sizeof (*iooptr));
2246         nioptr = lustre_msg_buf(request->rq_reqmsg, 2,
2247                                 sizeof (*nioptr) * page_count);
2248
2249         memcpy(&body->oa, oa, sizeof(body->oa));
2250
2251         obdo_to_ioobj(oa, iooptr);
2252         iooptr->ioo_bufcnt = page_count;
2253
2254         /* pack request */
2255         for (mapped = 0; mapped < page_count; mapped++, nioptr++) {
2256                 LASSERT(PageLocked(pga[mapped].pg));
2257                 LASSERT(mapped == 0 || pga[mapped].off > pga[mapped - 1].off);
2258
2259                 nioptr->offset = pga[mapped].off;
2260                 nioptr->len    = pga[mapped].count;
2261                 nioptr->flags  = pga[mapped].flag;
2262         }
2263
2264         size[1] = page_count * sizeof(*nioptr);
2265         request->rq_replen = lustre_msg_size(2, size);
2266
2267         rc = ptlrpc_queue_wait(request);
2268         if (rc)
2269                 GOTO(out_req, rc);
2270
2271         swab = lustre_msg_swabbed (request->rq_repmsg);
2272         LASSERT_REPSWAB (request, 1);
2273         nioptr = lustre_msg_buf(request->rq_repmsg, 1, size[1]);
2274         if (!nioptr) {
2275                 CERROR("absent/short niobuf array\n");
2276                 GOTO(out_req, rc = -EPROTO);
2277         }
2278
2279         /* actual write */
2280         for (mapped = 0; mapped < page_count; mapped++, nioptr++) {
2281                 struct page *page = pga[mapped].pg;
2282                 struct buffer_head *bh;
2283                 kdev_t dev;
2284
2285                 if (swab)
2286                         lustre_swab_niobuf_remote (nioptr);
2287
2288                 /* got san device associated */
2289                 LASSERT(exp->exp_obd != NULL);
2290                 dev = exp->exp_obd->u.cli.cl_sandev;
2291
2292                 if (!page->buffers) {
2293                         create_empty_buffers(page, dev, PAGE_SIZE);
2294                 } else {
2295                         /* checking */
2296                         LASSERT(!test_bit(BH_New, &page->buffers->b_state));
2297                         LASSERT(test_bit(BH_Mapped, &page->buffers->b_state));
2298                         LASSERT(page->buffers->b_blocknr ==
2299                                 (unsigned long)nioptr->offset);
2300                 }
2301                 bh = page->buffers;
2302
2303                 LASSERT(bh);
2304
2305                 /* if buffer locked, wait it's io completion */
2306                 if (test_bit(BH_Lock, &bh->b_state))
2307                         wait_on_buffer(bh);
2308
2309                 clear_bit(BH_New, &bh->b_state);
2310                 set_bit(BH_Mapped, &bh->b_state);
2311
2312                 /* override the block nr */
2313                 bh->b_blocknr = (unsigned long)nioptr->offset;
2314
2315                 /* we are about to write it, so set it
2316                  * uptodate/dirty
2317                  * page lock should garentee no race condition here */
2318                 set_bit(BH_Uptodate, &bh->b_state);
2319                 set_bit(BH_Dirty, &bh->b_state);
2320
2321                 ll_rw_block(WRITE, 1, &bh);
2322
2323                 /* must do syncronous write here */
2324                 wait_on_buffer(bh);
2325                 if (!buffer_uptodate(bh) || test_bit(BH_Dirty, &bh->b_state)) {
2326                         /* I/O error */
2327                         rc = -EIO;
2328                         goto out_req;
2329                 }
2330         }
2331
2332 out_req:
2333         ptlrpc_req_finished(request);
2334         RETURN(rc);
2335 }
2336
2337 static int sanosc_brw(int cmd, struct obd_export *exp, struct obdo *oa,
2338                       struct lov_stripe_md *lsm, obd_count page_count,
2339                       struct brw_page *pga, struct obd_trans_info *oti)
2340 {
2341         ENTRY;
2342
2343         while (page_count) {
2344                 obd_count pages_per_brw;
2345                 int rc;
2346
2347                 if (page_count > PTLRPC_MAX_BRW_PAGES)
2348                         pages_per_brw = PTLRPC_MAX_BRW_PAGES;
2349                 else
2350                         pages_per_brw = page_count;
2351
2352                 if (cmd & OBD_BRW_WRITE)
2353                         rc = sanosc_brw_write(exp, oa, lsm, pages_per_brw,pga);
2354                 else
2355                         rc = sanosc_brw_read(exp, oa, lsm, pages_per_brw, pga);
2356
2357                 if (rc != 0)
2358                         RETURN(rc);
2359
2360                 page_count -= pages_per_brw;
2361                 pga += pages_per_brw;
2362         }
2363         RETURN(0);
2364 }
2365 #endif
2366 #endif
2367
2368 static void osc_set_data_with_check(struct lustre_handle *lockh, void *data)
2369 {
2370         struct ldlm_lock *lock = ldlm_handle2lock(lockh);
2371
2372         LASSERT(lock != NULL);
2373         l_lock(&lock->l_resource->lr_namespace->ns_lock);
2374 #ifdef __KERNEL__
2375         if (lock->l_ast_data && lock->l_ast_data != data) {
2376                 struct inode *new_inode = data;
2377                 struct inode *old_inode = lock->l_ast_data;
2378                 LASSERTF(old_inode->i_state & I_FREEING,
2379                          "Found existing inode %p/%lu/%u state %lu in lock: "
2380                          "setting data to %p/%lu/%u\n", old_inode,
2381                          old_inode->i_ino, old_inode->i_generation,
2382                          old_inode->i_state,
2383                          new_inode, new_inode->i_ino, new_inode->i_generation);
2384         }
2385 #endif
2386         lock->l_ast_data = data;
2387         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
2388         LDLM_LOCK_PUT(lock);
2389 }
2390
2391 static int osc_change_cbdata(struct obd_export *exp, struct lov_stripe_md *lsm,
2392                              ldlm_iterator_t replace, void *data)
2393 {
2394         struct ldlm_res_id res_id = { .name = {lsm->lsm_object_id} };
2395         struct obd_device *obd = class_exp2obd(exp);
2396
2397         ldlm_change_cbdata(obd->obd_namespace, &res_id, replace, data);
2398         return 0;
2399 }
2400
2401 static int osc_enqueue(struct obd_export *exp, struct lov_stripe_md *lsm,
2402                        __u32 type, ldlm_policy_data_t *policy, __u32 mode,
2403                        int *flags, void *bl_cb, void *cp_cb, void *gl_cb,
2404                        void *data, __u32 lvb_len, void *lvb_swabber,
2405                        struct lustre_handle *lockh)
2406 {
2407         struct ldlm_res_id res_id = { .name = {lsm->lsm_object_id} };
2408         struct obd_device *obd = exp->exp_obd;
2409         struct ost_lvb lvb;
2410         int rc;
2411         ENTRY;
2412
2413         /* Filesystem lock extents are extended to page boundaries so that
2414          * dealing with the page cache is a little smoother.  */
2415         policy->l_extent.start -= policy->l_extent.start & ~PAGE_MASK;
2416         policy->l_extent.end |= ~PAGE_MASK;
2417
2418         if (lsm->lsm_oinfo->loi_kms_valid == 0)
2419                 goto no_match;
2420
2421         /* Next, search for already existing extent locks that will cover us */
2422         rc = ldlm_lock_match(obd->obd_namespace, 0, &res_id, type, policy, mode,
2423                              lockh);
2424         if (rc == 1) {
2425                 osc_set_data_with_check(lockh, data);
2426                 if (*flags & LDLM_FL_HAS_INTENT) {
2427                         /* I would like to be able to ASSERT here that rss <=
2428                          * kms, but I can't, for reasons which are explained in
2429                          * lov_enqueue() */
2430                 }
2431                 /* We already have a lock, and it's referenced */
2432                 RETURN(ELDLM_OK);
2433         }
2434
2435         /* If we're trying to read, we also search for an existing PW lock.  The
2436          * VFS and page cache already protect us locally, so lots of readers/
2437          * writers can share a single PW lock.
2438          *
2439          * There are problems with conversion deadlocks, so instead of
2440          * converting a read lock to a write lock, we'll just enqueue a new
2441          * one.
2442          *
2443          * At some point we should cancel the read lock instead of making them
2444          * send us a blocking callback, but there are problems with canceling
2445          * locks out from other users right now, too. */
2446
2447         if (mode == LCK_PR) {
2448                 rc = ldlm_lock_match(obd->obd_namespace, 0, &res_id, type,
2449                                      policy, LCK_PW, lockh);
2450                 if (rc == 1) {
2451                         /* FIXME: This is not incredibly elegant, but it might
2452                          * be more elegant than adding another parameter to
2453                          * lock_match.  I want a second opinion. */
2454                         ldlm_lock_addref(lockh, LCK_PR);
2455                         ldlm_lock_decref(lockh, LCK_PW);
2456                         osc_set_data_with_check(lockh, data);
2457                         RETURN(ELDLM_OK);
2458                 }
2459         }
2460
2461  no_match:
2462         rc = ldlm_cli_enqueue(exp, NULL, obd->obd_namespace, res_id, type,
2463                               policy, mode, flags, bl_cb, cp_cb, gl_cb, data,
2464                               &lvb, sizeof(lvb), lustre_swab_ost_lvb, lockh);
2465
2466         if ((*flags & LDLM_FL_HAS_INTENT && rc == ELDLM_LOCK_ABORTED) || !rc) {
2467                 CDEBUG(D_INODE, "received kms == "LPU64", blocks == "LPU64"\n",
2468                        lvb.lvb_size, lvb.lvb_blocks);
2469                 lsm->lsm_oinfo->loi_rss = lvb.lvb_size;
2470                 lsm->lsm_oinfo->loi_blocks = lvb.lvb_blocks;
2471         }
2472
2473         RETURN(rc);
2474 }
2475
2476 static int osc_match(struct obd_export *exp, struct lov_stripe_md *lsm,
2477                      __u32 type, ldlm_policy_data_t *policy, __u32 mode,
2478                      int *flags, void *data, struct lustre_handle *lockh)
2479 {
2480         struct ldlm_res_id res_id = { .name = {lsm->lsm_object_id} };
2481         struct obd_device *obd = exp->exp_obd;
2482         int rc;
2483         ENTRY;
2484
2485         OBD_FAIL_RETURN(OBD_FAIL_OSC_MATCH, -EIO);
2486
2487         /* Filesystem lock extents are extended to page boundaries so that
2488          * dealing with the page cache is a little smoother */
2489         policy->l_extent.start -= policy->l_extent.start & ~PAGE_MASK;
2490         policy->l_extent.end |= ~PAGE_MASK;
2491
2492         /* Next, search for already existing extent locks that will cover us */
2493         rc = ldlm_lock_match(obd->obd_namespace, *flags, &res_id, type,
2494                              policy, mode, lockh);
2495         if (rc) {
2496                 osc_set_data_with_check(lockh, data);
2497                 RETURN(rc);
2498         }
2499         /* If we're trying to read, we also search for an existing PW lock.  The
2500          * VFS and page cache already protect us locally, so lots of readers/
2501          * writers can share a single PW lock. */
2502         if (mode == LCK_PR) {
2503                 rc = ldlm_lock_match(obd->obd_namespace, *flags, &res_id, type,
2504                                      policy, LCK_PW, lockh);
2505                 if (rc == 1 && !(*flags & LDLM_FL_TEST_LOCK)) {
2506                         /* FIXME: This is not incredibly elegant, but it might
2507                          * be more elegant than adding another parameter to
2508                          * lock_match.  I want a second opinion. */
2509                         osc_set_data_with_check(lockh, data);
2510                         ldlm_lock_addref(lockh, LCK_PR);
2511                         ldlm_lock_decref(lockh, LCK_PW);
2512                 }
2513         }
2514         RETURN(rc);
2515 }
2516
2517 static int osc_cancel(struct obd_export *exp, struct lov_stripe_md *md,
2518                       __u32 mode, struct lustre_handle *lockh)
2519 {
2520         ENTRY;
2521
2522         if (mode == LCK_GROUP)
2523                 ldlm_lock_decref_and_cancel(lockh, mode);
2524         else
2525                 ldlm_lock_decref(lockh, mode);
2526
2527         RETURN(0);
2528 }
2529
2530 static int osc_cancel_unused(struct obd_export *exp,
2531                              struct lov_stripe_md *lsm, int flags, void *opaque)
2532 {
2533         struct obd_device *obd = class_exp2obd(exp);
2534         struct ldlm_res_id res_id = { .name = {lsm->lsm_object_id} };
2535
2536         return ldlm_cli_cancel_unused(obd->obd_namespace, &res_id, flags,
2537                                       opaque);
2538 }
2539
2540 static int osc_statfs(struct obd_device *obd, struct obd_statfs *osfs,
2541                       unsigned long max_age)
2542 {
2543         struct obd_statfs *msfs;
2544         struct ptlrpc_request *request;
2545         int rc, size = sizeof(*osfs);
2546         ENTRY;
2547
2548         /* We could possibly pass max_age in the request (as an absolute
2549          * timestamp or a "seconds.usec ago") so the target can avoid doing
2550          * extra calls into the filesystem if that isn't necessary (e.g.
2551          * during mount that would help a bit).  Having relative timestamps
2552          * is not so great if request processing is slow, while absolute
2553          * timestamps are not ideal because they need time synchronization. */
2554         request = ptlrpc_prep_req(obd->u.cli.cl_import, OST_STATFS,0,NULL,NULL);
2555         if (!request)
2556                 RETURN(-ENOMEM);
2557
2558         request->rq_replen = lustre_msg_size(1, &size);
2559         request->rq_request_portal = OST_CREATE_PORTAL; //XXX FIXME bug 249
2560
2561         rc = ptlrpc_queue_wait(request);
2562         if (rc)
2563                 GOTO(out, rc);
2564
2565         msfs = lustre_swab_repbuf(request, 0, sizeof(*msfs),
2566                                   lustre_swab_obd_statfs);
2567         if (msfs == NULL) {
2568                 CERROR("Can't unpack obd_statfs\n");
2569                 GOTO(out, rc = -EPROTO);
2570         }
2571
2572         memcpy(osfs, msfs, sizeof(*osfs));
2573
2574         EXIT;
2575  out:
2576         ptlrpc_req_finished(request);
2577         return rc;
2578 }
2579
2580 /* Retrieve object striping information.
2581  *
2582  * @lmmu is a pointer to an in-core struct with lmm_ost_count indicating
2583  * the maximum number of OST indices which will fit in the user buffer.
2584  * lmm_magic must be LOV_MAGIC (we only use 1 slot here).
2585  */
2586 static int osc_getstripe(struct lov_stripe_md *lsm, struct lov_user_md *lump)
2587 {
2588         struct lov_user_md lum, *lumk;
2589         int rc, lum_size;
2590         ENTRY;
2591
2592         if (!lsm)
2593                 RETURN(-ENODATA);
2594
2595         rc = copy_from_user(&lum, lump, sizeof(lum));
2596         if (rc)
2597                 RETURN(-EFAULT);
2598
2599         if (lum.lmm_magic != LOV_USER_MAGIC)
2600                 RETURN(-EINVAL);
2601
2602         if (lum.lmm_stripe_count > 0) {
2603                 lum_size = sizeof(lum) + sizeof(lum.lmm_objects[0]);
2604                 OBD_ALLOC(lumk, lum_size);
2605                 if (!lumk)
2606                         RETURN(-ENOMEM);
2607
2608                 lumk->lmm_objects[0].l_object_id = lsm->lsm_object_id;
2609         } else {
2610                 lum_size = sizeof(lum);
2611                 lumk = &lum;
2612         }
2613
2614         lumk->lmm_object_id = lsm->lsm_object_id;
2615         lumk->lmm_stripe_count = 1;
2616
2617         if (copy_to_user(lump, lumk, lum_size))
2618                 rc = -EFAULT;
2619
2620         if (lumk != &lum)
2621                 OBD_FREE(lumk, lum_size);
2622
2623         RETURN(rc);
2624 }
2625
2626 static int osc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
2627                          void *karg, void *uarg)
2628 {
2629         struct obd_device *obd = exp->exp_obd;
2630         struct obd_ioctl_data *data = karg;
2631         int err = 0;
2632         ENTRY;
2633
2634         MOD_INC_USE_COUNT;
2635
2636         switch (cmd) {
2637         case OBD_IOC_LOV_GET_CONFIG: {
2638                 char *buf;
2639                 struct lov_desc *desc;
2640                 struct obd_uuid uuid;
2641
2642                 buf = NULL;
2643                 len = 0;
2644                 if (obd_ioctl_getdata(&buf, &len, (void *)uarg))
2645                         GOTO(out, err = -EINVAL);
2646
2647                 data = (struct obd_ioctl_data *)buf;
2648
2649                 if (sizeof(*desc) > data->ioc_inllen1) {
2650                         OBD_FREE(buf, len);
2651                         GOTO(out, err = -EINVAL);
2652                 }
2653
2654                 if (data->ioc_inllen2 < sizeof(uuid)) {
2655                         OBD_FREE(buf, len);
2656                         GOTO(out, err = -EINVAL);
2657                 }
2658
2659                 desc = (struct lov_desc *)data->ioc_inlbuf1;
2660                 desc->ld_tgt_count = 1;
2661                 desc->ld_active_tgt_count = 1;
2662                 desc->ld_default_stripe_count = 1;
2663                 desc->ld_default_stripe_size = 0;
2664                 desc->ld_default_stripe_offset = 0;
2665                 desc->ld_pattern = 0;
2666                 memcpy(&desc->ld_uuid, &obd->obd_uuid, sizeof(uuid));
2667
2668                 memcpy(data->ioc_inlbuf2, &obd->obd_uuid, sizeof(uuid));
2669
2670                 err = copy_to_user((void *)uarg, buf, len);
2671                 if (err)
2672                         err = -EFAULT;
2673                 obd_ioctl_freedata(buf, len);
2674                 GOTO(out, err);
2675         }
2676         case LL_IOC_LOV_SETSTRIPE:
2677                 err = obd_alloc_memmd(exp, karg);
2678                 if (err > 0)
2679                         err = 0;
2680                 GOTO(out, err);
2681         case LL_IOC_LOV_GETSTRIPE:
2682                 err = osc_getstripe(karg, uarg);
2683                 GOTO(out, err);
2684         case OBD_IOC_CLIENT_RECOVER:
2685                 err = ptlrpc_recover_import(obd->u.cli.cl_import,
2686                                             data->ioc_inlbuf1);
2687                 if (err > 0)
2688                         err = 0;
2689                 GOTO(out, err);
2690         case IOC_OSC_SET_ACTIVE:
2691                 err = ptlrpc_set_import_active(obd->u.cli.cl_import,
2692                                                data->ioc_offset);
2693                 GOTO(out, err);
2694         default:
2695                 CDEBUG(D_INODE, "unrecognised ioctl %#x by %s\n", cmd, current->comm);
2696                 GOTO(out, err = -ENOTTY);
2697         }
2698 out:
2699         MOD_DEC_USE_COUNT;
2700         return err;
2701 }
2702
2703 static int osc_get_info(struct obd_export *exp, obd_count keylen,
2704                         void *key, __u32 *vallen, void *val)
2705 {
2706         ENTRY;
2707         if (!vallen || !val)
2708                 RETURN(-EFAULT);
2709
2710         if (keylen > strlen("lock_to_stripe") &&
2711             strcmp(key, "lock_to_stripe") == 0) {
2712                 __u32 *stripe = val;
2713                 *vallen = sizeof(*stripe);
2714                 *stripe = 0;
2715                 RETURN(0);
2716         } else if (keylen >= strlen("last_id") && strcmp(key, "last_id") == 0) {
2717                 struct ptlrpc_request *req;
2718                 obd_id *reply;
2719                 char *bufs[1] = {key};
2720                 int rc;
2721                 req = ptlrpc_prep_req(class_exp2cliimp(exp), OST_GET_INFO, 1,
2722                                       &keylen, bufs);
2723                 if (req == NULL)
2724                         RETURN(-ENOMEM);
2725
2726                 req->rq_replen = lustre_msg_size(1, vallen);
2727                 rc = ptlrpc_queue_wait(req);
2728                 if (rc)
2729                         GOTO(out, rc);
2730
2731                 reply = lustre_swab_repbuf(req, 0, sizeof(*reply),
2732                                            lustre_swab_ost_last_id);
2733                 if (reply == NULL) {
2734                         CERROR("Can't unpack OST last ID\n");
2735                         GOTO(out, rc = -EPROTO);
2736                 }
2737                 *((obd_id *)val) = *reply;
2738         out:
2739                 ptlrpc_req_finished(req);
2740                 RETURN(rc);
2741         }
2742         RETURN(-EINVAL);
2743 }
2744
2745 static int osc_set_info(struct obd_export *exp, obd_count keylen,
2746                         void *key, obd_count vallen, void *val)
2747 {
2748         struct ptlrpc_request *req;
2749         struct obd_device  *obd = exp->exp_obd;
2750         struct obd_import *imp = class_exp2cliimp(exp);
2751         struct llog_ctxt *ctxt;
2752         int rc, size = keylen;
2753         char *bufs[1] = {key};
2754         ENTRY;
2755
2756         if (keylen == strlen("next_id") &&
2757             memcmp(key, "next_id", strlen("next_id")) == 0) {
2758                 if (vallen != sizeof(obd_id))
2759                         RETURN(-EINVAL);
2760                 obd->u.cli.cl_oscc.oscc_next_id = *((obd_id*)val) + 1;
2761                 CDEBUG(D_INODE, "%s: set oscc_next_id = "LPU64"\n",
2762                        exp->exp_obd->obd_name,
2763                        obd->u.cli.cl_oscc.oscc_next_id);
2764
2765                 RETURN(0);
2766         }
2767
2768         if (keylen == strlen("growth_count") &&
2769             memcmp(key, "growth_count", strlen("growth_count")) == 0) {
2770                 if (vallen != sizeof(int))
2771                         RETURN(-EINVAL);
2772                 obd->u.cli.cl_oscc.oscc_grow_count = *((int*)val);
2773                 RETURN(0);
2774         }
2775
2776         if (keylen == strlen("unlinked") &&
2777             memcmp(key, "unlinked", keylen) == 0) {
2778                 struct osc_creator *oscc = &obd->u.cli.cl_oscc;
2779                 spin_lock(&oscc->oscc_lock);
2780                 oscc->oscc_flags &= ~OSCC_FLAG_NOSPC;
2781                 spin_unlock(&oscc->oscc_lock);
2782                 RETURN(0);
2783         }
2784
2785
2786         if (keylen == strlen("initial_recov") &&
2787             memcmp(key, "initial_recov", strlen("initial_recov")) == 0) {
2788                 struct obd_import *imp = exp->exp_obd->u.cli.cl_import;
2789                 if (vallen != sizeof(int))
2790                         RETURN(-EINVAL);
2791                 imp->imp_initial_recov = *(int *)val;
2792                 CDEBUG(D_HA, "%s: set imp_no_init_recov = %d\n",
2793                        exp->exp_obd->obd_name,
2794                        imp->imp_initial_recov);
2795                 RETURN(0);
2796         }
2797
2798         if (keylen < strlen("mds_conn") ||
2799             memcmp(key, "mds_conn", strlen("mds_conn")) != 0)
2800                 RETURN(-EINVAL);
2801
2802
2803         req = ptlrpc_prep_req(imp, OST_SET_INFO, 1, &size, bufs);
2804         if (req == NULL)
2805                 RETURN(-ENOMEM);
2806
2807         req->rq_replen = lustre_msg_size(0, NULL);
2808         rc = ptlrpc_queue_wait(req);
2809         ptlrpc_req_finished(req);
2810
2811         ctxt = llog_get_context(exp->exp_obd, LLOG_UNLINK_ORIG_CTXT);
2812         if (ctxt) {
2813                 rc = llog_initiator_connect(ctxt);
2814                 if (rc)
2815                         RETURN(rc);
2816         }
2817
2818         imp->imp_server_timeout = 1;
2819         CDEBUG(D_HA, "pinging OST %s\n", imp->imp_target_uuid.uuid);
2820         imp->imp_pingable = 1;
2821
2822         RETURN(rc);
2823 }
2824
2825
2826 static struct llog_operations osc_size_repl_logops = {
2827         lop_cancel: llog_obd_repl_cancel
2828 };
2829
2830 static struct llog_operations osc_unlink_orig_logops;
2831 static int osc_llog_init(struct obd_device *obd, struct obd_device *tgt,
2832                         int count, struct llog_catid *catid)
2833 {
2834         int rc;
2835         ENTRY;
2836
2837         osc_unlink_orig_logops = llog_lvfs_ops;
2838         osc_unlink_orig_logops.lop_setup = llog_obd_origin_setup;
2839         osc_unlink_orig_logops.lop_cleanup = llog_obd_origin_cleanup;
2840         osc_unlink_orig_logops.lop_add = llog_obd_origin_add;
2841         osc_unlink_orig_logops.lop_connect = llog_origin_connect;
2842
2843         rc = llog_setup(obd, LLOG_UNLINK_ORIG_CTXT, tgt, count,
2844                         &catid->lci_logid, &osc_unlink_orig_logops);
2845         if (rc)
2846                 RETURN(rc);
2847
2848         rc = llog_setup(obd, LLOG_SIZE_REPL_CTXT, tgt, count, NULL,
2849                         &osc_size_repl_logops);
2850         RETURN(rc);
2851 }
2852
2853 static int osc_llog_finish(struct obd_device *obd, int count)
2854 {
2855         int rc;
2856         ENTRY;
2857
2858         rc = llog_cleanup(llog_get_context(obd, LLOG_UNLINK_ORIG_CTXT));
2859         if (rc)
2860                 RETURN(rc);
2861
2862         rc = llog_cleanup(llog_get_context(obd, LLOG_SIZE_REPL_CTXT));
2863         RETURN(rc);
2864 }
2865
2866
2867 static int osc_connect(struct lustre_handle *exph,
2868                        struct obd_device *obd, struct obd_uuid *cluuid)
2869 {
2870         int rc;
2871
2872         rc = client_connect_import(exph, obd, cluuid);
2873
2874         return rc;
2875 }
2876
2877 static int osc_disconnect(struct obd_export *exp, int flags)
2878 {
2879         struct obd_device *obd = class_exp2obd(exp);
2880         struct llog_ctxt *ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
2881         int rc;
2882
2883         if (obd->u.cli.cl_conn_count == 1)
2884                 /* flush any remaining cancel messages out to the target */
2885                 llog_sync(ctxt, exp);
2886
2887         rc = client_disconnect_export(exp, flags);
2888         return rc;
2889 }
2890
2891 static int osc_import_event(struct obd_device *obd,
2892                             struct obd_import *imp,
2893                             enum obd_import_event event)
2894 {
2895         struct client_obd *cli;
2896         int rc = 0;
2897
2898         LASSERT(imp->imp_obd == obd);
2899
2900         switch (event) {
2901         case IMP_EVENT_DISCON: {
2902                 /* Only do this on the MDS OSC's */
2903                 if (imp->imp_server_timeout) {
2904                         struct osc_creator *oscc = &obd->u.cli.cl_oscc;
2905
2906                         spin_lock(&oscc->oscc_lock);
2907                         oscc->oscc_flags |= OSCC_FLAG_RECOVERING;
2908                         spin_unlock(&oscc->oscc_lock);
2909                 }
2910                 break;
2911         }
2912         case IMP_EVENT_INACTIVE: {
2913                 if (obd->obd_observer)
2914                         rc = obd_notify(obd->obd_observer, obd, 0);
2915                 break;
2916         }
2917         case IMP_EVENT_INVALIDATE: {
2918                 struct ldlm_namespace *ns = obd->obd_namespace;
2919
2920                 /* Reset grants */
2921                 cli = &obd->u.cli;
2922                 spin_lock(&cli->cl_loi_list_lock);
2923                 cli->cl_avail_grant = 0;
2924                 cli->cl_lost_grant = 0;
2925                 /* all pages go to failing rpcs due to the invalid import */
2926                 osc_check_rpcs(cli);
2927                 spin_unlock(&cli->cl_loi_list_lock);
2928
2929                 ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
2930
2931                 break;
2932         }
2933         case IMP_EVENT_ACTIVE: {
2934                 if (obd->obd_observer)
2935                         rc = obd_notify(obd->obd_observer, obd, 1);
2936                 break;
2937         }
2938         default:
2939                 CERROR("Unknown import event %d\n", event);
2940                 LBUG();
2941         }
2942         RETURN(rc);
2943 }
2944
2945 int osc_setup(struct obd_device *obd, obd_count len, void *buf)
2946 {
2947         int rc;
2948
2949         rc = ptlrpcd_addref();
2950         if (rc)
2951                 return rc;
2952
2953         rc = client_obd_setup(obd, len, buf);
2954         if (rc)
2955                 ptlrpcd_decref();
2956         else
2957                 oscc_init(obd);
2958
2959         RETURN(rc);
2960 }
2961
2962 int osc_cleanup(struct obd_device *obd, int flags)
2963 {
2964         int rc;
2965
2966         rc = client_obd_cleanup(obd, flags);
2967         ptlrpcd_decref();
2968         RETURN(rc);
2969 }
2970
2971
2972 struct obd_ops osc_obd_ops = {
2973         o_owner:        THIS_MODULE,
2974         o_attach:       osc_attach,
2975         o_detach:       osc_detach,
2976         o_setup:        osc_setup,
2977         o_cleanup:      osc_cleanup,
2978         o_connect:      osc_connect,
2979         o_disconnect:   osc_disconnect,
2980         o_statfs:       osc_statfs,
2981         o_packmd:       osc_packmd,
2982         o_unpackmd:     osc_unpackmd,
2983         o_create:       osc_create,
2984         o_destroy:      osc_destroy,
2985         o_getattr:      osc_getattr,
2986         o_getattr_async:osc_getattr_async,
2987         o_setattr:      osc_setattr,
2988         o_brw:          osc_brw,
2989         o_brw_async:    osc_brw_async,
2990         .o_prep_async_page =            osc_prep_async_page,
2991         .o_queue_async_io =             osc_queue_async_io,
2992         .o_set_async_flags =            osc_set_async_flags,
2993         .o_queue_group_io =             osc_queue_group_io,
2994         .o_trigger_group_io =           osc_trigger_group_io,
2995         .o_teardown_async_page =        osc_teardown_async_page,
2996         o_punch:        osc_punch,
2997         o_sync:         osc_sync,
2998         o_enqueue:      osc_enqueue,
2999         o_match:        osc_match,
3000         o_change_cbdata:osc_change_cbdata,
3001         o_cancel:       osc_cancel,
3002         o_cancel_unused:osc_cancel_unused,
3003         o_iocontrol:    osc_iocontrol,
3004         o_get_info:     osc_get_info,
3005         o_set_info:     osc_set_info,
3006         o_import_event: osc_import_event,
3007         o_llog_init:    osc_llog_init,
3008         o_llog_finish:  osc_llog_finish,
3009 };
3010
3011 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
3012 struct obd_ops sanosc_obd_ops = {
3013         o_owner:        THIS_MODULE,
3014         o_attach:       osc_attach,
3015         o_detach:       osc_detach,
3016         o_cleanup:      client_obd_cleanup,
3017         o_connect:      osc_connect,
3018         o_disconnect:   client_disconnect_export,
3019         o_statfs:       osc_statfs,
3020         o_packmd:       osc_packmd,
3021         o_unpackmd:     osc_unpackmd,
3022         o_create:       osc_real_create,
3023         o_destroy:      osc_destroy,
3024         o_getattr:      osc_getattr,
3025         o_getattr_async:osc_getattr_async,
3026         o_setattr:      osc_setattr,
3027         o_setup:        client_sanobd_setup,
3028         o_brw:          sanosc_brw,
3029         o_punch:        osc_punch,
3030         o_sync:         osc_sync,
3031         o_enqueue:      osc_enqueue,
3032         o_match:        osc_match,
3033         o_change_cbdata:osc_change_cbdata,
3034         o_cancel:       osc_cancel,
3035         o_cancel_unused:osc_cancel_unused,
3036         o_iocontrol:    osc_iocontrol,
3037         o_import_event: osc_import_event,
3038         o_llog_init:    osc_llog_init,
3039         o_llog_finish:  osc_llog_finish,
3040 };
3041 #endif
3042
3043 int __init osc_init(void)
3044 {
3045         struct lprocfs_static_vars lvars;
3046 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
3047         struct lprocfs_static_vars sanlvars;
3048 #endif
3049         int rc;
3050         ENTRY;
3051
3052         lprocfs_init_vars(osc, &lvars);
3053 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
3054         lprocfs_init_vars(osc, &sanlvars);
3055 #endif
3056
3057         rc = class_register_type(&osc_obd_ops, lvars.module_vars,
3058                                  LUSTRE_OSC_NAME);
3059         if (rc)
3060                 RETURN(rc);
3061
3062 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
3063         rc = class_register_type(&sanosc_obd_ops, sanlvars.module_vars,
3064                                  LUSTRE_SANOSC_NAME);
3065         if (rc)
3066                 class_unregister_type(LUSTRE_OSC_NAME);
3067 #endif
3068
3069         RETURN(rc);
3070 }
3071
3072 static void /*__exit*/ osc_exit(void)
3073 {
3074 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
3075         class_unregister_type(LUSTRE_SANOSC_NAME);
3076 #endif
3077         class_unregister_type(LUSTRE_OSC_NAME);
3078 }
3079
3080 #ifdef __KERNEL__
3081 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
3082 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC)");
3083 MODULE_LICENSE("GPL");
3084
3085 module_init(osc_init);
3086 module_exit(osc_exit);
3087 #endif