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         ENTRY;
 320
 321         if (desc->b_flags & PTL_RPC_FL_INTR)
 322                 CERROR("got signal\n");
 323
 324         if (cb_data->callback)
 325                 cb_data->callback(cb_data->cb_data);
 326
 327         OBD_FREE(cb_data->obd_data, cb_data->obd_size);
 328         OBD_FREE(cb_data, sizeof(*cb_data));
 329
 330         /* We can't kunmap the desc from interrupt context, so we do it from
 331          * the bottom half above. */
 332         INIT_TQUEUE(&desc->b_queue, 0, 0);
 333         PREPARE_TQUEUE(&desc->b_queue, unmap_and_decref_bulk_desc, desc);
 334         schedule_task(&desc->b_queue);
 335
 336         EXIT;
 337 }
 338
 339 static int osc_brw_read(struct lustre_handle *conn, struct lov_stripe_md *md,
 340                         obd_count page_count, struct page **page_array,
 341                         obd_size *count, obd_off *offset, obd_flag *flags,
 342                         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(page_array[mapped]);
 394                 bulk->b_page = page_array[mapped];
 395                 bulk->b_buflen = PAGE_SIZE;
 396                 ost_pack_niobuf(&nioptr, offset[mapped], count[mapped],
 397                                 flags[mapped], 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         if (!callback)
 413                 ptlrpc_bulk_addref(desc);
 414         rc = ptlrpc_register_bulk(desc);
 415         if (rc)
 416                 GOTO(out_desc2, rc);
 417
 418         request->rq_replen = lustre_msg_size(1, size);
 419         rc = ptlrpc_queue_wait(request);
 420         rc = ptlrpc_check_status(request, rc);
 421
 422         /* XXX: Mike, this is the only place I'm not sure of.  If we had
 423          *      an error here, will we always call brw_finish?  If yes, then
 424          *      out_desc_2 will do too much and we should jump to out_desc.
 425          *      If maybe, then we are screwed, and we need to set things up
 426          *      so that bulk_sink_callback is called for each bulk page,
 427          *      even on error so brw_finish is always called.  It would need
 428          *      to be passed an error code as a parameter to know what to do.
 429          *
 430          *      That would also help with the partial completion case, so
 431          *      we could say in brw_finish "these pages are done, don't
 432          *      restart them" and osc_brw callers can know this.
 433          */
 434         if (rc)
 435                 GOTO(out_desc2, rc);
 436
 437         /* Callbacks cause asynchronous handling. */
 438         if (callback)
 439                 GOTO(out_req, rc = 0);
 440
 441         /* If there's no callback function, sleep here until complete. */
 442         l_wait_event_killable(desc->b_waitq, ptlrpc_check_bulk_received(desc));
 443         if (desc->b_flags & PTL_RPC_FL_INTR)
 444                 GOTO(out_desc, rc = -EINTR);
 445
 446         EXIT;
 447 out_desc:
 448         ptlrpc_bulk_decref(desc);
 449 out_req:
 450         ptlrpc_req_finished(request);
 451         RETURN(rc);
 452
 453         /* Clean up on error. */
 454 out_desc2:
 455         if (!callback)
 456                 ptlrpc_bulk_decref(desc);
 457 out_unmap:
 458         while (mapped-- > 0)
 459                 kunmap(page_array[mapped]);
 460         OBD_FREE(cb_data, sizeof(*cb_data));
 461         goto out_desc;
 462 }
 463
 464 static int osc_brw_write(struct lustre_handle *conn,
 465                          struct lov_stripe_md *md, obd_count page_count,
 466                          struct page **pagearray, obd_size *count,
 467                          obd_off *offset, obd_flag *flags,
 468                          brw_callback_t callback, void *data)
 469 {
 470         struct ptlrpc_connection *connection = client_conn2cli(conn)->cl_conn;
 471         struct ptlrpc_request *request = NULL;
 472         struct ptlrpc_bulk_desc *desc = NULL;
 473         struct ost_body *body;
 474         struct niobuf_local *local = NULL;
 475         struct niobuf_remote *remote;
 476         struct osc_brw_cb_data *cb_data = NULL;
 477         int rc, j, size[3] = {sizeof(*body)};
 478         void *iooptr, *nioptr;
 479         int mapped = 0;
 480         ENTRY;
 481
 482         size[1] = sizeof(struct obd_ioobj);
 483         size[2] = page_count * sizeof(*remote);
 484
 485         request = ptlrpc_prep_req2(conn, OST_WRITE, 3, size, NULL);
 486         if (!request)
 487                 RETURN(-ENOMEM);
 488
 489         body = lustre_msg_buf(request->rq_reqmsg, 0);
 490
 491         desc = ptlrpc_prep_bulk(connection);
 492         if (!desc)
 493                 GOTO(out_req, rc = -ENOMEM);
 494         desc->b_portal = OSC_BULK_PORTAL;
 495         desc->b_cb = brw_finish;
 496         OBD_ALLOC(cb_data, sizeof(*cb_data));
 497         if (!cb_data)
 498                 GOTO(out_desc, rc = -ENOMEM);
 499
 500         cb_data->callback = callback;
 501         cb_data->cb_data = data;
 502         desc->b_cb_data = cb_data;
 503
 504         iooptr = lustre_msg_buf(request->rq_reqmsg, 1);
 505         nioptr = lustre_msg_buf(request->rq_reqmsg, 2);
 506         ost_pack_ioo(&iooptr, md, page_count);
 507         /* end almost identical to brw_read case */
 508
 509         OBD_ALLOC(local, page_count * sizeof(*local));
 510         if (!local)
 511                 GOTO(out_cb, rc = -ENOMEM);
 512
 513         cb_data->obd_data = local;
 514         cb_data->obd_size = page_count * sizeof(*local);
 515
 516         for (mapped = 0; mapped < page_count; mapped++) {
 517                 local[mapped].addr = kmap(pagearray[mapped]);
 518                 local[mapped].offset = offset[mapped];
 519                 local[mapped].len = count[mapped];
 520                 ost_pack_niobuf(&nioptr, offset[mapped], count[mapped],
 521                                 flags[mapped], 0);
 522         }
 523
 524         size[1] = page_count * sizeof(*remote);
 525         request->rq_replen = lustre_msg_size(2, size);
 526         rc = ptlrpc_queue_wait(request);
 527         rc = ptlrpc_check_status(request, rc);
 528         if (rc)
 529                 GOTO(out_unmap, rc);
 530
 531         nioptr = lustre_msg_buf(request->rq_repmsg, 1);
 532         if (!nioptr)
 533                 GOTO(out_unmap, rc = -EINVAL);
 534
 535         if (request->rq_repmsg->buflens[1] != size[1]) {
 536                 CERROR("buffer length wrong (%d vs. %d)\n",
 537                        request->rq_repmsg->buflens[1], size[1]);
 538                 GOTO(out_unmap, rc = -EINVAL);
 539         }
 540
 541         for (j = 0; j < page_count; j++) {
 542                 struct ptlrpc_bulk_page *bulk;
 543
 544                 ost_unpack_niobuf(&nioptr, &remote);
 545
 546                 bulk = ptlrpc_prep_bulk_page(desc);
 547                 if (!bulk)
 548                         GOTO(out_unmap, rc = -ENOMEM);
 549
 550                 bulk->b_buf = (void *)(unsigned long)local[j].addr;
 551                 bulk->b_buflen = local[j].len;
 552                 bulk->b_xid = remote->xid;
 553                 bulk->b_page = pagearray[j];
 554         }
 555
 556         if (desc->b_page_count != page_count)
 557                 LBUG();
 558
 559         /*
 560          * One reference is released when brw_finish is complete, the
 561          * other here when we finish waiting on it if we don't have a callback.
 562          *
 563          * We don't reference the bulk descriptor again here if there is a
 564          * callback, so we don't need an additional refcount on it.
 565          */
 566         if (!callback)
 567                 ptlrpc_bulk_addref(desc);
 568         rc = ptlrpc_send_bulk(desc);
 569
 570         /* XXX: Mike, same question as in osc_brw_read. */
 571         if (rc)
 572                 GOTO(out_desc2, rc);
 573
 574         /* Callbacks cause asynchronous handling. */
 575         if (callback)
 576                 GOTO(out_req, rc = 0);
 577
 578         /* If there's no callback function, sleep here until complete. */
 579         l_wait_event_killable(desc->b_waitq, ptlrpc_check_bulk_sent(desc));
 580         if (desc->b_flags & PTL_RPC_FL_INTR)
 581                 GOTO(out_desc, rc = -EINTR);
 582
 583         EXIT;
 584 out_desc:
 585         ptlrpc_bulk_decref(desc);
 586 out_req:
 587         ptlrpc_req_finished(request);
 588         return rc;
 589
 590         /* Clean up on error. */
 591 out_desc2:
 592         if (!callback)
 593                 ptlrpc_bulk_decref(desc);
 594 out_unmap:
 595         while (mapped-- > 0)
 596                 kunmap(pagearray[mapped]);
 597
 598         OBD_FREE(local, page_count * sizeof(*local));
 599 out_cb:
 600         OBD_FREE(cb_data, sizeof(*cb_data));
 601         goto out_desc;
 602 }
 603
 604 static int osc_brw(int cmd, struct lustre_handle *conn,
 605                    struct lov_stripe_md *md, obd_count page_count,
 606                    struct page **page_array, obd_size *count, obd_off *offset,
 607                    obd_flag *flags, brw_callback_t callback, void *data)
 608 {
 609         if (cmd & OBD_BRW_WRITE)
 610                 return osc_brw_write(conn, md, page_count, page_array, count,
 611                                      offset, flags, callback, data);
 612         else
 613                 return osc_brw_read(conn, md, page_count, page_array, count,
 614                                     offset, flags, callback, data);
 615 }
 616
 617 static int osc_enqueue(struct lustre_handle *connh,
 618                        struct lustre_handle *parent_lock, __u64 *res_id,
 619                        __u32 type, void *extentp, int extent_len, __u32 mode,
 620                        int *flags, void *callback, void *data, int datalen,
 621                        struct lustre_handle *lockh)
 622 {
 623         struct obd_device *obddev = class_conn2obd(connh);
 624         struct ldlm_extent *extent = extentp;
 625         int rc;
 626         __u32 mode2;
 627
 628         /* Filesystem locks are given a bit of special treatment: first we
 629          * fixup the lock to start and end on page boundaries. */
 630         extent->start &= PAGE_MASK;
 631         extent->end = (extent->end + PAGE_SIZE - 1) & PAGE_MASK;
 632
 633         /* Next, search for already existing extent locks that will cover us */
 634         //osc_con2dlmcl(conn, &cl, &connection, &rconn);
 635         rc = ldlm_lock_match(obddev->obd_namespace, res_id, type, extent,
 636                              sizeof(extent), mode, lockh);
 637         if (rc == 1) {
 638                 /* We already have a lock, and it's referenced */
 639                 return 0;
 640         }
 641
 642         /* Next, search for locks that we can upgrade (if we're trying to write)
 643          * or are more than we need (if we're trying to read).  Because the VFS
 644          * and page cache already protect us locally, lots of readers/writers
 645          * can share a single PW lock. */
 646         if (mode == LCK_PW)
 647                 mode2 = LCK_PR;
 648         else
 649                 mode2 = LCK_PW;
 650
 651         rc = ldlm_lock_match(obddev->obd_namespace, res_id, type, extent,
 652                              sizeof(extent), mode2, lockh);
 653         if (rc == 1) {
 654                 int flags;
 655                 /* FIXME: This is not incredibly elegant, but it might
 656                  * be more elegant than adding another parameter to
 657                  * lock_match.  I want a second opinion. */
 658                 ldlm_lock_addref(lockh, mode);
 659                 ldlm_lock_decref(lockh, mode2);
 660
 661                 if (mode == LCK_PR)
 662                         return 0;
 663
 664                 rc = ldlm_cli_convert(lockh, mode, &flags);
 665                 if (rc)
 666                         LBUG();
 667
 668                 return rc;
 669         }
 670
 671         rc = ldlm_cli_enqueue(connh, NULL,obddev->obd_namespace,
 672                               parent_lock, res_id, type, extent, sizeof(extent),
 673                               mode, flags, ldlm_completion_ast, callback, data, datalen, lockh);
 674         return rc;
 675 }
 676
 677 static int osc_cancel(struct lustre_handle *oconn, __u32 mode,
 678                       struct lustre_handle *lockh)
 679 {
 680         ENTRY;
 681
 682         ldlm_lock_decref(lockh, mode);
 683
 684         RETURN(0);
 685 }
 686
 687 static int osc_statfs(struct lustre_handle *conn, struct statfs *sfs)
 688 {
 689         struct ptlrpc_request *request;
 690         struct obd_statfs *osfs;
 691         int rc, size = sizeof(*osfs);
 692         ENTRY;
 693
 694         request = ptlrpc_prep_req2(conn, OST_STATFS, 0, NULL, NULL);
 695         if (!request)
 696                 RETURN(-ENOMEM);
 697
 698         request->rq_replen = lustre_msg_size(1, &size);
 699
 700         rc = ptlrpc_queue_wait(request);
 701         rc = ptlrpc_check_status(request, rc);
 702         if (rc) {
 703                 CERROR("%s failed: rc = %d\n", __FUNCTION__, rc);
 704                 GOTO(out, rc);
 705         }
 706
 707         osfs = lustre_msg_buf(request->rq_repmsg, 0);
 708         obd_statfs_unpack(osfs, sfs);
 709
 710         EXIT;
 711  out:
 712         ptlrpc_free_req(request);
 713         return rc;
 714 }
 715
 716 struct obd_ops osc_obd_ops = {
 717         o_setup:        client_obd_setup,
 718         o_cleanup:      client_obd_cleanup,
 719         o_statfs:       osc_statfs,
 720         o_create:       osc_create,
 721         o_destroy:      osc_destroy,
 722         o_getattr:      osc_getattr,
 723         o_setattr:      osc_setattr,
 724         o_open:         osc_open,
 725         o_close:        osc_close,
 726         o_connect:      client_obd_connect,
 727         o_disconnect:   client_obd_disconnect,
 728         o_brw:          osc_brw,
 729         o_punch:        osc_punch,
 730         o_enqueue:      osc_enqueue,
 731         o_cancel:       osc_cancel
 732 };
 733
 734 static int __init osc_init(void)
 735 {
 736         return class_register_type(&osc_obd_ops, LUSTRE_OSC_NAME);
 737 }
 738
 739 static void __exit osc_exit(void)
 740 {
 741         class_unregister_type(LUSTRE_OSC_NAME);
 742 }
 743
 744 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 745 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC) v1.0");
 746 MODULE_LICENSE("GPL");
 747
 748 module_init(osc_init);
 749 module_exit(osc_exit);