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/version.h>
  23 #include <linux/module.h>
  24 #include <linux/mm.h>
  25 #include <linux/highmem.h>
  26 #include <linux/lustre_dlm.h>
  27 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
  28 #include <linux/workqueue.h>
  29 #endif
  30 #include <linux/kp30.h>
  31 #include <linux/lustre_mds.h> /* for mds_objid */
  32 #include <linux/obd_ost.h>
  33 #include <linux/obd_lov.h>
  34 #include <linux/ctype.h>
  35 #include <linux/init.h>
  36 #include <linux/lustre_ha.h>
  37 #include <linux/obd_support.h> /* for OBD_FAIL_CHECK */
  38 #include <linux/lustre_lite.h> /* for ll_i2info */
  39 #include <portals/lib-types.h> /* for PTL_MD_MAX_IOV */
  40
  41 static int osc_getattr(struct lustre_handle *conn, struct obdo *oa,
  42                        struct lov_stripe_md *md)
  43 {
  44         struct ptlrpc_request *request;
  45         struct ost_body *body;
  46         int rc, size = sizeof(*body);
  47         ENTRY;
  48
  49         request = ptlrpc_prep_req(class_conn2cliimp(conn), OST_GETATTR, 1,
  50                                   &size, NULL);
  51         if (!request)
  52                 RETURN(-ENOMEM);
  53
  54         body = lustre_msg_buf(request->rq_reqmsg, 0);
  55 #warning FIXME: pack only valid fields instead of memcpy, endianness
  56         memcpy(&body->oa, oa, sizeof(*oa));
  57
  58         request->rq_replen = lustre_msg_size(1, &size);
  59
  60         rc = ptlrpc_queue_wait(request);
  61         rc = ptlrpc_check_status(request, rc);
  62         if (rc) {
  63                 CERROR("%s failed: rc = %d\n", __FUNCTION__, rc);
  64                 GOTO(out, rc);
  65         }
  66
  67         body = lustre_msg_buf(request->rq_repmsg, 0);
  68         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
  69         if (oa)
  70                 memcpy(oa, &body->oa, sizeof(*oa));
  71
  72         EXIT;
  73  out:
  74         ptlrpc_req_finished(request);
  75         return rc;
  76 }
  77
  78 static int osc_open(struct lustre_handle *conn, struct obdo *oa,
  79                     struct lov_stripe_md *md)
  80 {
  81         struct ptlrpc_request *request;
  82         struct ost_body *body;
  83         int rc, size = sizeof(*body);
  84         ENTRY;
  85
  86         request = ptlrpc_prep_req(class_conn2cliimp(conn), OST_OPEN, 1, &size,
  87                                   NULL);
  88         if (!request)
  89                 RETURN(-ENOMEM);
  90
  91         body = lustre_msg_buf(request->rq_reqmsg, 0);
  92 #warning FIXME: pack only valid fields instead of memcpy, endianness
  93         memcpy(&body->oa, oa, sizeof(*oa));
  94
  95         request->rq_replen = lustre_msg_size(1, &size);
  96
  97         rc = ptlrpc_queue_wait(request);
  98         rc = ptlrpc_check_status(request, rc);
  99         if (rc)
 100                 GOTO(out, rc);
 101
 102         body = lustre_msg_buf(request->rq_repmsg, 0);
 103         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
 104         if (oa)
 105                 memcpy(oa, &body->oa, sizeof(*oa));
 106
 107         EXIT;
 108  out:
 109         ptlrpc_req_finished(request);
 110         return rc;
 111 }
 112
 113 static int osc_close(struct lustre_handle *conn, struct obdo *oa,
 114                      struct lov_stripe_md *md)
 115 {
 116         struct ptlrpc_request *request;
 117         struct ost_body *body;
 118         int rc, size = sizeof(*body);
 119         ENTRY;
 120
 121         request = ptlrpc_prep_req(class_conn2cliimp(conn), OST_CLOSE, 1, &size,
 122                                   NULL);
 123         if (!request)
 124                 RETURN(-ENOMEM);
 125
 126         body = lustre_msg_buf(request->rq_reqmsg, 0);
 127 #warning FIXME: pack only valid fields instead of memcpy, endianness
 128         memcpy(&body->oa, oa, sizeof(*oa));
 129
 130         request->rq_replen = lustre_msg_size(1, &size);
 131
 132         rc = ptlrpc_queue_wait(request);
 133         rc = ptlrpc_check_status(request, rc);
 134         if (rc)
 135                 GOTO(out, rc);
 136
 137         body = lustre_msg_buf(request->rq_repmsg, 0);
 138         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
 139         if (oa)
 140                 memcpy(oa, &body->oa, sizeof(*oa));
 141
 142         EXIT;
 143  out:
 144         ptlrpc_req_finished(request);
 145         return rc;
 146 }
 147
 148 static int osc_setattr(struct lustre_handle *conn, struct obdo *oa,
 149                        struct lov_stripe_md *md)
 150 {
 151         struct ptlrpc_request *request;
 152         struct ost_body *body;
 153         int rc, size = sizeof(*body);
 154         ENTRY;
 155
 156         request = ptlrpc_prep_req(class_conn2cliimp(conn), OST_SETATTR, 1,
 157                                   &size, NULL);
 158         if (!request)
 159                 RETURN(-ENOMEM);
 160
 161         body = lustre_msg_buf(request->rq_reqmsg, 0);
 162         memcpy(&body->oa, oa, sizeof(*oa));
 163
 164         request->rq_replen = lustre_msg_size(1, &size);
 165
 166         rc = ptlrpc_queue_wait(request);
 167         rc = ptlrpc_check_status(request, rc);
 168
 169         ptlrpc_req_finished(request);
 170         return rc;
 171 }
 172
 173 static int osc_create(struct lustre_handle *conn, struct obdo *oa,
 174                       struct lov_stripe_md **ea)
 175 {
 176         struct ptlrpc_request *request;
 177         struct ost_body *body;
 178         struct lov_stripe_md *lsm;
 179         int rc, size = sizeof(*body);
 180         ENTRY;
 181
 182         LASSERT(oa);
 183         LASSERT(ea);
 184
 185         lsm = *ea;
 186         if (!lsm) {
 187                 // XXX check oa->o_valid & OBD_MD_FLEASIZE first...
 188                 OBD_ALLOC(lsm, oa->o_easize);
 189                 if (!lsm)
 190                         RETURN(-ENOMEM);
 191                 lsm->lsm_mds_easize = oa->o_easize;
 192         }
 193
 194         request = ptlrpc_prep_req(class_conn2cliimp(conn), OST_CREATE, 1, &size,
 195                                   NULL);
 196         if (!request)
 197                 GOTO(out, rc = -ENOMEM);
 198
 199         body = lustre_msg_buf(request->rq_reqmsg, 0);
 200         memcpy(&body->oa, oa, sizeof(*oa));
 201
 202         request->rq_replen = lustre_msg_size(1, &size);
 203
 204         rc = ptlrpc_queue_wait(request);
 205         rc = ptlrpc_check_status(request, rc);
 206         if (rc)
 207                 GOTO(out_req, rc);
 208
 209         body = lustre_msg_buf(request->rq_repmsg, 0);
 210         memcpy(oa, &body->oa, sizeof(*oa));
 211
 212         lsm->lsm_object_id = oa->o_id;
 213         lsm->lsm_stripe_count = 0;
 214         *ea = lsm;
 215         EXIT;
 216 out_req:
 217         ptlrpc_req_finished(request);
 218 out:
 219         if (rc && !*ea)
 220                 OBD_FREE(lsm, oa->o_easize);
 221         return rc;
 222 }
 223
 224 static int osc_punch(struct lustre_handle *conn, struct obdo *oa,
 225                      struct lov_stripe_md *md, obd_size start,
 226                      obd_size end)
 227 {
 228         struct ptlrpc_request *request;
 229         struct ost_body *body;
 230         int rc, size = sizeof(*body);
 231         ENTRY;
 232
 233         if (!oa) {
 234                 CERROR("oa NULL\n");
 235                 RETURN(-EINVAL);
 236         }
 237
 238         request = ptlrpc_prep_req(class_conn2cliimp(conn), OST_PUNCH, 1, &size,
 239                                   NULL);
 240         if (!request)
 241                 RETURN(-ENOMEM);
 242
 243         body = lustre_msg_buf(request->rq_reqmsg, 0);
 244 #warning FIXME: pack only valid fields instead of memcpy, endianness, valid
 245         memcpy(&body->oa, oa, sizeof(*oa));
 246
 247         /* overload the size and blocks fields in the oa with start/end */
 248         body->oa.o_size = HTON__u64(start);
 249         body->oa.o_blocks = HTON__u64(end);
 250         body->oa.o_valid |= HTON__u32(OBD_MD_FLSIZE | OBD_MD_FLBLOCKS);
 251
 252         request->rq_replen = lustre_msg_size(1, &size);
 253
 254         rc = ptlrpc_queue_wait(request);
 255         rc = ptlrpc_check_status(request, rc);
 256         if (rc)
 257                 GOTO(out, rc);
 258
 259         body = lustre_msg_buf(request->rq_repmsg, 0);
 260         memcpy(oa, &body->oa, sizeof(*oa));
 261
 262         EXIT;
 263  out:
 264         ptlrpc_req_finished(request);
 265         return rc;
 266 }
 267
 268 static int osc_destroy(struct lustre_handle *conn, struct obdo *oa,
 269                        struct lov_stripe_md *ea)
 270 {
 271         struct ptlrpc_request *request;
 272         struct ost_body *body;
 273         int rc, size = sizeof(*body);
 274         ENTRY;
 275
 276         if (!oa) {
 277                 CERROR("oa NULL\n");
 278                 RETURN(-EINVAL);
 279         }
 280         request = ptlrpc_prep_req(class_conn2cliimp(conn), OST_DESTROY, 1,
 281                                   &size, NULL);
 282         if (!request)
 283                 RETURN(-ENOMEM);
 284
 285         body = lustre_msg_buf(request->rq_reqmsg, 0);
 286 #warning FIXME: pack only valid fields instead of memcpy, endianness
 287         memcpy(&body->oa, oa, sizeof(*oa));
 288
 289         request->rq_replen = lustre_msg_size(1, &size);
 290
 291         rc = ptlrpc_queue_wait(request);
 292         rc = ptlrpc_check_status(request, rc);
 293         if (rc)
 294                 GOTO(out, rc);
 295
 296         body = lustre_msg_buf(request->rq_repmsg, 0);
 297         memcpy(oa, &body->oa, sizeof(*oa));
 298
 299         EXIT;
 300  out:
 301         ptlrpc_req_finished(request);
 302         return rc;
 303 }
 304
 305 struct osc_brw_cb_data {
 306         brw_callback_t callback;
 307         void *cb_data;
 308         void *obd_data;
 309         size_t obd_size;
 310 };
 311
 312 /* Our bulk-unmapping bottom half. */
 313 static void unmap_and_decref_bulk_desc(void *data)
 314 {
 315         struct ptlrpc_bulk_desc *desc = data;
 316         struct list_head *tmp;
 317         ENTRY;
 318
 319         /* This feels wrong to me. */
 320         list_for_each(tmp, &desc->bd_page_list) {
 321                 struct ptlrpc_bulk_page *bulk;
 322                 bulk = list_entry(tmp, struct ptlrpc_bulk_page, bp_link);
 323
 324                 kunmap(bulk->bp_page);
 325         }
 326
 327         ptlrpc_bulk_decref(desc);
 328         EXIT;
 329 }
 330
 331 static void brw_finish(struct ptlrpc_bulk_desc *desc, void *data)
 332 {
 333         struct osc_brw_cb_data *cb_data = data;
 334         int err = 0;
 335         ENTRY;
 336
 337         if (desc->bd_flags & PTL_RPC_FL_TIMEOUT) {
 338                 err = (desc->bd_flags & PTL_RPC_FL_INTR ? -ERESTARTSYS :
 339                        -ETIMEDOUT);
 340         }
 341
 342         if (cb_data->callback)
 343                 cb_data->callback(cb_data->cb_data, err, CB_PHASE_FINISH);
 344
 345         if (cb_data->obd_data)
 346                 OBD_FREE(cb_data->obd_data, cb_data->obd_size);
 347         OBD_FREE(cb_data, sizeof(*cb_data));
 348
 349         /* We can't kunmap the desc from interrupt context, so we do it from
 350          * the bottom half above. */
 351         prepare_work(&desc->bd_queue, unmap_and_decref_bulk_desc, desc);
 352         schedule_work(&desc->bd_queue);
 353
 354         EXIT;
 355 }
 356
 357 static int osc_brw_read(struct lustre_handle *conn, struct lov_stripe_md *lsm,
 358                         obd_count page_count, struct brw_page *pga,
 359                         brw_callback_t callback, struct io_cb_data *data)
 360 {
 361         struct ptlrpc_connection *connection =
 362                 client_conn2cli(conn)->cl_import.imp_connection;
 363         struct ptlrpc_request *request = NULL;
 364         struct ptlrpc_bulk_desc *desc = NULL;
 365         struct ost_body *body;
 366         struct osc_brw_cb_data *cb_data = NULL;
 367         int rc, size[3] = {sizeof(*body)};
 368         void *iooptr, *nioptr;
 369         int mapped = 0;
 370         __u32 xid;
 371         ENTRY;
 372
 373         size[1] = sizeof(struct obd_ioobj);
 374         size[2] = page_count * sizeof(struct niobuf_remote);
 375
 376         request = ptlrpc_prep_req(class_conn2cliimp(conn), OST_READ, 3, size,
 377                                   NULL);
 378         if (!request)
 379                 RETURN(-ENOMEM);
 380
 381         body = lustre_msg_buf(request->rq_reqmsg, 0);
 382
 383         desc = ptlrpc_prep_bulk(connection);
 384         if (!desc)
 385                 GOTO(out_req, rc = -ENOMEM);
 386         desc->bd_portal = OST_BULK_PORTAL;
 387         desc->bd_cb = brw_finish;
 388         OBD_ALLOC(cb_data, sizeof(*cb_data));
 389         if (!cb_data)
 390                 GOTO(out_desc, rc = -ENOMEM);
 391
 392         cb_data->callback = callback;
 393         cb_data->cb_data = data;
 394         data->desc = desc;
 395         desc->bd_cb_data = cb_data;
 396
 397         iooptr = lustre_msg_buf(request->rq_reqmsg, 1);
 398         nioptr = lustre_msg_buf(request->rq_reqmsg, 2);
 399         ost_pack_ioo(&iooptr, lsm, page_count);
 400         /* end almost identical to brw_write case */
 401
 402         spin_lock(&connection->c_lock);
 403         xid = ++connection->c_xid_out;       /* single xid for all pages */
 404         spin_unlock(&connection->c_lock);
 405
 406         for (mapped = 0; mapped < page_count; mapped++) {
 407                 struct ptlrpc_bulk_page *bulk = ptlrpc_prep_bulk_page(desc);
 408                 if (bulk == NULL)
 409                         GOTO(out_unmap, rc = -ENOMEM);
 410
 411                 bulk->bp_xid = xid;           /* single xid for all pages */
 412
 413                 bulk->bp_buf = kmap(pga[mapped].pg);
 414                 bulk->bp_page = pga[mapped].pg;
 415                 bulk->bp_buflen = PAGE_SIZE;
 416                 ost_pack_niobuf(&nioptr, pga[mapped].off, pga[mapped].count,
 417                                 pga[mapped].flag, bulk->bp_xid);
 418         }
 419
 420         /*
 421          * Register the bulk first, because the reply could arrive out of order,
 422          * and we want to be ready for the bulk data.
 423          *
 424          * The reference is released when brw_finish is complete.
 425          *
 426          * On error, we never do the brw_finish, so we handle all decrefs.
 427          */
 428         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_READ_BULK)) {
 429                 CERROR("obd_fail_loc=%x, skipping register_bulk\n",
 430                        OBD_FAIL_OSC_BRW_READ_BULK);
 431         } else {
 432                 rc = ptlrpc_register_bulk(desc);
 433                 if (rc)
 434                         GOTO(out_unmap, rc);
 435         }
 436
 437         request->rq_replen = lustre_msg_size(1, size);
 438         rc = ptlrpc_queue_wait(request);
 439         rc = ptlrpc_check_status(request, rc);
 440
 441         /*
 442          * XXX: If there is an error during the processing of the callback,
 443          *      such as a timeout in a sleep that it performs, brw_finish
 444          *      will never get called, and we'll leak the desc, fail to kunmap
 445          *      things, cats will live with dogs.  One solution would be to
 446          *      export brw_finish as osc_brw_finish, so that the timeout case
 447          *      and its kin could call it for proper cleanup.  An alternative
 448          *      would be for an error return from the callback to cause us to
 449          *      clean up, but that doesn't help the truly async cases (like
 450          *      LOV), which will immediately return from their PHASE_START
 451          *      callback, before any such cleanup-requiring error condition can
 452          *      be detected.
 453          */
 454         if (rc)
 455                 GOTO(out_req, rc);
 456
 457         /* Callbacks cause asynchronous handling. */
 458         rc = callback(data, 0, CB_PHASE_START);
 459
 460 out_req:
 461         ptlrpc_req_finished(request);
 462         RETURN(rc);
 463
 464         /* Clean up on error. */
 465 out_unmap:
 466         while (mapped-- > 0)
 467                 kunmap(pga[mapped].pg);
 468         OBD_FREE(cb_data, sizeof(*cb_data));
 469 out_desc:
 470         ptlrpc_bulk_decref(desc);
 471         goto out_req;
 472 }
 473
 474 static int osc_brw_write(struct lustre_handle *conn, struct lov_stripe_md *md,
 475                          obd_count page_count, struct brw_page *pga,
 476                          brw_callback_t callback, struct io_cb_data *data)
 477 {
 478         struct ptlrpc_connection *connection =
 479                 client_conn2cli(conn)->cl_import.imp_connection;
 480         struct ptlrpc_request *request = NULL;
 481         struct ptlrpc_bulk_desc *desc = NULL;
 482         struct ost_body *body;
 483         struct niobuf_local *local = NULL;
 484         struct niobuf_remote *remote;
 485         struct osc_brw_cb_data *cb_data = NULL;
 486         int rc, j, size[3] = {sizeof(*body)};
 487         void *iooptr, *nioptr;
 488         int mapped = 0;
 489         ENTRY;
 490
 491         size[1] = sizeof(struct obd_ioobj);
 492         size[2] = page_count * sizeof(*remote);
 493
 494         request = ptlrpc_prep_req(class_conn2cliimp(conn), OST_WRITE, 3, size,
 495                                   NULL);
 496         if (!request)
 497                 RETURN(-ENOMEM);
 498
 499         body = lustre_msg_buf(request->rq_reqmsg, 0);
 500
 501         desc = ptlrpc_prep_bulk(connection);
 502         if (!desc)
 503                 GOTO(out_req, rc = -ENOMEM);
 504         desc->bd_portal = OSC_BULK_PORTAL;
 505         desc->bd_cb = brw_finish;
 506         OBD_ALLOC(cb_data, sizeof(*cb_data));
 507         if (!cb_data)
 508                 GOTO(out_desc, rc = -ENOMEM);
 509
 510         cb_data->callback = callback;
 511         cb_data->cb_data = data;
 512         data->desc = desc;
 513         desc->bd_cb_data = cb_data;
 514
 515         iooptr = lustre_msg_buf(request->rq_reqmsg, 1);
 516         nioptr = lustre_msg_buf(request->rq_reqmsg, 2);
 517         ost_pack_ioo(&iooptr, md, page_count);
 518         /* end almost identical to brw_read case */
 519
 520         OBD_ALLOC(local, page_count * sizeof(*local));
 521         if (!local)
 522                 GOTO(out_cb, rc = -ENOMEM);
 523
 524         cb_data->obd_data = local;
 525         cb_data->obd_size = page_count * sizeof(*local);
 526
 527         for (mapped = 0; mapped < page_count; mapped++) {
 528                 local[mapped].addr = kmap(pga[mapped].pg);
 529
 530                 CDEBUG(D_INFO, "kmap(pg) = %p ; pg->flags = %lx ; pg->count = "
 531                        "%d ; page %d of %d\n",
 532                        local[mapped].addr, pga[mapped].pg->flags,
 533                        page_count(pga[mapped].pg),
 534                        mapped, page_count - 1);
 535
 536                 local[mapped].offset = pga[mapped].off;
 537                 local[mapped].len = pga[mapped].count;
 538                 ost_pack_niobuf(&nioptr, pga[mapped].off, pga[mapped].count,
 539                                 pga[mapped].flag, 0);
 540         }
 541
 542         size[1] = page_count * sizeof(*remote);
 543         request->rq_replen = lustre_msg_size(2, size);
 544         rc = ptlrpc_queue_wait(request);
 545         rc = ptlrpc_check_status(request, rc);
 546         if (rc)
 547                 GOTO(out_unmap, rc);
 548
 549         nioptr = lustre_msg_buf(request->rq_repmsg, 1);
 550         if (!nioptr)
 551                 GOTO(out_unmap, rc = -EINVAL);
 552
 553         if (request->rq_repmsg->buflens[1] != size[1]) {
 554                 CERROR("buffer length wrong (%d vs. %d)\n",
 555                        request->rq_repmsg->buflens[1], size[1]);
 556                 GOTO(out_unmap, rc = -EINVAL);
 557         }
 558
 559         for (j = 0; j < page_count; j++) {
 560                 struct ptlrpc_bulk_page *bulk;
 561
 562                 ost_unpack_niobuf(&nioptr, &remote);
 563
 564                 bulk = ptlrpc_prep_bulk_page(desc);
 565                 if (!bulk)
 566                         GOTO(out_unmap, rc = -ENOMEM);
 567
 568                 bulk->bp_buf = (void *)(unsigned long)local[j].addr;
 569                 bulk->bp_buflen = local[j].len;
 570                 bulk->bp_xid = remote->xid;
 571                 bulk->bp_page = pga[j].pg;
 572         }
 573
 574         if (desc->bd_page_count != page_count)
 575                 LBUG();
 576
 577         if (OBD_FAIL_CHECK(OBD_FAIL_OSC_BRW_WRITE_BULK))
 578                 GOTO(out_unmap, rc = 0);
 579
 580         /* Our reference is released when brw_finish is complete. */
 581         rc = ptlrpc_send_bulk(desc);
 582
 583         /* XXX: Mike, same question as in osc_brw_read. */
 584         if (rc)
 585                 GOTO(out_req, rc);
 586
 587         /* Callbacks cause asynchronous handling. */
 588         rc = callback(data, 0, CB_PHASE_START);
 589
 590 out_req:
 591         ptlrpc_req_finished(request);
 592         RETURN(rc);
 593
 594         /* Clean up on error. */
 595 out_unmap:
 596         while (mapped-- > 0)
 597                 kunmap(pga[mapped].pg);
 598
 599         OBD_FREE(local, page_count * sizeof(*local));
 600 out_cb:
 601         OBD_FREE(cb_data, sizeof(*cb_data));
 602 out_desc:
 603         ptlrpc_bulk_decref(desc);
 604         goto out_req;
 605 }
 606
 607 static int osc_brw(int cmd, struct lustre_handle *conn,
 608                    struct lov_stripe_md *md, obd_count page_count,
 609                    struct brw_page *pga, brw_callback_t callback,
 610                    struct io_cb_data *data)
 611 {
 612         ENTRY;
 613
 614         while (page_count) {
 615                 obd_count pages_per_brw;
 616                 int rc;
 617
 618                 if (page_count > PTL_MD_MAX_IOV)
 619                         pages_per_brw = PTL_MD_MAX_IOV;
 620                 else
 621                         pages_per_brw = page_count;
 622
 623                 if (cmd & OBD_BRW_WRITE)
 624                         rc = osc_brw_write(conn, md, pages_per_brw, pga,
 625                                            callback, data);
 626                 else
 627                         rc = osc_brw_read(conn, md, pages_per_brw, pga,
 628                                           callback, data);
 629
 630                 if (rc != 0)
 631                         RETURN(rc);
 632
 633                 page_count -= pages_per_brw;
 634                 pga += pages_per_brw;
 635         }
 636         RETURN(0);
 637 }
 638
 639 static int osc_enqueue(struct lustre_handle *connh, struct lov_stripe_md *lsm,
 640                        struct lustre_handle *parent_lock,
 641                        __u32 type, void *extentp, int extent_len, __u32 mode,
 642                        int *flags, void *callback, void *data, int datalen,
 643                        struct lustre_handle *lockh)
 644 {
 645         __u64 res_id[RES_NAME_SIZE] = { lsm->lsm_object_id };
 646         struct obd_device *obddev = class_conn2obd(connh);
 647         struct ldlm_extent *extent = extentp;
 648         int rc;
 649         ENTRY;
 650
 651         /* Filesystem locks are given a bit of special treatment: if
 652          * this is not a file size lock (which has end == -1), we
 653          * fixup the lock to start and end on page boundaries. */
 654         if (extent->end != OBD_OBJECT_EOF) {
 655                 extent->start &= PAGE_MASK;
 656                 extent->end = (extent->end + PAGE_SIZE - 1) & PAGE_MASK;
 657         }
 658
 659         /* Next, search for already existing extent locks that will cover us */
 660         rc = ldlm_lock_match(obddev->obd_namespace, res_id, type, extent,
 661                              sizeof(extent), mode, lockh);
 662         if (rc == 1)
 663                 /* We already have a lock, and it's referenced */
 664                 RETURN(ELDLM_OK);
 665
 666         /* If we're trying to read, we also search for an existing PW lock.  The
 667          * VFS and page cache already protect us locally, so lots of readers/
 668          * writers can share a single PW lock.
 669          *
 670          * There are problems with conversion deadlocks, so instead of
 671          * converting a read lock to a write lock, we'll just enqueue a new
 672          * one.
 673          *
 674          * At some point we should cancel the read lock instead of making them
 675          * send us a blocking callback, but there are problems with canceling
 676          * locks out from other users right now, too. */
 677
 678         if (mode == LCK_PR) {
 679                 rc = ldlm_lock_match(obddev->obd_namespace, res_id, type,
 680                                      extent, sizeof(extent), LCK_PW, lockh);
 681                 if (rc == 1) {
 682                         /* FIXME: This is not incredibly elegant, but it might
 683                          * be more elegant than adding another parameter to
 684                          * lock_match.  I want a second opinion. */
 685                         ldlm_lock_addref(lockh, LCK_PR);
 686                         ldlm_lock_decref(lockh, LCK_PW);
 687
 688                         RETURN(ELDLM_OK);
 689                 }
 690         }
 691
 692         rc = ldlm_cli_enqueue(connh, NULL, obddev->obd_namespace, parent_lock,
 693                               res_id, type, extent, sizeof(extent), mode, flags,
 694                               ldlm_completion_ast, callback, data, datalen,
 695                               lockh);
 696         RETURN(rc);
 697 }
 698
 699 static int osc_cancel(struct lustre_handle *oconn, struct lov_stripe_md *md,
 700                       __u32 mode, struct lustre_handle *lockh)
 701 {
 702         ENTRY;
 703
 704         ldlm_lock_decref(lockh, mode);
 705
 706         RETURN(0);
 707 }
 708
 709 static int osc_cancel_unused(struct lustre_handle *connh,
 710                              struct lov_stripe_md *lsm, int local)
 711 {
 712         struct obd_device *obddev = class_conn2obd(connh);
 713         __u64 res_id[RES_NAME_SIZE] = { lsm->lsm_object_id };
 714
 715         return ldlm_cli_cancel_unused(obddev->obd_namespace, res_id, local);
 716 }
 717
 718 static int osc_statfs(struct lustre_handle *conn, struct obd_statfs *osfs)
 719 {
 720         struct ptlrpc_request *request;
 721         int rc, size = sizeof(*osfs);
 722         ENTRY;
 723
 724         request = ptlrpc_prep_req(class_conn2cliimp(conn), OST_STATFS, 0, NULL,
 725                                   NULL);
 726         if (!request)
 727                 RETURN(-ENOMEM);
 728
 729         request->rq_replen = lustre_msg_size(1, &size);
 730
 731         rc = ptlrpc_queue_wait(request);
 732         rc = ptlrpc_check_status(request, rc);
 733         if (rc) {
 734                 CERROR("%s failed: rc = %d\n", __FUNCTION__, rc);
 735                 GOTO(out, rc);
 736         }
 737
 738         obd_statfs_unpack(osfs, lustre_msg_buf(request->rq_repmsg, 0));
 739
 740         EXIT;
 741  out:
 742         ptlrpc_req_finished(request);
 743         return rc;
 744 }
 745
 746 static int osc_iocontrol(long cmd, struct lustre_handle *conn, int len,
 747                          void *karg, void *uarg)
 748 {
 749         struct obd_device *obddev = class_conn2obd(conn);
 750         struct obd_ioctl_data *data = karg;
 751         int err = 0;
 752         ENTRY;
 753
 754         switch (cmd) {
 755         case IOC_LDLM_TEST: {
 756                 err = ldlm_test(obddev, conn);
 757                 CERROR("-- done err %d\n", err);
 758                 GOTO(out, err);
 759         }
 760         case IOC_LDLM_REGRESS_START: {
 761                 unsigned int numthreads = 1;
 762                 unsigned int numheld = 10;
 763                 unsigned int numres = 10;
 764                 unsigned int numext = 10;
 765                 char *parse;
 766
 767                 if (data->ioc_inllen1) {
 768                         parse = data->ioc_inlbuf1;
 769                         if (*parse != '\0') {
 770                                 while(isspace(*parse)) parse++;
 771                                 numthreads = simple_strtoul(parse, &parse, 0);
 772                                 while(isspace(*parse)) parse++;
 773                         }
 774                         if (*parse != '\0') {
 775                                 while(isspace(*parse)) parse++;
 776                                 numheld = simple_strtoul(parse, &parse, 0);
 777                                 while(isspace(*parse)) parse++;
 778                         }
 779                         if (*parse != '\0') {
 780                                 while(isspace(*parse)) parse++;
 781                                 numres = simple_strtoul(parse, &parse, 0);
 782                                 while(isspace(*parse)) parse++;
 783                         }
 784                         if (*parse != '\0') {
 785                                 while(isspace(*parse)) parse++;
 786                                 numext = simple_strtoul(parse, &parse, 0);
 787                                 while(isspace(*parse)) parse++;
 788                         }
 789                 }
 790
 791                 err = ldlm_regression_start(obddev, conn, numthreads,
 792                                 numheld, numres, numext);
 793
 794                 CERROR("-- done err %d\n", err);
 795                 GOTO(out, err);
 796         }
 797         case IOC_LDLM_REGRESS_STOP: {
 798                 err = ldlm_regression_stop();
 799                 CERROR("-- done err %d\n", err);
 800                 GOTO(out, err);
 801         }
 802         case IOC_OSC_REGISTER_LOV: {
 803                 if (obddev->u.cli.cl_containing_lov)
 804                         GOTO(out, err = -EALREADY);
 805                 obddev->u.cli.cl_containing_lov = (struct obd_device *)karg;
 806                 GOTO(out, err);
 807         }
 808
 809         default:
 810                 GOTO(out, err = -ENOTTY);
 811         }
 812 out:
 813         return err;
 814 }
 815
 816 struct obd_ops osc_obd_ops = {
 817         o_setup:        client_obd_setup,
 818         o_cleanup:      client_obd_cleanup,
 819         o_statfs:       osc_statfs,
 820         o_create:       osc_create,
 821         o_destroy:      osc_destroy,
 822         o_getattr:      osc_getattr,
 823         o_setattr:      osc_setattr,
 824         o_open:         osc_open,
 825         o_close:        osc_close,
 826         o_connect:      client_obd_connect,
 827         o_disconnect:   client_obd_disconnect,
 828         o_brw:          osc_brw,
 829         o_punch:        osc_punch,
 830         o_enqueue:      osc_enqueue,
 831         o_cancel:       osc_cancel,
 832         o_cancel_unused: osc_cancel_unused,
 833         o_iocontrol:    osc_iocontrol
 834 };
 835
 836 static int __init osc_init(void)
 837 {
 838         return class_register_type(&osc_obd_ops, LUSTRE_OSC_NAME);
 839 }
 840
 841 static void __exit osc_exit(void)
 842 {
 843         class_unregister_type(LUSTRE_OSC_NAME);
 844 }
 845
 846 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 847 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC) v1.0");
 848 MODULE_LICENSE("GPL");
 849
 850 module_init(osc_init);
 851 module_exit(osc_exit);