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, 2002 Cluster File Systems, Inc.
   5  *
   6  *  This code is issued under the GNU General Public License.
   7  *  See the file COPYING in this distribution
   8  *
   9  *  Author Peter Braam <braam@clusterfs.com>
  10  *
  11  *  This server is single threaded at present (but can easily be multi
  12  *  threaded). For testing and management it is treated as an
  13  *  obd_device, although it does not export a full OBD method table
  14  *  (the requests are coming in over the wire, so object target
  15  *  modules do not have a full method table.)
  16  *
  17  */
  18
  19 #define EXPORT_SYMTAB
  20 #define DEBUG_SUBSYSTEM S_OSC
  21
  22 #include <linux/module.h>
  23 #include <linux/lustre_dlm.h>
  24 #include <linux/lustre_mds.h> /* for mds_objid */
  25 #include <linux/obd_ost.h>
  26 #include <linux/obd_lov.h>
  27 #include <linux/init.h>
  28
  29
  30
  31 static int osc_getattr(struct lustre_handle *conn, struct obdo *oa,
  32                        struct lov_stripe_md *md)
  33 {
  34         struct ptlrpc_request *request;
  35         struct ost_body *body;
  36         int rc, size = sizeof(*body);
  37         ENTRY;
  38
  39         request = ptlrpc_prep_req2(conn, OST_GETATTR, 1, &size, NULL);
  40         if (!request)
  41                 RETURN(-ENOMEM);
  42
  43         body = lustre_msg_buf(request->rq_reqmsg, 0);
  44 #warning FIXME: pack only valid fields instead of memcpy, endianness
  45         memcpy(&body->oa, oa, sizeof(*oa));
  46
  47         request->rq_replen = lustre_msg_size(1, &size);
  48
  49         rc = ptlrpc_queue_wait(request);
  50         rc = ptlrpc_check_status(request, rc);
  51         if (rc) {
  52                 CERROR("%s failed: rc = %d\n", __FUNCTION__, rc);
  53                 GOTO(out, rc);
  54         }
  55
  56         body = lustre_msg_buf(request->rq_repmsg, 0);
  57         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
  58         if (oa)
  59                 memcpy(oa, &body->oa, sizeof(*oa));
  60
  61         EXIT;
  62  out:
  63         ptlrpc_free_req(request);
  64         return rc;
  65 }
  66
  67 static int osc_open(struct lustre_handle *conn, struct obdo *oa,
  68                     struct lov_stripe_md *md)
  69 {
  70         struct ptlrpc_request *request;
  71         struct ost_body *body;
  72         int rc, size = sizeof(*body);
  73         ENTRY;
  74
  75         request = ptlrpc_prep_req2(conn, OST_OPEN, 1, &size, NULL);
  76         if (!request)
  77                 RETURN(-ENOMEM);
  78
  79         body = lustre_msg_buf(request->rq_reqmsg, 0);
  80 #warning FIXME: pack only valid fields instead of memcpy, endianness
  81         memcpy(&body->oa, oa, sizeof(*oa));
  82
  83         request->rq_replen = lustre_msg_size(1, &size);
  84
  85         rc = ptlrpc_queue_wait(request);
  86         rc = ptlrpc_check_status(request, rc);
  87         if (rc)
  88                 GOTO(out, rc);
  89
  90         body = lustre_msg_buf(request->rq_repmsg, 0);
  91         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
  92         if (oa)
  93                 memcpy(oa, &body->oa, sizeof(*oa));
  94
  95         EXIT;
  96  out:
  97         ptlrpc_free_req(request);
  98         return rc;
  99 }
 100
 101 static int osc_close(struct lustre_handle *conn, struct obdo *oa,
 102                      struct lov_stripe_md *md)
 103 {
 104         struct ptlrpc_request *request;
 105         struct ost_body *body;
 106         int rc, size = sizeof(*body);
 107         ENTRY;
 108
 109         request = ptlrpc_prep_req2(conn, OST_CLOSE, 1, &size, NULL);
 110         if (!request)
 111                 RETURN(-ENOMEM);
 112
 113         body = lustre_msg_buf(request->rq_reqmsg, 0);
 114 #warning FIXME: pack only valid fields instead of memcpy, endianness
 115         memcpy(&body->oa, oa, sizeof(*oa));
 116
 117         request->rq_replen = lustre_msg_size(1, &size);
 118
 119         rc = ptlrpc_queue_wait(request);
 120         rc = ptlrpc_check_status(request, rc);
 121         if (rc)
 122                 GOTO(out, rc);
 123
 124         body = lustre_msg_buf(request->rq_repmsg, 0);
 125         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
 126         if (oa)
 127                 memcpy(oa, &body->oa, sizeof(*oa));
 128
 129         EXIT;
 130  out:
 131         ptlrpc_free_req(request);
 132         return rc;
 133 }
 134
 135 static int osc_setattr(struct lustre_handle *conn, struct obdo *oa,
 136                        struct lov_stripe_md *md)
 137 {
 138         struct ptlrpc_request *request;
 139         struct ost_body *body;
 140         int rc, size = sizeof(*body);
 141         ENTRY;
 142
 143         request = ptlrpc_prep_req2(conn, OST_SETATTR, 1, &size, NULL);
 144         if (!request)
 145                 RETURN(-ENOMEM);
 146
 147         body = lustre_msg_buf(request->rq_reqmsg, 0);
 148         memcpy(&body->oa, oa, sizeof(*oa));
 149
 150         request->rq_replen = lustre_msg_size(1, &size);
 151
 152         rc = ptlrpc_queue_wait(request);
 153         rc = ptlrpc_check_status(request, rc);
 154         GOTO(out, rc);
 155
 156  out:
 157         ptlrpc_free_req(request);
 158         return rc;
 159 }
 160
 161 static int osc_create(struct lustre_handle *conn, struct obdo *oa,
 162                       struct lov_stripe_md **ea)
 163 {
 164         struct ptlrpc_request *request;
 165         struct ost_body *body;
 166         int rc, size = sizeof(*body);
 167         ENTRY;
 168
 169         if (!oa) {
 170                 CERROR("oa NULL\n");
 171                 RETURN(-EINVAL);
 172         }
 173
 174         if (!ea) {
 175                 LBUG();
 176         }
 177
 178         if (!*ea) {
 179                 OBD_ALLOC(*ea, oa->o_easize);
 180                 if (!*ea)
 181                         RETURN(-ENOMEM);
 182                 (*ea)->lmd_easize = oa->o_easize;
 183         }
 184
 185         request = ptlrpc_prep_req2(conn, OST_CREATE, 1, &size, NULL);
 186         if (!request)
 187                 RETURN(-ENOMEM);
 188
 189         body = lustre_msg_buf(request->rq_reqmsg, 0);
 190         memcpy(&body->oa, oa, sizeof(*oa));
 191
 192         request->rq_replen = lustre_msg_size(1, &size);
 193
 194         rc = ptlrpc_queue_wait(request);
 195         rc = ptlrpc_check_status(request, rc);
 196         if (rc)
 197                 GOTO(out, rc);
 198
 199         body = lustre_msg_buf(request->rq_repmsg, 0);
 200         memcpy(oa, &body->oa, sizeof(*oa));
 201
 202         (*ea)->lmd_object_id = oa->o_id;
 203         (*ea)->lmd_stripe_count = 1;
 204         EXIT;
 205  out:
 206         ptlrpc_free_req(request);
 207         return rc;
 208 }
 209
 210 static int osc_punch(struct lustre_handle *conn, struct obdo *oa,
 211                      struct lov_stripe_md *md, obd_size start,
 212                      obd_size end)
 213 {
 214         struct ptlrpc_request *request;
 215         struct ost_body *body;
 216         int rc, size = sizeof(*body);
 217         ENTRY;
 218
 219         if (!oa) {
 220                 CERROR("oa NULL\n");
 221                 RETURN(-EINVAL);
 222         }
 223
 224         request = ptlrpc_prep_req2(conn, OST_PUNCH, 1, &size, NULL);
 225         if (!request)
 226                 RETURN(-ENOMEM);
 227
 228         body = lustre_msg_buf(request->rq_reqmsg, 0);
 229 #warning FIXME: pack only valid fields instead of memcpy, endianness, valid
 230         memcpy(&body->oa, oa, sizeof(*oa));
 231
 232         /* overload the blocks and size fields in the oa with start/end */
 233 #warning FIXME: endianness, size=start, blocks=end?
 234         body->oa.o_blocks = start;
 235         body->oa.o_size = end;
 236         body->oa.o_valid |= OBD_MD_FLBLOCKS | OBD_MD_FLSIZE;
 237
 238         request->rq_replen = lustre_msg_size(1, &size);
 239
 240         rc = ptlrpc_queue_wait(request);
 241         rc = ptlrpc_check_status(request, rc);
 242         if (rc)
 243                 GOTO(out, rc);
 244
 245         body = lustre_msg_buf(request->rq_repmsg, 0);
 246         memcpy(oa, &body->oa, sizeof(*oa));
 247
 248         EXIT;
 249  out:
 250         ptlrpc_free_req(request);
 251         return rc;
 252 }
 253
 254 static int osc_destroy(struct lustre_handle *conn, struct obdo *oa,
 255                        struct lov_stripe_md *ea)
 256 {
 257         struct ptlrpc_request *request;
 258         struct ost_body *body;
 259         int rc, size = sizeof(*body);
 260         ENTRY;
 261
 262         if (!oa) {
 263                 CERROR("oa NULL\n");
 264                 RETURN(-EINVAL);
 265         }
 266         request = ptlrpc_prep_req2(conn, OST_DESTROY, 1, &size, NULL);
 267         if (!request)
 268                 RETURN(-ENOMEM);
 269
 270         body = lustre_msg_buf(request->rq_reqmsg, 0);
 271 #warning FIXME: pack only valid fields instead of memcpy, endianness
 272         memcpy(&body->oa, oa, sizeof(*oa));
 273
 274         request->rq_replen = lustre_msg_size(1, &size);
 275
 276         rc = ptlrpc_queue_wait(request);
 277         rc = ptlrpc_check_status(request, rc);
 278         if (rc)
 279                 GOTO(out, rc);
 280
 281         body = lustre_msg_buf(request->rq_repmsg, 0);
 282         memcpy(oa, &body->oa, sizeof(*oa));
 283
 284         EXIT;
 285  out:
 286         ptlrpc_free_req(request);
 287         return rc;
 288 }
 289
 290 struct osc_brw_cb_data {
 291         brw_callback_t callback;
 292         void *cb_data;
 293         void *obd_data;
 294         size_t obd_size;
 295 };
 296
 297 /* Our bulk-unmapping bottom half. */
 298 static void unmap_and_decref_bulk_desc(void *data)
 299 {
 300         struct ptlrpc_bulk_desc *desc = data;
 301         struct list_head *tmp;
 302         ENTRY;
 303
 304         /* This feels wrong to me. */
 305         list_for_each(tmp, &desc->b_page_list) {
 306                 struct ptlrpc_bulk_page *bulk;
 307                 bulk = list_entry(tmp, struct ptlrpc_bulk_page, b_link);
 308
 309                 kunmap(bulk->b_page);
 310         }
 311
 312         ptlrpc_bulk_decref(desc);
 313         EXIT;
 314 }
 315
 316 static void brw_finish(struct ptlrpc_bulk_desc *desc, void *data)
 317 {
 318         struct osc_brw_cb_data *cb_data = data;
 319         int err = 0;
 320         ENTRY;
 321
 322         if (desc->b_flags & PTL_RPC_FL_INTR) {
 323                 err = -ERESTARTSYS;
 324                 CERROR("got signal\n");
 325         }
 326
 327         if (cb_data->callback)
 328                 cb_data->callback(cb_data->cb_data, err, CB_PHASE_FINISH);
 329
 330         OBD_FREE(cb_data->obd_data, cb_data->obd_size);
 331         OBD_FREE(cb_data, sizeof(*cb_data));
 332
 333         /* We can't kunmap the desc from interrupt context, so we do it from
 334          * the bottom half above. */
 335         INIT_TQUEUE(&desc->b_queue, 0, 0);
 336         PREPARE_TQUEUE(&desc->b_queue, unmap_and_decref_bulk_desc, desc);
 337         schedule_task(&desc->b_queue);
 338
 339         EXIT;
 340 }
 341
 342 static int osc_brw_read(struct lustre_handle *conn, struct lov_stripe_md *md, obd_count page_count, struct brw_page *pga, brw_callback_t callback, void *data)
 343 {
 344         struct ptlrpc_connection *connection = client_conn2cli(conn)->cl_conn;
 345         struct ptlrpc_request *request = NULL;
 346         struct ptlrpc_bulk_desc *desc = NULL;
 347         struct ost_body *body;
 348         struct osc_brw_cb_data *cb_data = NULL;
 349         int rc, size[3] = {sizeof(*body)};
 350         void *iooptr, *nioptr;
 351         int mapped = 0;
 352         __u32 xid;
 353         ENTRY;
 354
 355         size[1] = sizeof(struct obd_ioobj);
 356         size[2] = page_count * sizeof(struct niobuf_remote);
 357
 358         request = ptlrpc_prep_req2(conn, OST_READ, 3, size, NULL);
 359         if (!request)
 360                 RETURN(-ENOMEM);
 361
 362         body = lustre_msg_buf(request->rq_reqmsg, 0);
 363
 364         desc = ptlrpc_prep_bulk(connection);
 365         if (!desc)
 366                 GOTO(out_req, rc = -ENOMEM);
 367         desc->b_portal = OST_BULK_PORTAL;
 368         desc->b_cb = brw_finish;
 369         OBD_ALLOC(cb_data, sizeof(*cb_data));
 370         if (!cb_data)
 371                 GOTO(out_desc, rc = -ENOMEM);
 372
 373         cb_data->callback = callback;
 374         cb_data->cb_data = data;
 375         desc->b_cb_data = cb_data;
 376
 377         iooptr = lustre_msg_buf(request->rq_reqmsg, 1);
 378         nioptr = lustre_msg_buf(request->rq_reqmsg, 2);
 379         ost_pack_ioo(&iooptr, md, page_count);
 380         /* end almost identical to brw_write case */
 381
 382         spin_lock(&connection->c_lock);
 383         xid = ++connection->c_xid_out;       /* single xid for all pages */
 384         spin_unlock(&connection->c_lock);
 385
 386         for (mapped = 0; mapped < page_count; mapped++) {
 387                 struct ptlrpc_bulk_page *bulk = ptlrpc_prep_bulk_page(desc);
 388                 if (bulk == NULL)
 389                         GOTO(out_unmap, rc = -ENOMEM);
 390
 391                 bulk->b_xid = xid;           /* single xid for all pages */
 392
 393                 bulk->b_buf = kmap(pga[mapped].pg);
 394                 bulk->b_page = pga[mapped].pg;
 395                 bulk->b_buflen = PAGE_SIZE;
 396                 ost_pack_niobuf(&nioptr, pga[mapped].off, pga[mapped].count,
 397                                 pga[mapped].flag, bulk->b_xid);
 398         }
 399
 400         /*
 401          * Register the bulk first, because the reply could arrive out of order,
 402          * and we want to be ready for the bulk data.
 403          *
 404          * One reference is released when brw_finish is complete, the
 405          * other here when we finish waiting on it if we don't have a callback.
 406          *
 407          * We don't reference the bulk descriptor again here if there is a
 408          * callback, so we don't need an additional refcount on it.
 409          *
 410          * On error, we never do the brw_finish, so we handle all decrefs.
 411          */
 412         rc = ptlrpc_register_bulk(desc);
 413         if (rc)
 414                 GOTO(out_unmap, rc);
 415
 416         request->rq_replen = lustre_msg_size(1, size);
 417         rc = ptlrpc_queue_wait(request);
 418         rc = ptlrpc_check_status(request, rc);
 419
 420         /* XXX: Mike, this is the only place I'm not sure of.  If we had
 421          *      an error here, will we always call brw_finish?  If yes, then
 422          *      out_desc_2 will do too much and we should jump to out_desc.
 423          *      If maybe, then we are screwed, and we need to set things up
 424          *      so that bulk_sink_callback is called for each bulk page,
 425          *      even on error so brw_finish is always called.  It would need
 426          *      to be passed an error code as a parameter to know what to do.
 427          *
 428          *      That would also help with the partial completion case, so
 429          *      we could say in brw_finish "these pages are done, don't
 430          *      restart them" and osc_brw callers can know this.
 431          */
 432         if (rc)
 433                 GOTO(out_unmap, rc);
 434
 435         /* Callbacks cause asynchronous handling. */
 436         rc = callback(data, 0, CB_PHASE_START);
 437
 438         EXIT;
 439 out_desc:
 440         ptlrpc_bulk_decref(desc);
 441 out_req:
 442         ptlrpc_req_finished(request);
 443         RETURN(rc);
 444
 445         /* Clean up on error. */
 446 out_unmap:
 447         while (mapped-- > 0)
 448                 kunmap(page_array[mapped]);
 449         OBD_FREE(cb_data, sizeof(*cb_data));
 450         goto out_desc;
 451 }
 452
 453 static int osc_brw_write(struct lustre_handle *conn,
 454                          struct lov_stripe_md *md, obd_count page_count,
 455                          struct brw_page *pga,
 456                          brw_callback_t callback, void *data)
 457 {
 458         struct ptlrpc_connection *connection = client_conn2cli(conn)->cl_conn;
 459         struct ptlrpc_request *request = NULL;
 460         struct ptlrpc_bulk_desc *desc = NULL;
 461         struct ost_body *body;
 462         struct niobuf_local *local = NULL;
 463         struct niobuf_remote *remote;
 464         struct osc_brw_cb_data *cb_data = NULL;
 465         int rc, j, size[3] = {sizeof(*body)};
 466         void *iooptr, *nioptr;
 467         int mapped = 0;
 468         ENTRY;
 469
 470         size[1] = sizeof(struct obd_ioobj);
 471         size[2] = page_count * sizeof(*remote);
 472
 473         request = ptlrpc_prep_req2(conn, OST_WRITE, 3, size, NULL);
 474         if (!request)
 475                 RETURN(-ENOMEM);
 476
 477         body = lustre_msg_buf(request->rq_reqmsg, 0);
 478
 479         desc = ptlrpc_prep_bulk(connection);
 480         if (!desc)
 481                 GOTO(out_req, rc = -ENOMEM);
 482         desc->b_portal = OSC_BULK_PORTAL;
 483         desc->b_cb = brw_finish;
 484         OBD_ALLOC(cb_data, sizeof(*cb_data));
 485         if (!cb_data)
 486                 GOTO(out_desc, rc = -ENOMEM);
 487
 488         cb_data->callback = callback;
 489         cb_data->cb_data = data;
 490         desc->b_cb_data = cb_data;
 491
 492         iooptr = lustre_msg_buf(request->rq_reqmsg, 1);
 493         nioptr = lustre_msg_buf(request->rq_reqmsg, 2);
 494         ost_pack_ioo(&iooptr, md, page_count);
 495         /* end almost identical to brw_read case */
 496
 497         OBD_ALLOC(local, page_count * sizeof(*local));
 498         if (!local)
 499                 GOTO(out_cb, rc = -ENOMEM);
 500
 501         cb_data->obd_data = local;
 502         cb_data->obd_size = page_count * sizeof(*local);
 503
 504         for (mapped = 0; mapped < page_count; mapped++) {
 505                 local[mapped].addr = kmap(pga[mapped].pg);
 506                 local[mapped].offset = pga[mapped].off;
 507                 local[mapped].len = pga[mapped].count;
 508                 ost_pack_niobuf(&nioptr, pga[mapped].off, pga[mapped].count,
 509                                 pga[mapped].flag, 0);
 510         }
 511
 512         size[1] = page_count * sizeof(*remote);
 513         request->rq_replen = lustre_msg_size(2, size);
 514         rc = ptlrpc_queue_wait(request);
 515         rc = ptlrpc_check_status(request, rc);
 516         if (rc)
 517                 GOTO(out_unmap, rc);
 518
 519         nioptr = lustre_msg_buf(request->rq_repmsg, 1);
 520         if (!nioptr)
 521                 GOTO(out_unmap, rc = -EINVAL);
 522
 523         if (request->rq_repmsg->buflens[1] != size[1]) {
 524                 CERROR("buffer length wrong (%d vs. %d)\n",
 525                        request->rq_repmsg->buflens[1], size[1]);
 526                 GOTO(out_unmap, rc = -EINVAL);
 527         }
 528
 529         for (j = 0; j < page_count; j++) {
 530                 struct ptlrpc_bulk_page *bulk;
 531
 532                 ost_unpack_niobuf(&nioptr, &remote);
 533
 534                 bulk = ptlrpc_prep_bulk_page(desc);
 535                 if (!bulk)
 536                         GOTO(out_unmap, rc = -ENOMEM);
 537
 538                 bulk->b_buf = (void *)(unsigned long)local[j].addr;
 539                 bulk->b_buflen = local[j].len;
 540                 bulk->b_xid = remote->xid;
 541                 bulk->b_page = pga[j].pg;
 542         }
 543
 544         if (desc->b_page_count != page_count)
 545                 LBUG();
 546
 547         /*
 548          * One reference is released when brw_finish is complete, the
 549          * other here when we finish waiting on it if we don't have a callback.
 550          */
 551         rc = ptlrpc_send_bulk(desc);
 552
 553         /* XXX: Mike, same question as in osc_brw_read. */
 554         if (rc)
 555                 GOTO(out_desc2, rc);
 556
 557         /* Callbacks cause asynchronous handling. */
 558         rc = callback(data, 0, CB_PHASE_START);
 559
 560         EXIT;
 561 out_desc:
 562         ptlrpc_bulk_decref(desc);
 563 out_req:
 564         ptlrpc_req_finished(request);
 565         return rc;
 566
 567         /* Clean up on error. */
 568 out_desc2:
 569         if (!callback)
 570                 ptlrpc_bulk_decref(desc);
 571 out_unmap:
 572         while (mapped-- > 0)
 573                 kunmap(pagearray[mapped]);
 574
 575         OBD_FREE(local, page_count * sizeof(*local));
 576 out_cb:
 577         OBD_FREE(cb_data, sizeof(*cb_data));
 578         goto out_desc;
 579 }
 580
 581 static int osc_brw(int cmd, struct lustre_handle *conn,
 582                    struct lov_stripe_md *md, obd_count page_count,
 583                    struct brw_page *pagear, brw_callback_t callback,
 584                    void *data)
 585 {
 586         if (cmd & OBD_BRW_WRITE)
 587                 return osc_brw_write(conn, md, page_count, pagear, callback, data);
 588         else
 589                 return osc_brw_read(conn, md, page_count, pagear, callback, data);
 590 }
 591
 592 static int osc_enqueue(struct lustre_handle *connh, struct lov_stripe_md *md,
 593                        struct lustre_handle *parent_lock,
 594                        __u32 type, void *extentp, int extent_len, __u32 mode,
 595                        int *flags, void *callback, void *data, int datalen,
 596                        struct lustre_handle *lockh)
 597 {
 598         __u64 res_id = { md->lmd_object_id };
 599         struct obd_device *obddev = class_conn2obd(connh);
 600         struct ldlm_extent *extent = extentp;
 601         int rc;
 602         __u32 mode2;
 603
 604         /* Filesystem locks are given a bit of special treatment: first we
 605          * fixup the lock to start and end on page boundaries. */
 606         extent->start &= PAGE_MASK;
 607         extent->end = (extent->end + PAGE_SIZE - 1) & PAGE_MASK;
 608
 609         /* Next, search for already existing extent locks that will cover us */
 610         //osc_con2dlmcl(conn, &cl, &connection, &rconn);
 611         rc = ldlm_lock_match(obddev->obd_namespace, &res_id, type, extent,
 612                              sizeof(extent), mode, lockh);
 613         if (rc == 1) {
 614                 /* We already have a lock, and it's referenced */
 615                 return 0;
 616         }
 617
 618         /* Next, search for locks that we can upgrade (if we're trying to write)
 619          * or are more than we need (if we're trying to read).  Because the VFS
 620          * and page cache already protect us locally, lots of readers/writers
 621          * can share a single PW lock. */
 622         if (mode == LCK_PW)
 623                 mode2 = LCK_PR;
 624         else
 625                 mode2 = LCK_PW;
 626
 627         rc = ldlm_lock_match(obddev->obd_namespace, &res_id, type, extent,
 628                              sizeof(extent), mode2, lockh);
 629         if (rc == 1) {
 630                 int flags;
 631                 /* FIXME: This is not incredibly elegant, but it might
 632                  * be more elegant than adding another parameter to
 633                  * lock_match.  I want a second opinion. */
 634                 ldlm_lock_addref(lockh, mode);
 635                 ldlm_lock_decref(lockh, mode2);
 636
 637                 if (mode == LCK_PR)
 638                         return 0;
 639
 640                 rc = ldlm_cli_convert(lockh, mode, &flags);
 641                 if (rc)
 642                         LBUG();
 643
 644                 return rc;
 645         }
 646
 647         rc = ldlm_cli_enqueue(connh, NULL,obddev->obd_namespace,
 648                               parent_lock, &res_id, type, extent, sizeof(extent),
 649                               mode, flags, ldlm_completion_ast, callback, data, datalen, lockh);
 650         return rc;
 651 }
 652
 653 static int osc_cancel(struct lustre_handle *oconn, struct lov_stripe_md *md, __u32 mode,
 654                       struct lustre_handle *lockh)
 655 {
 656         ENTRY;
 657
 658         ldlm_lock_decref(lockh, mode);
 659
 660         RETURN(0);
 661 }
 662
 663 static int osc_statfs(struct lustre_handle *conn, struct statfs *sfs)
 664 {
 665         struct ptlrpc_request *request;
 666         struct obd_statfs *osfs;
 667         int rc, size = sizeof(*osfs);
 668         ENTRY;
 669
 670         request = ptlrpc_prep_req2(conn, OST_STATFS, 0, NULL, NULL);
 671         if (!request)
 672                 RETURN(-ENOMEM);
 673
 674         request->rq_replen = lustre_msg_size(1, &size);
 675
 676         rc = ptlrpc_queue_wait(request);
 677         rc = ptlrpc_check_status(request, rc);
 678         if (rc) {
 679                 CERROR("%s failed: rc = %d\n", __FUNCTION__, rc);
 680                 GOTO(out, rc);
 681         }
 682
 683         osfs = lustre_msg_buf(request->rq_repmsg, 0);
 684         obd_statfs_unpack(osfs, sfs);
 685
 686         EXIT;
 687  out:
 688         ptlrpc_free_req(request);
 689         return rc;
 690 }
 691
 692 struct obd_ops osc_obd_ops = {
 693         o_setup:        client_obd_setup,
 694         o_cleanup:      client_obd_cleanup,
 695         o_statfs:       osc_statfs,
 696         o_create:       osc_create,
 697         o_destroy:      osc_destroy,
 698         o_getattr:      osc_getattr,
 699         o_setattr:      osc_setattr,
 700         o_open:         osc_open,
 701         o_close:        osc_close,
 702         o_connect:      client_obd_connect,
 703         o_disconnect:   client_obd_disconnect,
 704         o_brw:          osc_brw,
 705         o_punch:        osc_punch,
 706         o_enqueue:      osc_enqueue,
 707         o_cancel:       osc_cancel
 708 };
 709
 710 static int __init osc_init(void)
 711 {
 712         return class_register_type(&osc_obd_ops, LUSTRE_OSC_NAME);
 713 }
 714
 715 static void __exit osc_exit(void)
 716 {
 717         class_unregister_type(LUSTRE_OSC_NAME);
 718 }
 719
 720 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 721 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC) v1.0");
 722 MODULE_LICENSE("GPL");
 723
 724 module_init(osc_init);
 725 module_exit(osc_exit);