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  * Copryright (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
  28 static void osc_con2cl(struct lustre_handle *conn, struct ptlrpc_client **cl,
  29                        struct ptlrpc_connection **connection)
  30 {
  31         struct osc_obd *osc = &class_conn2obd(conn)->u.osc;
  32         *cl = osc->osc_client;
  33         *connection = osc->osc_conn;
  34 }
  35
  36 static void osc_con2dlmcl(struct lustre_handle *conn, struct ptlrpc_client **cl,
  37                           struct ptlrpc_connection **connection)
  38 {
  39         struct osc_obd *osc = &class_conn2obd(conn)->u.osc;
  40         *cl = osc->osc_ldlm_client;
  41         *connection = osc->osc_conn;
  42 }
  43
  44 static int osc_connect(struct lustre_handle *conn, struct obd_device *obd)
  45 {
  46         struct osc_obd *osc = &obd->u.osc;
  47         struct obd_import *import;
  48         struct ptlrpc_request *request;
  49         char *tmp = osc->osc_target_uuid;
  50         int rc, size = sizeof(osc->osc_target_uuid);
  51         ENTRY;
  52
  53         OBD_ALLOC(import, sizeof(*import));
  54         if (!import)
  55                 RETURN(-ENOMEM);
  56
  57         MOD_INC_USE_COUNT;
  58         rc = class_connect(conn, obd);
  59         if (rc)
  60                 RETURN(rc);
  61
  62         request = ptlrpc_prep_req(osc->osc_client, osc->osc_conn,
  63                                   OST_CONNECT, 1, &size, &tmp);
  64         if (!request)
  65                 GOTO(out_disco, rc = -ENOMEM);
  66
  67         request->rq_level = LUSTRE_CONN_NEW;
  68         request->rq_replen = lustre_msg_size(0, NULL);
  69
  70         rc = ptlrpc_queue_wait(request);
  71         rc = ptlrpc_check_status(request, rc);
  72         if (rc) {
  73                 CERROR("%s failed: rc = %d\n", __FUNCTION__, rc);
  74                 GOTO(out, rc);
  75         }
  76
  77         /* XXX: Make this a handle */
  78         osc->osc_connh.addr = request->rq_repmsg->addr;
  79         osc->osc_connh.cookie = request->rq_repmsg->cookie;
  80
  81         EXIT;
  82  out:
  83         ptlrpc_free_req(request);
  84  out_disco:
  85         if (rc) {
  86                 class_disconnect(conn);
  87                 MOD_DEC_USE_COUNT;
  88         }
  89         return rc;
  90 }
  91
  92 static int osc_disconnect(struct lustre_handle *conn)
  93 {
  94         struct ptlrpc_request *request;
  95         struct ptlrpc_client *cl;
  96         struct ptlrpc_connection *connection;
  97         struct osc_obd *osc = &class_conn2obd(conn)->u.osc;
  98         int rc;
  99         ENTRY;
 100
 101         osc_con2cl(conn, &cl, &connection);
 102         request = ptlrpc_prep_req2(cl, connection, &osc->osc_connh,
 103                                   OST_DISCONNECT, 0, NULL, NULL);
 104         if (!request)
 105                 RETURN(-ENOMEM);
 106         request->rq_replen = lustre_msg_size(0, NULL);
 107
 108         rc = ptlrpc_queue_wait(request);
 109         if (rc)
 110                 GOTO(out, rc);
 111         rc = class_disconnect(conn);
 112         if (!rc)
 113                 MOD_DEC_USE_COUNT;
 114
 115  out:
 116         ptlrpc_free_req(request);
 117         return rc;
 118 }
 119
 120 static int osc_getattr(struct lustre_handle *conn, struct obdo *oa)
 121 {
 122         struct ptlrpc_request *request;
 123         struct ptlrpc_client *cl;
 124         struct osc_obd *osc = &class_conn2obd(conn)->u.osc;
 125         struct ptlrpc_connection *connection;
 126         struct ost_body *body;
 127         int rc, size = sizeof(*body);
 128         ENTRY;
 129
 130         osc_con2cl(conn, &cl, &connection);
 131         request = ptlrpc_prep_req2(cl, connection, &osc->osc_connh,
 132                                    OST_GETATTR, 1, &size, NULL);
 133         if (!request)
 134                 RETURN(-ENOMEM);
 135
 136         body = lustre_msg_buf(request->rq_reqmsg, 0);
 137         memcpy(&body->oa, oa, sizeof(*oa));
 138         body->oa.o_valid = ~0;
 139
 140         request->rq_replen = lustre_msg_size(1, &size);
 141
 142         rc = ptlrpc_queue_wait(request);
 143         rc = ptlrpc_check_status(request, rc);
 144         if (rc) {
 145                 CERROR("%s failed: rc = %d\n", __FUNCTION__, rc);
 146                 GOTO(out, rc);
 147         }
 148
 149         body = lustre_msg_buf(request->rq_repmsg, 0);
 150         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
 151         if (oa)
 152                 memcpy(oa, &body->oa, sizeof(*oa));
 153
 154         EXIT;
 155  out:
 156         ptlrpc_free_req(request);
 157         return rc;
 158 }
 159
 160 static int osc_open(struct lustre_handle *conn, struct obdo *oa)
 161 {
 162         struct ptlrpc_request *request;
 163         struct ptlrpc_client *cl;
 164         struct osc_obd *osc = &class_conn2obd(conn)->u.osc;
 165         struct ptlrpc_connection *connection;
 166         struct ost_body *body;
 167         int rc, size = sizeof(*body);
 168         ENTRY;
 169
 170         osc_con2cl(conn, &cl, &connection);
 171         request = ptlrpc_prep_req2(cl, connection, &osc->osc_connh,
 172                                    OST_OPEN, 1, &size, NULL);
 173         if (!request)
 174                 RETURN(-ENOMEM);
 175
 176         body = lustre_msg_buf(request->rq_reqmsg, 0);
 177         memcpy(&body->oa, oa, sizeof(*oa));
 178         body->oa.o_valid = (OBD_MD_FLMODE | OBD_MD_FLID);
 179
 180         request->rq_replen = lustre_msg_size(1, &size);
 181
 182         rc = ptlrpc_queue_wait(request);
 183         rc = ptlrpc_check_status(request, rc);
 184         if (rc)
 185                 GOTO(out, rc);
 186
 187         body = lustre_msg_buf(request->rq_repmsg, 0);
 188         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
 189         if (oa)
 190                 memcpy(oa, &body->oa, sizeof(*oa));
 191
 192         EXIT;
 193  out:
 194         ptlrpc_free_req(request);
 195         return rc;
 196 }
 197
 198 static int osc_close(struct lustre_handle *conn, struct obdo *oa)
 199 {
 200         struct ptlrpc_request *request;
 201         struct ptlrpc_client *cl;
 202         struct osc_obd *osc = &class_conn2obd(conn)->u.osc;
 203         struct ptlrpc_connection *connection;
 204         struct ost_body *body;
 205         int rc, size = sizeof(*body);
 206         ENTRY;
 207
 208         osc_con2cl(conn, &cl, &connection);
 209         request = ptlrpc_prep_req2(cl, connection, &osc->osc_connh,
 210                                    OST_CLOSE, 1, &size, NULL);
 211         if (!request)
 212                 RETURN(-ENOMEM);
 213
 214         body = lustre_msg_buf(request->rq_reqmsg, 0);
 215         memcpy(&body->oa, oa, sizeof(*oa));
 216
 217         request->rq_replen = lustre_msg_size(1, &size);
 218
 219         rc = ptlrpc_queue_wait(request);
 220         rc = ptlrpc_check_status(request, rc);
 221         if (rc)
 222                 GOTO(out, rc);
 223
 224         body = lustre_msg_buf(request->rq_repmsg, 0);
 225         CDEBUG(D_INODE, "mode: %o\n", body->oa.o_mode);
 226         if (oa)
 227                 memcpy(oa, &body->oa, sizeof(*oa));
 228
 229         EXIT;
 230  out:
 231         ptlrpc_free_req(request);
 232         return rc;
 233 }
 234
 235 static int osc_setattr(struct lustre_handle *conn, struct obdo *oa)
 236 {
 237         struct ptlrpc_request *request;
 238         struct ptlrpc_client *cl;
 239         struct osc_obd *osc = &class_conn2obd(conn)->u.osc;
 240         struct ptlrpc_connection *connection;
 241         struct ost_body *body;
 242         int rc, size = sizeof(*body);
 243         ENTRY;
 244
 245         osc_con2cl(conn, &cl, &connection);
 246         request = ptlrpc_prep_req2(cl, connection, &osc->osc_connh,
 247                                   OST_SETATTR, 1, &size, NULL);
 248         if (!request)
 249                 RETURN(-ENOMEM);
 250
 251         body = lustre_msg_buf(request->rq_reqmsg, 0);
 252         memcpy(&body->oa, oa, sizeof(*oa));
 253
 254         request->rq_replen = lustre_msg_size(1, &size);
 255
 256         rc = ptlrpc_queue_wait(request);
 257         rc = ptlrpc_check_status(request, rc);
 258         GOTO(out, rc);
 259
 260  out:
 261         ptlrpc_free_req(request);
 262         return rc;
 263 }
 264
 265 static int osc_create(struct lustre_handle *conn, struct obdo *oa)
 266 {
 267         struct ptlrpc_request *request;
 268         struct ptlrpc_client *cl;
 269         struct ptlrpc_connection *connection;
 270         struct osc_obd *osc = &class_conn2obd(conn)->u.osc;
 271         struct ost_body *body;
 272         struct mds_objid *objid;
 273         struct lov_object_id *lov_id;
 274         int rc, size = sizeof(*body);
 275         ENTRY;
 276
 277         if (!oa) {
 278                 CERROR("oa NULL\n");
 279                 RETURN(-EINVAL);
 280         }
 281         osc_con2cl(conn, &cl, &connection);
 282         request = ptlrpc_prep_req2(cl, connection, &osc->osc_connh,
 283                                   OST_CREATE, 1, &size, NULL);
 284         if (!request)
 285                 RETURN(-ENOMEM);
 286
 287         body = lustre_msg_buf(request->rq_reqmsg, 0);
 288         memcpy(&body->oa, oa, sizeof(*oa));
 289
 290         request->rq_replen = lustre_msg_size(1, &size);
 291
 292         rc = ptlrpc_queue_wait(request);
 293         rc = ptlrpc_check_status(request, rc);
 294         if (rc)
 295                 GOTO(out, rc);
 296
 297         body = lustre_msg_buf(request->rq_repmsg, 0);
 298         memcpy(oa, &body->oa, sizeof(*oa));
 299
 300         memset(oa->o_inline, 0, sizeof(oa->o_inline));
 301         objid = (struct mds_objid *)oa->o_inline;
 302         objid->mo_lov_md.lmd_object_id = oa->o_id;
 303         objid->mo_lov_md.lmd_stripe_count = 1;
 304         lov_id = (struct lov_object_id *)(oa->o_inline + sizeof(*objid));
 305         lov_id->l_device_id = 0;
 306         lov_id->l_object_id = oa->o_id;
 307
 308         EXIT;
 309  out:
 310         ptlrpc_free_req(request);
 311         return rc;
 312 }
 313
 314 static int osc_punch(struct lustre_handle *conn, struct obdo *oa, obd_size count,
 315                      obd_off offset)
 316 {
 317         struct ptlrpc_request *request;
 318         struct ptlrpc_client *cl;
 319         struct ptlrpc_connection *connection;
 320         struct osc_obd *osc = &class_conn2obd(conn)->u.osc;
 321         struct ost_body *body;
 322         int rc, size = sizeof(*body);
 323         ENTRY;
 324
 325         if (!oa) {
 326                 CERROR("oa NULL\n");
 327                 RETURN(-EINVAL);
 328         }
 329         osc_con2cl(conn, &cl, &connection);
 330         request = ptlrpc_prep_req2(cl, connection, &osc->osc_connh,
 331                                    OST_PUNCH, 1, &size, NULL);
 332         if (!request)
 333                 RETURN(-ENOMEM);
 334
 335         body = lustre_msg_buf(request->rq_reqmsg, 0);
 336         memcpy(&body->oa, oa, sizeof(*oa));
 337         body->oa.o_blocks = count;
 338         body->oa.o_valid |= OBD_MD_FLBLOCKS;
 339
 340         request->rq_replen = lustre_msg_size(1, &size);
 341
 342         rc = ptlrpc_queue_wait(request);
 343         rc = ptlrpc_check_status(request, rc);
 344         if (rc)
 345                 GOTO(out, rc);
 346
 347         body = lustre_msg_buf(request->rq_repmsg, 0);
 348         memcpy(oa, &body->oa, sizeof(*oa));
 349
 350         EXIT;
 351  out:
 352         ptlrpc_free_req(request);
 353         return rc;
 354 }
 355
 356 static int osc_destroy(struct lustre_handle *conn, struct obdo *oa)
 357 {
 358         struct ptlrpc_request *request;
 359         struct ptlrpc_client *cl;
 360         struct ptlrpc_connection *connection;
 361         struct osc_obd *osc = &class_conn2obd(conn)->u.osc;
 362         struct ost_body *body;
 363         int rc, size = sizeof(*body);
 364         ENTRY;
 365
 366         if (!oa) {
 367                 CERROR("oa NULL\n");
 368                 RETURN(-EINVAL);
 369         }
 370         osc_con2cl(conn, &cl, &connection);
 371         request = ptlrpc_prep_req2(cl, connection, &osc->osc_connh,
 372                                    OST_DESTROY, 1, &size, NULL);
 373         if (!request)
 374                 RETURN(-ENOMEM);
 375
 376         body = lustre_msg_buf(request->rq_reqmsg, 0);
 377         memcpy(&body->oa, oa, sizeof(*oa));
 378         body->oa.o_valid = ~0;
 379
 380         request->rq_replen = lustre_msg_size(1, &size);
 381
 382         rc = ptlrpc_queue_wait(request);
 383         rc = ptlrpc_check_status(request, rc);
 384         if (rc)
 385                 GOTO(out, rc);
 386
 387         body = lustre_msg_buf(request->rq_repmsg, 0);
 388         memcpy(oa, &body->oa, sizeof(*oa));
 389
 390         EXIT;
 391  out:
 392         ptlrpc_free_req(request);
 393         return rc;
 394 }
 395
 396 struct osc_brw_cb_data {
 397         struct page **buf;
 398         struct ptlrpc_request *req;
 399         bulk_callback_t callback;
 400         void *cb_data;
 401 };
 402
 403 static void brw_read_finish(struct ptlrpc_bulk_desc *desc, void *data)
 404 {
 405         struct osc_brw_cb_data *cb_data = data;
 406
 407         if (desc->b_flags & PTL_RPC_FL_INTR)
 408                 CERROR("got signal\n");
 409
 410         (cb_data->callback)(desc, cb_data->cb_data);
 411
 412         ptlrpc_free_bulk(desc);
 413         ptlrpc_free_req(cb_data->req);
 414
 415         OBD_FREE(cb_data, sizeof(*cb_data));
 416 }
 417
 418 static int osc_brw_read(struct lustre_handle *conn, obd_count num_oa,
 419                         struct obdo **oa, obd_count *oa_bufs, struct page **buf,
 420                         obd_size *count, obd_off *offset, obd_flag *flags,
 421                         bulk_callback_t callback)
 422 {
 423         struct ptlrpc_client *cl;
 424         struct ptlrpc_connection *connection;
 425         struct osc_obd *osc = &class_conn2obd(conn)->u.osc;
 426         struct ptlrpc_request *request;
 427         struct ost_body *body;
 428         struct list_head *tmp;
 429         int pages, rc, i, j, size[3] = {sizeof(*body)};
 430         void *ptr1, *ptr2;
 431         struct ptlrpc_bulk_desc *desc;
 432         ENTRY;
 433
 434         size[1] = num_oa * sizeof(struct obd_ioobj);
 435         pages = 0;
 436         for (i = 0; i < num_oa; i++)
 437                 pages += oa_bufs[i];
 438         size[2] = pages * sizeof(struct niobuf_remote);
 439
 440         osc_con2cl(conn, &cl, &connection);
 441         request = ptlrpc_prep_req2(cl, connection, &osc->osc_connh,
 442                                   OST_BRW, 3, size, NULL);
 443         if (!request)
 444                 GOTO(out, rc = -ENOMEM);
 445
 446         body = lustre_msg_buf(request->rq_reqmsg, 0);
 447         body->data = OBD_BRW_READ;
 448
 449         desc = ptlrpc_prep_bulk(connection);
 450         if (!desc)
 451                 GOTO(out2, rc = -ENOMEM);
 452         desc->b_portal = OST_BULK_PORTAL;
 453
 454         ptr1 = lustre_msg_buf(request->rq_reqmsg, 1);
 455         ptr2 = lustre_msg_buf(request->rq_reqmsg, 2);
 456         for (pages = 0, i = 0; i < num_oa; i++) {
 457                 ost_pack_ioo(&ptr1, oa[i], oa_bufs[i]);
 458                 /* FIXME: this inner loop is wrong for multiple OAs */
 459                 for (j = 0; j < oa_bufs[i]; j++, pages++) {
 460                         struct ptlrpc_bulk_page *bulk;
 461                         bulk = ptlrpc_prep_bulk_page(desc);
 462                         if (bulk == NULL)
 463                                 GOTO(out3, rc = -ENOMEM);
 464
 465                         spin_lock(&connection->c_lock);
 466                         bulk->b_xid = ++connection->c_xid_out;
 467                         spin_unlock(&connection->c_lock);
 468
 469                         bulk->b_buf = kmap(buf[pages]);
 470                         bulk->b_page = buf[pages];
 471                         bulk->b_buflen = PAGE_SIZE;
 472                         ost_pack_niobuf(&ptr2, offset[pages], count[pages],
 473                                         flags[pages], bulk->b_xid);
 474                 }
 475         }
 476
 477         rc = ptlrpc_register_bulk(desc);
 478         if (rc)
 479                 GOTO(out3, rc);
 480
 481         request->rq_replen = lustre_msg_size(1, size);
 482         rc = ptlrpc_queue_wait(request);
 483         rc = ptlrpc_check_status(request, rc);
 484         if (rc)
 485                 ptlrpc_abort_bulk(desc);
 486         GOTO(out3, rc);
 487
 488  out3:
 489         list_for_each(tmp, &desc->b_page_list) {
 490                 struct ptlrpc_bulk_page *bulk;
 491                 bulk = list_entry(tmp, struct ptlrpc_bulk_page, b_link);
 492                 if (bulk->b_page != NULL)
 493                         kunmap(bulk->b_page);
 494         }
 495         ptlrpc_free_bulk(desc);
 496  out2:
 497         ptlrpc_free_req(request);
 498  out:
 499         return rc;
 500 }
 501
 502 static void brw_write_finish(struct ptlrpc_bulk_desc *desc, void *data)
 503 {
 504         struct osc_brw_cb_data *cb_data = data;
 505         int i;
 506         ENTRY;
 507
 508         if (desc->b_flags & PTL_RPC_FL_INTR)
 509                 CERROR("got signal\n");
 510
 511         for (i = 0; i < desc->b_page_count; i++)
 512                 kunmap(cb_data->buf[i]);
 513
 514         (cb_data->callback)(desc, cb_data->cb_data);
 515
 516         ptlrpc_free_bulk(desc);
 517         ptlrpc_free_req(cb_data->req);
 518
 519         OBD_FREE(cb_data, sizeof(*cb_data));
 520         EXIT;
 521 }
 522
 523 static int osc_brw_write(struct lustre_handle *conn, obd_count num_oa,
 524                          struct obdo **oa, obd_count *oa_bufs,
 525                          struct page **pagearray, obd_size *count,
 526                          obd_off *offset, obd_flag *flags,
 527                          bulk_callback_t callback)
 528 {
 529         struct ptlrpc_client *cl;
 530         struct ptlrpc_connection *connection;
 531         struct ptlrpc_request *request;
 532         struct ptlrpc_bulk_desc *desc;
 533         struct osc_obd *osc = &class_conn2obd(conn)->u.osc;
 534         struct obd_ioobj ioo;
 535         struct ost_body *body;
 536         struct niobuf_local *local;
 537         struct niobuf_remote *remote;
 538         struct osc_brw_cb_data *cb_data;
 539         long pages;
 540         int rc, i, j, size[3] = {sizeof(*body)};
 541         void *ptr1, *ptr2;
 542         ENTRY;
 543
 544         size[1] = num_oa * sizeof(ioo);
 545         pages = 0;
 546         for (i = 0; i < num_oa; i++)
 547                 pages += oa_bufs[i];
 548         size[2] = pages * sizeof(*remote);
 549
 550         OBD_ALLOC(local, pages * sizeof(*local));
 551         if (local == NULL)
 552                 RETURN(-ENOMEM);
 553
 554         osc_con2cl(conn, &cl, &connection);
 555         request = ptlrpc_prep_req2(cl, connection, &osc->osc_connh,
 556                                   OST_BRW, 3, size, NULL);
 557         if (!request)
 558                 GOTO(out, rc = -ENOMEM);
 559         body = lustre_msg_buf(request->rq_reqmsg, 0);
 560         body->data = OBD_BRW_WRITE;
 561
 562         ptr1 = lustre_msg_buf(request->rq_reqmsg, 1);
 563         ptr2 = lustre_msg_buf(request->rq_reqmsg, 2);
 564         for (pages = 0, i = 0; i < num_oa; i++) {
 565                 ost_pack_ioo(&ptr1, oa[i], oa_bufs[i]);
 566                 for (j = 0; j < oa_bufs[i]; j++, pages++) {
 567                         local[pages].addr = kmap(pagearray[pages]);
 568                         local[pages].offset = offset[pages];
 569                         local[pages].len = count[pages];
 570                         ost_pack_niobuf(&ptr2, offset[pages], count[pages],
 571                                         flags[pages], 0);
 572                 }
 573         }
 574
 575         size[1] = pages * sizeof(struct niobuf_remote);
 576         request->rq_replen = lustre_msg_size(2, size);
 577
 578         rc = ptlrpc_queue_wait(request);
 579         rc = ptlrpc_check_status(request, rc);
 580         if (rc)
 581                 GOTO(out2, rc);
 582
 583         ptr2 = lustre_msg_buf(request->rq_repmsg, 1);
 584         if (ptr2 == NULL)
 585                 GOTO(out2, rc = -EINVAL);
 586
 587         if (request->rq_repmsg->buflens[1] !=
 588             pages * sizeof(struct niobuf_remote)) {
 589                 CERROR("buffer length wrong (%d vs. %ld)\n",
 590                        request->rq_repmsg->buflens[1],
 591                        pages * sizeof(struct niobuf_remote));
 592                 GOTO(out2, rc = -EINVAL);
 593         }
 594
 595         desc = ptlrpc_prep_bulk(connection);
 596         desc->b_portal = OSC_BULK_PORTAL;
 597         if (callback) {
 598                 desc->b_cb = brw_write_finish;
 599                 OBD_ALLOC(cb_data, sizeof(*cb_data));
 600                 cb_data->buf = pagearray;
 601                 cb_data->callback = callback;
 602                 desc->b_cb_data = cb_data;
 603         }
 604
 605         for (pages = 0, i = 0; i < num_oa; i++) {
 606                 for (j = 0; j < oa_bufs[i]; j++, pages++) {
 607                         struct ptlrpc_bulk_page *page;
 608
 609                         ost_unpack_niobuf(&ptr2, &remote);
 610
 611                         page = ptlrpc_prep_bulk_page(desc);
 612                         if (page == NULL)
 613                                 GOTO(out3, rc = -ENOMEM);
 614
 615                         page->b_buf = (void *)(unsigned long)local[pages].addr;
 616                         page->b_buflen = local[pages].len;
 617                         page->b_xid = remote->xid;
 618                 }
 619         }
 620
 621         if (desc->b_page_count != pages)
 622                 LBUG();
 623
 624         rc = ptlrpc_send_bulk(desc);
 625         if (rc)
 626                 GOTO(out3, rc);
 627         if (callback)
 628                 GOTO(out, rc);
 629
 630         /* If there's no callback function, sleep here until complete. */
 631         l_wait_event_killable(desc->b_waitq, ptlrpc_check_bulk_sent(desc));
 632         if (desc->b_flags & PTL_RPC_FL_INTR)
 633                 rc = -EINTR;
 634
 635         GOTO(out3, rc);
 636
 637  out3:
 638         ptlrpc_free_bulk(desc);
 639  out2:
 640         ptlrpc_free_req(request);
 641         for (pages = 0, i = 0; i < num_oa; i++)
 642                 for (j = 0; j < oa_bufs[i]; j++, pages++)
 643                         kunmap(pagearray[pages]);
 644  out:
 645         OBD_FREE(local, pages * sizeof(*local));
 646
 647         return rc;
 648 }
 649
 650 static int osc_brw(int cmd, struct lustre_handle *conn, obd_count num_oa,
 651                    struct obdo **oa, obd_count *oa_bufs, struct page **buf,
 652                    obd_size *count, obd_off *offset, obd_flag *flags,
 653                    void *callback)
 654 {
 655         if (num_oa != 1)
 656                 LBUG();
 657
 658         if (cmd & OBD_BRW_WRITE)
 659                 return osc_brw_write(conn, num_oa, oa, oa_bufs, buf, count,
 660                                      offset, flags, (bulk_callback_t)callback);
 661         else
 662                 return osc_brw_read(conn, num_oa, oa, oa_bufs, buf, count,
 663                                     offset, flags, (bulk_callback_t)callback);
 664 }
 665
 666 static int osc_enqueue(struct lustre_handle *oconn,
 667                        struct lustre_handle *parent_lock, __u64 *res_id,
 668                        __u32 type, void *extentp, int extent_len, __u32 mode,
 669                        int *flags, void *callback, void *data, int datalen,
 670                        struct lustre_handle *lockh)
 671 {
 672         struct obd_device *obddev = class_conn2obd(oconn);
 673         struct osc_obd *osc = &obddev->u.osc;
 674         struct ptlrpc_connection *conn;
 675         struct ptlrpc_client *cl;
 676         struct ldlm_extent *extent = extentp;
 677         int rc;
 678         __u32 mode2;
 679
 680         /* Filesystem locks are given a bit of special treatment: first we
 681          * fixup the lock to start and end on page boundaries. */
 682         extent->start &= PAGE_MASK;
 683         extent->end = (extent->end + PAGE_SIZE - 1) & PAGE_MASK;
 684
 685         /* Next, search for already existing extent locks that will cover us */
 686         osc_con2dlmcl(oconn, &cl, &conn);
 687         rc = ldlm_lock_match(obddev->obd_namespace, res_id, type, extent,
 688                              sizeof(extent), mode, lockh);
 689         if (rc == 1) {
 690                 /* We already have a lock, and it's referenced */
 691                 return 0;
 692         }
 693
 694         /* Next, search for locks that we can upgrade (if we're trying to write)
 695          * or are more than we need (if we're trying to read).  Because the VFS
 696          * and page cache already protect us locally, lots of readers/writers
 697          * can share a single PW lock. */
 698         if (mode == LCK_PW)
 699                 mode2 = LCK_PR;
 700         else
 701                 mode2 = LCK_PW;
 702
 703         rc = ldlm_lock_match(obddev->obd_namespace, res_id, type, extent,
 704                              sizeof(extent), mode2, lockh);
 705         if (rc == 1) {
 706                 int flags;
 707                 /* FIXME: This is not incredibly elegant, but it might
 708                  * be more elegant than adding another parameter to
 709                  * lock_match.  I want a second opinion. */
 710                 ldlm_lock_addref(lockh, mode);
 711                 ldlm_lock_decref(lockh, mode2);
 712
 713                 if (mode == LCK_PR)
 714                         return 0;
 715
 716                 rc = ldlm_cli_convert(cl, lockh, &osc->osc_connh,
 717                                       mode, &flags);
 718                 if (rc)
 719                         LBUG();
 720
 721                 return rc;
 722         }
 723
 724         rc = ldlm_cli_enqueue(cl, conn, &osc->osc_connh,
 725                               NULL, obddev->obd_namespace,
 726                               parent_lock, res_id, type, extent, sizeof(extent),
 727                               mode, flags, callback, data, datalen, lockh);
 728         return rc;
 729 }
 730
 731 static int osc_cancel(struct lustre_handle *oconn, __u32 mode,
 732                       struct lustre_handle *lockh)
 733 {
 734         ENTRY;
 735
 736         ldlm_lock_decref(lockh, mode);
 737
 738         RETURN(0);
 739 }
 740
 741 static int osc_setup(struct obd_device *obddev, obd_count len, void *buf)
 742 {
 743         struct obd_ioctl_data* data = buf;
 744         struct osc_obd *osc = &obddev->u.osc;
 745         char server_uuid[37];
 746         int rc;
 747         ENTRY;
 748
 749         if (data->ioc_inllen1 < 1) {
 750                 CERROR("osc setup requires a TARGET UUID\n");
 751                 RETURN(-EINVAL);
 752         }
 753
 754         if (data->ioc_inllen1 > 37) {
 755                 CERROR("osc TARGET UUID must be less than 38 characters\n");
 756                 RETURN(-EINVAL);
 757         }
 758
 759         if (data->ioc_inllen2 < 1) {
 760                 CERROR("osc setup requires a SERVER UUID\n");
 761                 RETURN(-EINVAL);
 762         }
 763
 764         if (data->ioc_inllen2 > 37) {
 765                 CERROR("osc SERVER UUID must be less than 38 characters\n");
 766                 RETURN(-EINVAL);
 767         }
 768
 769         memcpy(osc->osc_target_uuid, data->ioc_inlbuf1, data->ioc_inllen1);
 770         memcpy(server_uuid, data->ioc_inlbuf2, MIN(data->ioc_inllen2,
 771                                                    sizeof(server_uuid)));
 772
 773         osc->osc_conn = ptlrpc_uuid_to_connection(server_uuid);
 774         if (!osc->osc_conn)
 775                 RETURN(-ENOENT);
 776
 777         obddev->obd_namespace =
 778                 ldlm_namespace_new("osc", LDLM_NAMESPACE_CLIENT);
 779         if (obddev->obd_namespace == NULL)
 780                 GOTO(out_conn, rc = -ENOMEM);
 781
 782         OBD_ALLOC(osc->osc_client, sizeof(*osc->osc_client));
 783         if (osc->osc_client == NULL)
 784                 GOTO(out_ns, rc = -ENOMEM);
 785
 786         OBD_ALLOC(osc->osc_ldlm_client, sizeof(*osc->osc_ldlm_client));
 787         if (osc->osc_ldlm_client == NULL)
 788                 GOTO(out_client, rc = -ENOMEM);
 789
 790         ptlrpc_init_client(NULL, NULL, OST_REQUEST_PORTAL, OSC_REPLY_PORTAL,
 791                            osc->osc_client);
 792         ptlrpc_init_client(NULL, NULL, LDLM_REQUEST_PORTAL, LDLM_REPLY_PORTAL,
 793                            osc->osc_ldlm_client);
 794         osc->osc_client->cli_name = "osc";
 795         osc->osc_ldlm_client->cli_name = "ldlm";
 796
 797         MOD_INC_USE_COUNT;
 798         RETURN(0);
 799
 800  out_client:
 801         OBD_FREE(osc->osc_client, sizeof(*osc->osc_client));
 802  out_ns:
 803         ldlm_namespace_free(obddev->obd_namespace);
 804  out_conn:
 805         ptlrpc_put_connection(osc->osc_conn);
 806         return rc;
 807 }
 808
 809 static int osc_cleanup(struct obd_device * obddev)
 810 {
 811         struct osc_obd *osc = &obddev->u.osc;
 812
 813         ldlm_namespace_free(obddev->obd_namespace);
 814
 815         ptlrpc_cleanup_client(osc->osc_client);
 816         OBD_FREE(osc->osc_client, sizeof(*osc->osc_client));
 817         ptlrpc_cleanup_client(osc->osc_ldlm_client);
 818         OBD_FREE(osc->osc_ldlm_client, sizeof(*osc->osc_ldlm_client));
 819         ptlrpc_put_connection(osc->osc_conn);
 820
 821         MOD_DEC_USE_COUNT;
 822         return 0;
 823 }
 824
 825 static int osc_statfs(struct lustre_handle *conn, struct statfs *sfs)
 826 {
 827         struct ptlrpc_request *request;
 828         struct ptlrpc_client *cl;
 829         struct osc_obd *osc = &class_conn2obd(conn)->u.osc;
 830         struct ptlrpc_connection *connection;
 831         struct obd_statfs *osfs;
 832         int rc, size = sizeof(*osfs);
 833         ENTRY;
 834
 835         osc_con2cl(conn, &cl, &connection);
 836         request = ptlrpc_prep_req2(cl, connection, &osc->osc_connh,
 837                                    OST_STATFS, 0, NULL, NULL);
 838         if (!request)
 839                 RETURN(-ENOMEM);
 840
 841         request->rq_replen = lustre_msg_size(1, &size);
 842
 843         rc = ptlrpc_queue_wait(request);
 844         rc = ptlrpc_check_status(request, rc);
 845         if (rc) {
 846                 CERROR("%s failed: rc = %d\n", __FUNCTION__, rc);
 847                 GOTO(out, rc);
 848         }
 849
 850         osfs = lustre_msg_buf(request->rq_repmsg, 0);
 851         obd_statfs_unpack(osfs, sfs);
 852
 853         EXIT;
 854  out:
 855         ptlrpc_free_req(request);
 856         return rc;
 857 }
 858
 859 struct obd_ops osc_obd_ops = {
 860         o_setup:        osc_setup,
 861         o_cleanup:      osc_cleanup,
 862         o_statfs:       osc_statfs,
 863         o_create:       osc_create,
 864         o_destroy:      osc_destroy,
 865         o_getattr:      osc_getattr,
 866         o_setattr:      osc_setattr,
 867         o_open:         osc_open,
 868         o_close:        osc_close,
 869         o_connect:      osc_connect,
 870         o_disconnect:   osc_disconnect,
 871         o_brw:          osc_brw,
 872         o_punch:        osc_punch,
 873         o_enqueue:      osc_enqueue,
 874         o_cancel:       osc_cancel
 875 };
 876
 877 static int __init osc_init(void)
 878 {
 879         return class_register_type(&osc_obd_ops, LUSTRE_OSC_NAME);
 880 }
 881
 882 static void __exit osc_exit(void)
 883 {
 884         class_unregister_type(LUSTRE_OSC_NAME);
 885 }
 886
 887 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 888 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC) v1.0");
 889 MODULE_LICENSE("GPL");
 890
 891 module_init(osc_init);
 892 module_exit(osc_exit);