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 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
  21 #include <linux/config.h>
  22 #include <linux/module.h>
  23 #include <linux/kernel.h>
  24 #include <linux/mm.h>
  25 #include <linux/string.h>
  26 #include <linux/stat.h>
  27 #include <linux/errno.h>
  28 #include <linux/locks.h>
  29 #include <linux/unistd.h>
  30
  31 #include <asm/system.h>
  32 #include <asm/uaccess.h>
  33
  34 #include <linux/fs.h>
  35 #include <linux/stat.h>
  36 #include <asm/uaccess.h>
  37 #include <asm/segment.h>
  38 #include <linux/miscdevice.h>
  39
  40 #define DEBUG_SUBSYSTEM S_OSC
  41
  42 #include <linux/obd_support.h>
  43 #include <linux/obd_class.h>
  44 #include <linux/lustre_lib.h>
  45 #include <linux/lustre_idl.h>
  46
  47 struct ptlrpc_client *osc_con2cl(struct obd_conn *conn)
  48 {
  49         struct osc_obd *osc = &conn->oc_dev->u.osc;
  50         return &osc->osc_peer;
  51
  52 }
  53
  54 static int osc_connect(struct obd_conn *conn)
  55 {
  56         struct ptlrpc_request *request;
  57         struct ptlrpc_client *peer = osc_con2cl(conn);
  58         int rc;
  59         ENTRY;
  60
  61         request = ptlrpc_prep_req(peer, OST_CONNECT, 0, NULL, 0, NULL);
  62         if (!request) {
  63                 CERROR("cannot pack req!\n");
  64                 return -ENOMEM;
  65         }
  66
  67         request->rq_replen =
  68                 sizeof(struct ptlrep_hdr) + sizeof(struct ost_rep);
  69
  70         rc = ptlrpc_queue_wait(peer, request);
  71         if (rc) {
  72                 EXIT;
  73                 goto out;
  74         }
  75
  76         CDEBUG(D_INODE, "received connid %d\n", request->rq_rep.ost->connid);
  77
  78         conn->oc_id = request->rq_rep.ost->connid;
  79  out:
  80         ptlrpc_free_req(request);
  81         EXIT;
  82         return rc;
  83 }
  84
  85 static int osc_disconnect(struct obd_conn *conn)
  86 {
  87         struct ptlrpc_request *request;
  88         struct ptlrpc_client *peer = osc_con2cl(conn);
  89         int rc;
  90         ENTRY;
  91
  92         request = ptlrpc_prep_req(peer, OST_DISCONNECT, 0, NULL, 0, NULL);
  93         if (!request) {
  94                 CERROR("cannot pack req!\n");
  95                 return -ENOMEM;
  96         }
  97         request->rq_req.ost->connid = conn->oc_id;
  98         request->rq_replen =
  99                 sizeof(struct ptlrep_hdr) + sizeof(struct ost_rep);
 100
 101         rc = ptlrpc_queue_wait(peer, request);
 102         if (rc) {
 103                 EXIT;
 104                 goto out;
 105         }
 106  out:
 107         ptlrpc_free_req(request);
 108         EXIT;
 109         return rc;
 110 }
 111
 112
 113 static int osc_getattr(struct obd_conn *conn, struct obdo *oa)
 114 {
 115         struct ptlrpc_request *request;
 116         struct ptlrpc_client *peer = osc_con2cl(conn);
 117         int rc;
 118
 119         request = ptlrpc_prep_req(peer, OST_GETATTR, 0, NULL, 0, NULL);
 120         if (!request) {
 121                 CERROR("cannot pack req!\n");
 122                 return -ENOMEM;
 123         }
 124
 125         memcpy(&request->rq_req.ost->oa, oa, sizeof(*oa));
 126         request->rq_req.ost->oa.o_valid = ~0;
 127         request->rq_replen =
 128                 sizeof(struct ptlrep_hdr) + sizeof(struct ost_rep);
 129
 130         rc = ptlrpc_queue_wait(peer, request);
 131         if (rc) {
 132                 EXIT;
 133                 goto out;
 134         }
 135
 136         CDEBUG(D_INODE, "mode: %o\n", request->rq_rep.ost->oa.o_mode);
 137         if (oa) {
 138                 memcpy(oa, &request->rq_rep.ost->oa, sizeof(*oa));
 139         }
 140
 141  out:
 142         ptlrpc_free_req(request);
 143         return 0;
 144 }
 145
 146 static int osc_setattr(struct obd_conn *conn, struct obdo *oa)
 147 {
 148         struct ptlrpc_request *request;
 149         struct ptlrpc_client *peer = osc_con2cl(conn);
 150         int rc;
 151
 152         request = ptlrpc_prep_req(peer, OST_SETATTR, 0, NULL, 0, NULL);
 153         if (!request) {
 154                 CERROR("cannot pack req!\n");
 155                 return -ENOMEM;
 156         }
 157
 158         memcpy(&request->rq_req.ost->oa, oa, sizeof(*oa));
 159         request->rq_replen =
 160                 sizeof(struct ptlrep_hdr) + sizeof(struct ost_rep);
 161
 162         rc = ptlrpc_queue_wait(peer, request);
 163         if (rc) {
 164                 EXIT;
 165                 goto out;
 166         }
 167
 168  out:
 169         ptlrpc_free_req(request);
 170         return 0;
 171 }
 172
 173 static int osc_create(struct obd_conn *conn, struct obdo *oa)
 174 {
 175         struct ptlrpc_request *request;
 176         struct ptlrpc_client *peer = osc_con2cl(conn);
 177         int rc;
 178
 179         if (!oa) {
 180                 CERROR("oa NULL\n");
 181         }
 182         request = ptlrpc_prep_req(peer, OST_CREATE, 0, NULL, 0, NULL);
 183         if (!request) {
 184                 CERROR("cannot pack req!\n");
 185                 return -ENOMEM;
 186         }
 187
 188         memcpy(&request->rq_req.ost->oa, oa, sizeof(*oa));
 189         request->rq_req.ost->connid = conn->oc_id;
 190         request->rq_req.ost->oa.o_valid = ~0;
 191         request->rq_replen =
 192                 sizeof(struct ptlrep_hdr) + sizeof(struct ost_rep);
 193
 194         rc = ptlrpc_queue_wait(peer, request);
 195         if (rc) {
 196                 EXIT;
 197                 goto out;
 198         }
 199         memcpy(oa, &request->rq_rep.ost->oa, sizeof(*oa));
 200
 201  out:
 202         ptlrpc_free_req(request);
 203         return 0;
 204 }
 205
 206 static int osc_punch(struct obd_conn *conn, struct obdo *oa, obd_size count,
 207                      obd_off offset)
 208 {
 209         struct ptlrpc_request *request;
 210         struct ptlrpc_client *peer = osc_con2cl(conn);
 211         int rc;
 212
 213         if (!oa) {
 214                 CERROR("oa NULL\n");
 215         }
 216         request = ptlrpc_prep_req(peer, OST_PUNCH, 0, NULL, 0, NULL);
 217         if (!request) {
 218                 CERROR("cannot pack req!\n");
 219                 return -ENOMEM;
 220         }
 221
 222         memcpy(&request->rq_req.ost->oa, oa, sizeof(*oa));
 223         request->rq_req.ost->oa.o_valid = ~0;
 224         request->rq_req.ost->oa.o_size = offset;
 225         request->rq_req.ost->oa.o_blocks = count;
 226         request->rq_replen =
 227                 sizeof(struct ptlrep_hdr) + sizeof(struct ost_rep);
 228
 229         rc = ptlrpc_queue_wait(peer, request);
 230         if (rc) {
 231                 EXIT;
 232                 goto out;
 233         }
 234         memcpy(oa, &request->rq_rep.ost->oa, sizeof(*oa));
 235
 236  out:
 237         ptlrpc_free_req(request);
 238         return 0;
 239 }
 240
 241 static int osc_destroy(struct obd_conn *conn, struct obdo *oa)
 242 {
 243         struct ptlrpc_request *request;
 244         struct ptlrpc_client *peer = osc_con2cl(conn);
 245         int rc;
 246
 247         if (!oa) {
 248                 CERROR("oa NULL\n");
 249         }
 250         request = ptlrpc_prep_req(peer, OST_DESTROY, 0, NULL, 0, NULL);
 251         if (!request) {
 252                 CERROR("cannot pack req!\n");
 253                 return -ENOMEM;
 254         }
 255
 256         memcpy(&request->rq_req.ost->oa, oa, sizeof(*oa));
 257         request->rq_req.ost->oa.o_valid = ~0;
 258         request->rq_replen =
 259                 sizeof(struct ptlrep_hdr) + sizeof(struct ost_rep);
 260
 261         rc = ptlrpc_queue_wait(peer, request);
 262         if (rc) {
 263                 EXIT;
 264                 goto out;
 265         }
 266         memcpy(oa, &request->rq_rep.ost->oa, sizeof(*oa));
 267
 268  out:
 269         ptlrpc_free_req(request);
 270         return 0;
 271 }
 272
 273 int osc_sendpage(struct obd_conn *conn, struct ptlrpc_request *req,
 274                  struct niobuf *dst, struct niobuf *src)
 275 {
 276         struct ptlrpc_client *cl = osc_con2cl(conn);
 277
 278         if (cl->cli_obd) {
 279                 /* local sendpage */
 280                 memcpy((char *)(unsigned long)dst->addr,
 281                        (char *)(unsigned long)src->addr, src->len);
 282         } else {
 283                 struct ptlrpc_bulk_desc *bulk;
 284                 int rc;
 285
 286                 bulk = ptlrpc_prep_bulk(&cl->cli_server);
 287                 if (bulk == NULL)
 288                         return -ENOMEM;
 289
 290                 bulk->b_buf = (void *)(unsigned long)src->addr;
 291                 bulk->b_buflen = src->len;
 292                 bulk->b_xid = dst->xid;
 293                 rc = ptlrpc_send_bulk(bulk, OSC_BULK_PORTAL);
 294                 if (rc != 0) {
 295                         CERROR("send_bulk failed: %d\n", rc);
 296                         BUG();
 297                         return rc;
 298                 }
 299                 wait_event_interruptible(bulk->b_waitq,
 300                                          ptlrpc_check_bulk_sent(bulk));
 301
 302                 if (bulk->b_flags == PTL_RPC_INTR) {
 303                         EXIT;
 304                         /* FIXME: hey hey, we leak here. */
 305                         return -EINTR;
 306                 }
 307
 308                 OBD_FREE(bulk, sizeof(*bulk));
 309         }
 310
 311         return 0;
 312 }
 313
 314 int osc_brw_read(struct obd_conn *conn, obd_count num_oa, struct obdo **oa,
 315                  obd_count *oa_bufs, struct page **buf, obd_size *count,
 316                  obd_off *offset, obd_flag *flags)
 317 {
 318         struct ptlrpc_client *cl = osc_con2cl(conn);
 319         struct ptlrpc_request *request;
 320         int pages;
 321         int rc;
 322         struct obd_ioobj ioo;
 323         struct niobuf src;
 324         int size1, size2 = 0;
 325         void *ptr1, *ptr2;
 326         int i, j, n;
 327         struct ptlrpc_bulk_desc **bulk;
 328
 329         size1 = num_oa * sizeof(ioo);
 330         pages = 0;
 331         for (i = 0; i < num_oa; i++) {
 332                 size2 += oa_bufs[i] * sizeof(src);
 333                 pages += oa_bufs[i];
 334         }
 335
 336         /* We actually pack a _third_ buffer, with XIDs for bulk pages */
 337         size2 += pages * sizeof(__u32);
 338         request = ptlrpc_prep_req(cl, OST_BRW, size1, NULL, size2, NULL);
 339         if (!request) {
 340                 CERROR("cannot pack req!\n");
 341                 return -ENOMEM;
 342         }
 343         request->rq_req.ost->cmd = OBD_BRW_READ;
 344
 345         OBD_ALLOC(bulk, pages * sizeof(struct ptlrpc_bulk_desc *));
 346         if (bulk == NULL) {
 347                 CERROR("cannot alloc bulk desc vector\n");
 348                 return -ENOMEM;
 349         }
 350         memset(bulk, 0, pages * sizeof(struct ptlrpc_bulk_desc *));
 351
 352         n = 0;
 353         ptr1 = ost_req_buf1(request->rq_req.ost);
 354         ptr2 = ost_req_buf2(request->rq_req.ost);
 355         for (i = 0; i < num_oa; i++) {
 356                 ost_pack_ioo(&ptr1, oa[i], oa_bufs[i]);
 357                 for (j = 0; j < oa_bufs[i]; j++) {
 358                         bulk[n] = ptlrpc_prep_bulk(&cl->cli_server);
 359                         if (bulk[n] == NULL) {
 360                                 CERROR("cannot alloc bulk desc\n");
 361                                 rc = -ENOMEM;
 362                                 goto out;
 363                         }
 364
 365                         spin_lock(&cl->cli_lock);
 366                         bulk[n]->b_xid = cl->cli_xid++;
 367                         spin_unlock(&cl->cli_lock);
 368                         bulk[n]->b_buf = kmap(buf[n]);
 369                         bulk[n]->b_buflen = PAGE_SIZE;
 370                         bulk[n]->b_portal = OST_BULK_PORTAL;
 371                         ost_pack_niobuf(&ptr2, bulk[n]->b_buf, offset[n],
 372                                         count[n], flags[n], bulk[n]->b_xid);
 373                         n++;
 374                 }
 375         }
 376
 377         /* This is kinda silly--put the XIDs in the "third" buffer. */
 378         for (n = 0; n < pages; n++) {
 379                 *(__u32 *)ptr2 = bulk[n]->b_xid;
 380                 ptr2 = (char *)ptr2 + sizeof(__u32);
 381
 382                 rc = ptlrpc_register_bulk(bulk[n]);
 383                 if (rc)
 384                         goto out;
 385         }
 386
 387         request->rq_replen = sizeof(struct ptlrep_hdr) + sizeof(struct ost_rep);
 388         rc = ptlrpc_queue_wait(cl, request);
 389
 390  out:
 391         /* FIXME: if we've called ptlrpc_wait_bulk but rc != 0, we need to
 392          * abort those bulk listeners. */
 393
 394         if (request->rq_rephdr)
 395                 OBD_FREE(request->rq_rephdr, request->rq_replen);
 396         n = 0;
 397         for (i = 0; i < num_oa; i++) {
 398                 for (j = 0; j < oa_bufs[i]; j++) {
 399                         if (bulk[n] == NULL)
 400                                 continue;
 401                         kunmap(bulk[n]->b_buf);
 402                         OBD_FREE(bulk[n], sizeof(struct ptlrpc_bulk_desc));
 403                         n++;
 404                 }
 405         }
 406
 407         OBD_FREE(bulk, pages * sizeof(struct ptlrpc_bulk_desc *));
 408         ptlrpc_free_req(request);
 409         return rc;
 410 }
 411
 412 int osc_brw_write(struct obd_conn *conn, obd_count num_oa, struct obdo **oa,
 413                   obd_count *oa_bufs, struct page **buf, obd_size *count,
 414                   obd_off *offset, obd_flag *flags)
 415 {
 416         struct ptlrpc_client *cl = osc_con2cl(conn);
 417         struct ptlrpc_request *request, *req2 = NULL;
 418         struct obd_ioobj ioo;
 419         struct niobuf src;
 420         int pages, rc, i, j, n, size1, size2 = 0;
 421         void *ptr1, *ptr2, *reqbuf;
 422
 423         size1 = num_oa * sizeof(ioo);
 424         pages = 0;
 425         for (i = 0; i < num_oa; i++) {
 426                 size2 += oa_bufs[i] * sizeof(src);
 427                 pages += oa_bufs[i];
 428         }
 429
 430         request = ptlrpc_prep_req(cl, OST_BRW, size1, NULL, size2, NULL);
 431         if (!request) {
 432                 CERROR("cannot pack req!\n");
 433                 return -ENOMEM;
 434         }
 435         OBD_ALLOC(reqbuf, request->rq_reqlen);
 436         if (reqbuf == NULL) {
 437                 CERROR("cannot make duplicate buffer\n");
 438                 return -ENOMEM;
 439         }
 440         request->rq_req.ost->cmd = OBD_BRW_WRITE;
 441
 442         n = 0;
 443         ptr1 = ost_req_buf1(request->rq_req.ost);
 444         ptr2 = ost_req_buf2(request->rq_req.ost);
 445         for (i = 0; i < num_oa; i++) {
 446                 ost_pack_ioo(&ptr1, oa[i], oa_bufs[i]);
 447                 for (j = 0; j < oa_bufs[i]; j++) {
 448                         ost_pack_niobuf(&ptr2, kmap(buf[n]), offset[n],
 449                                         count[n], flags[n], 0);
 450                         n++;
 451                 }
 452         }
 453
 454         memcpy(reqbuf, request->rq_reqbuf, request->rq_reqlen);
 455         request->rq_replen = sizeof(struct ptlrep_hdr) +
 456                 sizeof(struct ost_rep) + pages * sizeof(struct niobuf);
 457         rc = ptlrpc_queue_wait(cl, request);
 458         if (rc) {
 459                 EXIT;
 460                 goto out;
 461         }
 462
 463         ptr2 = ost_rep_buf2(request->rq_rep.ost);
 464         if (request->rq_rep.ost->buflen2 != n * sizeof(struct niobuf)) {
 465                 CERROR("buffer length wrong (%d vs. %d)\n",
 466                        request->rq_rep.ost->buflen2, n * sizeof(struct niobuf));
 467                 EXIT;
 468                 goto out;
 469         }
 470
 471         n = 0;
 472         for (i = 0; i < num_oa; i++) {
 473                 for (j = 0; j < oa_bufs[i]; j++) {
 474                         struct niobuf *dst;
 475                         src.addr = (__u64)(unsigned long)buf[n];
 476                         src.len = count[n];
 477                         ost_unpack_niobuf(&ptr2, &dst);
 478                         osc_sendpage(conn, request, dst, &src);
 479                         n++;
 480                 }
 481         }
 482
 483         req2 = ptlrpc_prep_req(cl, OST_BRW_COMPLETE, size1, ptr1,
 484                                request->rq_rep.ost->buflen2, ptr2);
 485         ptr2 = ost_rep_buf2(request->rq_rep.ost);
 486         if (!req2) {
 487                 CERROR("cannot pack second request!\n");
 488                 return -ENOMEM;
 489         }
 490
 491         req2->rq_reqhdr->opc = OST_BRW_COMPLETE;
 492         req2->rq_replen = sizeof(struct ptlrep_hdr) + sizeof(struct ost_rep);
 493         rc = ptlrpc_queue_wait(cl, req2);
 494         if (rc) {
 495                 EXIT;
 496                 goto out;
 497         }
 498
 499  out:
 500         if (request->rq_rephdr)
 501                 OBD_FREE(request->rq_rephdr, request->rq_replen);
 502         if (req2 && req2->rq_rephdr)
 503                 OBD_FREE(req2->rq_rephdr, req2->rq_replen);
 504         n = 0;
 505         for (i = 0; i < num_oa; i++) {
 506                 for (j = 0; j < oa_bufs[i]; j++) {
 507                         kunmap(buf[n]);
 508                         n++;
 509                 }
 510         }
 511
 512         if (req2)
 513                 ptlrpc_free_req(req2);
 514         ptlrpc_free_req(request);
 515         return 0;
 516 }
 517
 518 int osc_brw(int rw, struct obd_conn *conn, obd_count num_oa,
 519               struct obdo **oa, obd_count *oa_bufs, struct page **buf,
 520               obd_size *count, obd_off *offset, obd_flag *flags)
 521 {
 522         if (rw == OBD_BRW_READ) {
 523                 return osc_brw_read(conn, num_oa, oa, oa_bufs, buf, count,
 524                                     offset, flags);
 525         } else {
 526                 return osc_brw_write(conn, num_oa, oa, oa_bufs, buf, count,
 527                                      offset, flags);
 528         }
 529 }
 530
 531 /* mount the file system (secretly) */
 532 static int osc_setup(struct obd_device *obddev, obd_count len,
 533                         void *buf)
 534
 535 {
 536         struct osc_obd *osc = &obddev->u.osc;
 537         struct obd_ioctl_data *data = (struct obd_ioctl_data *)buf;
 538         int rc;
 539         int dev = data->ioc_dev;
 540         ENTRY;
 541
 542         rc = ptlrpc_connect_client(dev, "ost",
 543                                    OST_REQUEST_PORTAL,
 544                                    OSC_REPLY_PORTAL,
 545                                    ost_pack_req,
 546                                    ost_unpack_rep,
 547                                    &osc->osc_peer);
 548
 549         MOD_INC_USE_COUNT;
 550         EXIT;
 551         return rc;
 552 }
 553
 554 static int osc_cleanup(struct obd_device * obddev)
 555 {
 556         MOD_DEC_USE_COUNT;
 557         return 0;
 558 }
 559
 560 struct obd_ops osc_obd_ops = {
 561         o_setup:   osc_setup,
 562         o_cleanup: osc_cleanup,
 563         o_create: osc_create,
 564         o_destroy: osc_destroy,
 565         o_getattr: osc_getattr,
 566         o_setattr: osc_setattr,
 567         o_connect: osc_connect,
 568         o_disconnect: osc_disconnect,
 569         o_brw: osc_brw,
 570         o_punch: osc_punch
 571 };
 572
 573 static int __init osc_init(void)
 574 {
 575         obd_register_type(&osc_obd_ops, LUSTRE_OSC_NAME);
 576         return 0;
 577 }
 578
 579 static void __exit osc_exit(void)
 580 {
 581         obd_unregister_type(LUSTRE_OSC_NAME);
 582 }
 583
 584 MODULE_AUTHOR("Peter J. Braam <braam@clusterfs.com>");
 585 MODULE_DESCRIPTION("Lustre Object Storage Client (OSC) v1.0");
 586 MODULE_LICENSE("GPL");
 587
 588 module_init(osc_init);
 589 module_exit(osc_exit);