lustre/ost/ost_handler.c

   1 /*
   2  *  ost/ost_handler.c
   3  *  Storage Target Handling functions
   4  *
   5  *  Lustre Object Server Module (OST)
   6  *
   7  *  Copyright (C) 2001, 2002 Cluster File Systems, Inc.
   8  *
   9  *  This code is issued under the GNU General Public License.
  10  *  See the file COPYING in this distribution
  11  *
  12  *  by Peter Braam <braam@clusterfs.com>
  13  *
  14  *  This server is single threaded at present (but can easily be multi
  15  *  threaded). For testing and management it is treated as an
  16  *  obd_device, although it does not export a full OBD method table
  17  *  (the requests are coming in over the wire, so object target
  18  *  modules do not have a full method table.)
  19  *
  20  */
  21
  22 #define EXPORT_SYMTAB
  23
  24 #include <linux/version.h>
  25 #include <linux/module.h>
  26 #include <linux/fs.h>
  27 #include <linux/stat.h>
  28 #include <linux/locks.h>
  29 #include <linux/ext2_fs.h>
  30 #include <linux/quotaops.h>
  31 #include <asm/unistd.h>
  32
  33 #define DEBUG_SUBSYSTEM S_OST
  34
  35 #include <linux/obd_support.h>
  36 #include <linux/obd.h>
  37 #include <linux/obd_class.h>
  38 #include <linux/lustre_lib.h>
  39 #include <linux/lustre_idl.h>
  40 #include <linux/lustre_mds.h>
  41 #include <linux/obd_class.h>
  42
  43 // for testing
  44 static int ost_queue_req(struct obd_device *obddev, struct ptlrpc_request *req)
  45 {
  46         struct ptlrpc_request *srv_req;
  47         struct ost_obd *ost = &obddev->u.ost;
  48
  49         if (!ost) {
  50                 EXIT;
  51                 return -1;
  52         }
  53
  54         OBD_ALLOC(srv_req, sizeof(*srv_req));
  55         if (!srv_req) {
  56                 EXIT;
  57                 return -ENOMEM;
  58         }
  59
  60         CDEBUG(0, "---> OST at %d %p, incoming req %p, srv_req %p\n",
  61                __LINE__, ost, req, srv_req);
  62
  63         memset(srv_req, 0, sizeof(*req));
  64
  65         /* move the request buffer */
  66         srv_req->rq_reqbuf = req->rq_reqbuf;
  67         srv_req->rq_reqlen    = req->rq_reqlen;
  68         srv_req->rq_ost = ost;
  69
  70         /* remember where it came from */
  71         srv_req->rq_reply_handle = req;
  72
  73         list_add(&srv_req->rq_list, &ost->ost_reqs);
  74         wake_up(&ost->ost_waitq);
  75         return 0;
  76 }
  77
  78 int ost_reply(struct obd_device *obddev, struct ptlrpc_request *req)
  79 {
  80         struct ptlrpc_request *clnt_req = req->rq_reply_handle;
  81
  82         ENTRY;
  83
  84         if (req->rq_ost->ost_service != NULL) {
  85                 /* This is a request that came from the network via portals. */
  86
  87                 /* FIXME: we need to increment the count of handled events */
  88                 ptl_send_buf(req, &req->rq_peer, OST_REPLY_PORTAL, 0);
  89         } else {
  90                 /* This is a local request that came from another thread. */
  91
  92                 /* move the reply to the client */
  93                 clnt_req->rq_replen = req->rq_replen;
  94                 clnt_req->rq_repbuf = req->rq_repbuf;
  95                 req->rq_repbuf = NULL;
  96                 req->rq_replen = 0;
  97
  98                 /* free the request buffer */
  99                 OBD_FREE(req->rq_reqbuf, req->rq_reqlen);
 100                 req->rq_reqbuf = NULL;
 101
 102                 /* wake up the client */
 103                 wake_up_interruptible(&clnt_req->rq_wait_for_rep);
 104         }
 105
 106         EXIT;
 107         return 0;
 108 }
 109
 110 int ost_error(struct obd_device *obddev, struct ptlrpc_request *req)
 111 {
 112         struct ptlrep_hdr *hdr;
 113
 114         ENTRY;
 115
 116         OBD_ALLOC(hdr, sizeof(*hdr));
 117         if (!hdr) {
 118                 EXIT;
 119                 return -ENOMEM;
 120         }
 121
 122         memset(hdr, 0, sizeof(*hdr));
 123
 124         hdr->seqno = req->rq_reqhdr->seqno;
 125         hdr->status = req->rq_status;
 126         hdr->type = OST_TYPE_ERR;
 127
 128         req->rq_repbuf = (char *)hdr;
 129         req->rq_replen = sizeof(*hdr);
 130
 131         EXIT;
 132         return ost_reply(obddev, req);
 133 }
 134
 135 static int ost_destroy(struct ost_obd *ost, struct ptlrpc_request *req)
 136 {
 137         struct obd_conn conn;
 138         int rc;
 139
 140         ENTRY;
 141
 142         conn.oc_id = req->rq_req.ost->connid;
 143         conn.oc_dev = ost->ost_tgt;
 144
 145         rc = ost_pack_rep(NULL, 0, NULL, 0, &req->rq_rephdr, &req->rq_rep.ost,
 146                           &req->rq_replen, &req->rq_repbuf);
 147         if (rc) {
 148                 CERROR("cannot pack reply\n");
 149                 return rc;
 150         }
 151
 152         req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_destroy
 153                 (&conn, &req->rq_req.ost->oa);
 154
 155         EXIT;
 156         return 0;
 157 }
 158
 159 static int ost_getattr(struct ost_obd *ost, struct ptlrpc_request *req)
 160 {
 161         struct obd_conn conn;
 162         int rc;
 163
 164         ENTRY;
 165
 166         conn.oc_id = req->rq_req.ost->connid;
 167         conn.oc_dev = ost->ost_tgt;
 168
 169         rc = ost_pack_rep(NULL, 0, NULL, 0, &req->rq_rephdr, &req->rq_rep.ost,
 170                           &req->rq_replen, &req->rq_repbuf);
 171         if (rc) {
 172                 CERROR("cannot pack reply\n");
 173                 return rc;
 174         }
 175         req->rq_rep.ost->oa.o_id = req->rq_req.ost->oa.o_id;
 176         req->rq_rep.ost->oa.o_valid = req->rq_req.ost->oa.o_valid;
 177
 178         req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_getattr
 179                 (&conn, &req->rq_rep.ost->oa);
 180
 181         EXIT;
 182         return 0;
 183 }
 184
 185 static int ost_create(struct ost_obd *ost, struct ptlrpc_request *req)
 186 {
 187         struct obd_conn conn;
 188         int rc;
 189
 190         ENTRY;
 191
 192         conn.oc_id = req->rq_req.ost->connid;
 193         conn.oc_dev = ost->ost_tgt;
 194
 195         rc = ost_pack_rep(NULL, 0, NULL, 0, &req->rq_rephdr, &req->rq_rep.ost,
 196                           &req->rq_replen, &req->rq_repbuf);
 197         if (rc) {
 198                 CERROR("cannot pack reply\n");
 199                 return rc;
 200         }
 201
 202         memcpy(&req->rq_rep.ost->oa, &req->rq_req.ost->oa, sizeof(req->rq_req.ost->oa));
 203
 204         req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_create
 205                 (&conn, &req->rq_rep.ost->oa);
 206
 207         EXIT;
 208         return 0;
 209 }
 210
 211
 212 static int ost_setattr(struct ost_obd *ost, struct ptlrpc_request *req)
 213 {
 214         struct obd_conn conn;
 215         int rc;
 216
 217         ENTRY;
 218
 219         conn.oc_id = req->rq_req.ost->connid;
 220         conn.oc_dev = ost->ost_tgt;
 221
 222         rc = ost_pack_rep(NULL, 0, NULL, 0, &req->rq_rephdr, &req->rq_rep.ost,
 223                           &req->rq_replen, &req->rq_repbuf);
 224         if (rc) {
 225                 CERROR("cannot pack reply\n");
 226                 return rc;
 227         }
 228
 229         memcpy(&req->rq_rep.ost->oa, &req->rq_req.ost->oa,
 230                sizeof(req->rq_req.ost->oa));
 231
 232         req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_setattr
 233                 (&conn, &req->rq_rep.ost->oa);
 234
 235         EXIT;
 236         return 0;
 237 }
 238
 239 static int ost_connect(struct ost_obd *ost, struct ptlrpc_request *req)
 240 {
 241         struct obd_conn conn;
 242         int rc;
 243
 244         ENTRY;
 245
 246         conn.oc_dev = ost->ost_tgt;
 247
 248         rc = ost_pack_rep(NULL, 0, NULL, 0, &req->rq_rephdr, &req->rq_rep.ost,
 249                           &req->rq_replen, &req->rq_repbuf);
 250         if (rc) {
 251                 CERROR("cannot pack reply\n");
 252                 return rc;
 253         }
 254
 255         req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_connect(&conn);
 256
 257         CDEBUG(0, "rep buffer %p, id %d\n", req->rq_repbuf,
 258                conn.oc_id);
 259         req->rq_rep.ost->connid = conn.oc_id;
 260         EXIT;
 261         return 0;
 262 }
 263
 264 static int ost_disconnect(struct ost_obd *ost, struct ptlrpc_request *req)
 265 {
 266         struct obd_conn conn;
 267         int rc;
 268
 269         ENTRY;
 270
 271         conn.oc_dev = ost->ost_tgt;
 272         conn.oc_id = req->rq_req.ost->connid;
 273
 274         rc = ost_pack_rep(NULL, 0, NULL, 0, &req->rq_rephdr, &req->rq_rep.ost,
 275                           &req->rq_replen, &req->rq_repbuf);
 276         if (rc) {
 277                 CERROR("cannot pack reply\n");
 278                 return rc;
 279         }
 280
 281         req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_disconnect(&conn);
 282
 283         EXIT;
 284         return 0;
 285 }
 286
 287 static int ost_get_info(struct ost_obd *ost, struct ptlrpc_request *req)
 288 {
 289         struct obd_conn conn;
 290         int rc;
 291         int vallen;
 292         void *val;
 293         char *ptr;
 294
 295         ENTRY;
 296
 297         conn.oc_id = req->rq_req.ost->connid;
 298         conn.oc_dev = ost->ost_tgt;
 299
 300         ptr = ost_req_buf1(req->rq_req.ost);
 301         req->rq_rep.ost->result =ost->ost_tgt->obd_type->typ_ops->o_get_info
 302                 (&conn, req->rq_req.ost->buflen1, ptr, &vallen, &val);
 303
 304         rc = ost_pack_rep(val, vallen, NULL, 0, &req->rq_rephdr,
 305                           &req->rq_rep.ost, &req->rq_replen, &req->rq_repbuf);
 306         if (rc) {
 307                 CERROR("cannot pack reply\n");
 308                 return rc;
 309         }
 310
 311         EXIT;
 312         return 0;
 313 }
 314
 315 int ost_brw(struct ost_obd *obddev, struct ptlrpc_request *req)
 316 {
 317         struct obd_conn conn;
 318         int rc;
 319         int i, j;
 320         int objcount, niocount;
 321         char *tmp1, *tmp2, *end2;
 322         char *res;
 323         int cmd;
 324         struct niobuf *nb, *src, *dst;
 325         struct obd_ioobj *ioo;
 326         struct ost_req *r = req->rq_req.ost;
 327
 328         ENTRY;
 329
 330         tmp1 = ost_req_buf1(r);
 331         tmp2 = ost_req_buf2(r);
 332         end2 = tmp2 + req->rq_req.ost->buflen2;
 333         objcount = r->buflen1 / sizeof(*ioo);
 334         niocount = r->buflen2 / sizeof(*nb);
 335         cmd = r->cmd;
 336
 337         conn.oc_id = req->rq_req.ost->connid;
 338         conn.oc_dev = req->rq_ost->ost_tgt;
 339
 340         rc = ost_pack_rep(NULL, niocount, NULL, 0,
 341                           &req->rq_rephdr, &req->rq_rep.ost,
 342                           &req->rq_replen, &req->rq_repbuf);
 343         if (rc) {
 344                 CERROR("cannot pack reply\n");
 345                 return rc;
 346         }
 347         res = ost_rep_buf1(req->rq_rep.ost);
 348
 349         for (i=0; i < objcount; i++) {
 350                 ost_unpack_ioo((void *)&tmp1, &ioo);
 351                 if (tmp2 + ioo->ioo_bufcnt > end2) {
 352                         rc = -EFAULT;
 353                         break;
 354                 }
 355                 for (j = 0 ; j < ioo->ioo_bufcnt ; j++) {
 356                         ost_unpack_niobuf((void *)&tmp2, &nb);
 357                 }
 358         }
 359
 360         /* The unpackers move tmp1 and tmp2, so reset them before using */
 361         tmp1 = ost_req_buf1(r);
 362         tmp2 = ost_req_buf2(r);
 363         req->rq_rep.ost->result =
 364                 req->rq_ost->ost_tgt->obd_type->typ_ops->o_preprw
 365                 (cmd, &conn, objcount, (struct obd_ioobj *)tmp1,
 366                  niocount, (struct niobuf *)tmp2, (struct niobuf *)res);
 367
 368         if (cmd == OBD_BRW_WRITE) {
 369                 for (i=0; i<niocount; i++) {
 370                         src = &((struct niobuf *)tmp2)[i];
 371                         dst = &((struct niobuf *)res)[i];
 372                         memcpy((void *)(unsigned long)dst->addr,
 373                                (void *)(unsigned long)src->addr,
 374                                src->len);
 375                 }
 376         } else {
 377                 for (i=0; i<niocount; i++) {
 378                         dst = &((struct niobuf *)tmp2)[i];
 379                         src = &((struct niobuf *)res)[i];
 380                         memcpy((void *)(unsigned long)dst->addr,
 381                                (void *)(unsigned long)src->addr,
 382                                PAGE_SIZE);
 383                 }
 384         }
 385
 386         req->rq_rep.ost->result =
 387                 req->rq_ost->ost_tgt->obd_type->typ_ops->o_commitrw
 388                 (cmd, &conn, objcount, (struct obd_ioobj *)tmp1,
 389                  niocount, (struct niobuf *)res);
 390
 391         EXIT;
 392         return 0;
 393 }
 394
 395 int ost_handle(struct obd_device *obddev, struct ptlrpc_request *req)
 396 {
 397         int rc;
 398         struct ost_obd *ost = &obddev->u.ost;
 399         struct ptlreq_hdr *hdr;
 400
 401         ENTRY;
 402         CDEBUG(0, "req at %p\n", req);
 403
 404         hdr = (struct ptlreq_hdr *)req->rq_reqbuf;
 405         if (NTOH__u32(hdr->type) != OST_TYPE_REQ) {
 406                 CERROR("lustre_ost: wrong packet type sent %d\n",
 407                        NTOH__u32(hdr->type));
 408                 rc = -EINVAL;
 409                 goto out;
 410         }
 411
 412         rc = ost_unpack_req(req->rq_reqbuf, req->rq_reqlen,
 413                             &req->rq_reqhdr, &req->rq_req.ost);
 414         if (rc) {
 415                 CERROR("lustre_ost: Invalid request\n");
 416                 EXIT;
 417                 goto out;
 418         }
 419
 420         switch (req->rq_reqhdr->opc) {
 421
 422         case OST_CONNECT:
 423                 CDEBUG(D_INODE, "connect\n");
 424                 rc = ost_connect(ost, req);
 425                 break;
 426         case OST_DISCONNECT:
 427                 CDEBUG(D_INODE, "disconnect\n");
 428                 rc = ost_disconnect(ost, req);
 429                 break;
 430         case OST_GET_INFO:
 431                 CDEBUG(D_INODE, "get_info\n");
 432                 rc = ost_get_info(ost, req);
 433                 break;
 434         case OST_CREATE:
 435                 CDEBUG(D_INODE, "create\n");
 436                 rc = ost_create(ost, req);
 437                 break;
 438         case OST_DESTROY:
 439                 CDEBUG(D_INODE, "destroy\n");
 440                 rc = ost_destroy(ost, req);
 441                 break;
 442         case OST_GETATTR:
 443                 CDEBUG(D_INODE, "getattr\n");
 444                 rc = ost_getattr(ost, req);
 445                 break;
 446         case OST_SETATTR:
 447                 CDEBUG(D_INODE, "setattr\n");
 448                 rc = ost_setattr(ost, req);
 449                 break;
 450         case OST_BRW:
 451                 CDEBUG(D_INODE, "brw\n");
 452                 rc = ost_brw(ost, req);
 453                 break;
 454         default:
 455                 req->rq_status = -ENOTSUPP;
 456                 return ost_error(obddev, req);
 457         }
 458
 459 out:
 460         req->rq_status = rc;
 461         if (rc) {
 462                 CERROR("ost: processing error %d\n", rc);
 463                 ost_error(obddev, req);
 464         } else {
 465                 CDEBUG(D_INODE, "sending reply\n");
 466                 ost_reply(obddev, req);
 467         }
 468
 469         return 0;
 470 }
 471
 472 int ost_main(void *arg)
 473 {
 474         struct obd_device *obddev = (struct obd_device *) arg;
 475         struct ost_obd *ost = &obddev->u.ost;
 476         ENTRY;
 477
 478         lock_kernel();
 479         daemonize();
 480         spin_lock_irq(&current->sigmask_lock);
 481         sigfillset(&current->blocked);
 482         recalc_sigpending(current);
 483         spin_unlock_irq(&current->sigmask_lock);
 484
 485         sprintf(current->comm, "lustre_ost");
 486
 487         /* Record that the  thread is running */
 488         ost->ost_thread = current;
 489         wake_up(&ost->ost_done_waitq);
 490
 491         /* XXX maintain a list of all managed devices: insert here */
 492
 493         /* And now, wait forever for commit wakeup events. */
 494         while (1) {
 495                 int rc;
 496
 497                 if (ost->ost_flags & OST_EXIT)
 498                         break;
 499
 500                 wake_up(&ost->ost_done_waitq);
 501                 interruptible_sleep_on(&ost->ost_waitq);
 502                 barrier();
 503                 CDEBUG(D_INODE, "lustre_ost wakes\n");
 504                 CDEBUG(D_INODE, "pick up req here and continue\n");
 505
 506
 507                 if (ost->ost_service != NULL) {
 508                         ptl_event_t ev;
 509
 510                         CDEBUG(D_INODE, "\n");
 511                         while (1) {
 512                                 struct ptlrpc_request request;
 513                                 struct ptlrpc_service *service;
 514                                 CDEBUG(D_INODE, "\n");
 515                                 rc = PtlEQGet(ost->ost_service->srv_eq_h, &ev);
 516                                 if (rc != PTL_OK && rc != PTL_EQ_DROPPED)
 517                                         break;
 518                                 CDEBUG(D_INODE, "\n");
 519
 520                                 service = (struct ptlrpc_service *)ev.mem_desc.user_ptr;
 521
 522                                 /* FIXME: If we move to an event-driven model,
 523                                  * we should put the request on the stack of
 524                                  * mds_handle instead. */
 525                                 memset(&request, 0, sizeof(request));
 526                                 request.rq_reqbuf = ev.mem_desc.start +
 527                                         ev.offset;
 528                                 request.rq_reqlen = ev.mem_desc.length;
 529                                 request.rq_ost = ost;
 530                                 request.rq_xid = ev.match_bits;
 531
 532                                 request.rq_peer.peer_nid = ev.initiator.nid;
 533                                 /* FIXME: this NI should be the incoming NI.
 534                                  * We don't know how to find that from here. */
 535                                 request.rq_peer.peer_ni =
 536                                         ost->ost_service->srv_self.peer_ni;
 537                                 rc = ost_handle(obddev, &request);
 538
 539                                 /* Inform the rpc layer the event has been handled */
 540                                 ptl_received_rpc(service);
 541                         }
 542                 } else {
 543                         struct ptlrpc_request *request;
 544
 545                         if (list_empty(&ost->ost_reqs)) {
 546                                 CDEBUG(D_INODE, "woke because of timer\n");
 547                         } else {
 548                                 request = list_entry(ost->ost_reqs.next,
 549                                                      struct ptlrpc_request,
 550                                                      rq_list);
 551                                 list_del(&request->rq_list);
 552                                 rc = ost_handle(obddev, request);
 553                         }
 554                 }
 555         }
 556
 557         /* XXX maintain a list of all managed devices: cleanup here */
 558
 559         ost->ost_thread = NULL;
 560         wake_up(&ost->ost_done_waitq);
 561         CERROR("lustre_ost: exiting\n");
 562         return 0;
 563 }
 564
 565 static void ost_stop_srv_thread(struct ost_obd *ost)
 566 {
 567         ost->ost_flags |= OST_EXIT;
 568
 569         while (ost->ost_thread) {
 570                 wake_up(&ost->ost_waitq);
 571                 sleep_on(&ost->ost_done_waitq);
 572         }
 573 }
 574
 575 static void ost_start_srv_thread(struct obd_device *obd)
 576 {
 577         struct ost_obd *ost = &obd->u.ost;
 578         ENTRY;
 579
 580         init_waitqueue_head(&ost->ost_waitq);
 581         init_waitqueue_head(&ost->ost_done_waitq);
 582         kernel_thread(ost_main, (void *)obd,
 583                       CLONE_VM | CLONE_FS | CLONE_FILES);
 584         while (!ost->ost_thread)
 585                 sleep_on(&ost->ost_done_waitq);
 586         EXIT;
 587 }
 588
 589 /* mount the file system (secretly) */
 590 static int ost_setup(struct obd_device *obddev, obd_count len,
 591                         void *buf)
 592
 593 {
 594         struct obd_ioctl_data* data = buf;
 595         struct ost_obd *ost = &obddev->u.ost;
 596         struct obd_device *tgt;
 597         struct lustre_peer peer;
 598         int err;
 599         ENTRY;
 600
 601         if (data->ioc_dev  < 0 || data->ioc_dev > MAX_OBD_DEVICES) {
 602                 EXIT;
 603                 return -ENODEV;
 604         }
 605
 606         tgt = &obd_dev[data->ioc_dev];
 607         ost->ost_tgt = tgt;
 608         if ( ! (tgt->obd_flags & OBD_ATTACHED) ||
 609              ! (tgt->obd_flags & OBD_SET_UP) ){
 610                 CERROR("device not attached or not set up (%d)\n",
 611                        data->ioc_dev);
 612                 EXIT;
 613                 return -EINVAL;
 614         }
 615
 616         ost->ost_conn.oc_dev = tgt;
 617         err = tgt->obd_type->typ_ops->o_connect(&ost->ost_conn);
 618         if (err) {
 619                 CERROR("lustre ost: fail to connect to device %d\n",
 620                        data->ioc_dev);
 621                 return -EINVAL;
 622         }
 623
 624         INIT_LIST_HEAD(&ost->ost_reqs);
 625         ost->ost_thread = NULL;
 626         ost->ost_flags = 0;
 627
 628         spin_lock_init(&obddev->u.ost.ost_lock);
 629
 630         err = kportal_uuid_to_peer("self", &peer);
 631         if (err == 0) {
 632                 OBD_ALLOC(ost->ost_service, sizeof(*ost->ost_service));
 633                 if (ost->ost_service == NULL)
 634                         return -ENOMEM;
 635                 ost->ost_service->srv_buf_size = 64 * 1024;
 636                 ost->ost_service->srv_portal = OST_REQUEST_PORTAL;
 637                 memcpy(&ost->ost_service->srv_self, &peer, sizeof(peer));
 638                 ost->ost_service->srv_wait_queue = &ost->ost_waitq;
 639
 640                 rpc_register_service(ost->ost_service, "self");
 641         }
 642
 643         ost_start_srv_thread(obddev);
 644
 645         MOD_INC_USE_COUNT;
 646         EXIT;
 647         return 0;
 648 }
 649
 650 static int ost_cleanup(struct obd_device * obddev)
 651 {
 652         struct ost_obd *ost = &obddev->u.ost;
 653         struct obd_device *tgt;
 654         int err;
 655
 656         ENTRY;
 657
 658         if ( !(obddev->obd_flags & OBD_SET_UP) ) {
 659                 EXIT;
 660                 return 0;
 661         }
 662
 663         if ( !list_empty(&obddev->obd_gen_clients) ) {
 664                 CERROR("still has clients!\n");
 665                 EXIT;
 666                 return -EBUSY;
 667         }
 668
 669         ost_stop_srv_thread(ost);
 670         rpc_unregister_service(ost->ost_service);
 671         OBD_FREE(ost->ost_service, sizeof(*ost->ost_service));
 672
 673         if (!list_empty(&ost->ost_reqs)) {
 674                 // XXX reply with errors and clean up
 675                 CDEBUG(D_INODE, "Request list not empty!\n");
 676         }
 677
 678         tgt = ost->ost_tgt;
 679         err = tgt->obd_type->typ_ops->o_disconnect(&ost->ost_conn);
 680         if (err) {
 681                 CERROR("lustre ost: fail to disconnect device\n");
 682                 return -EINVAL;
 683         }
 684
 685
 686         MOD_DEC_USE_COUNT;
 687         EXIT;
 688         return 0;
 689 }
 690
 691 /* use obd ops to offer management infrastructure */
 692 static struct obd_ops ost_obd_ops = {
 693         o_setup:       ost_setup,
 694         o_cleanup:     ost_cleanup,
 695 };
 696
 697 static int __init ost_init(void)
 698 {
 699         obd_register_type(&ost_obd_ops, LUSTRE_OST_NAME);
 700         return 0;
 701 }
 702
 703 static void __exit ost_exit(void)
 704 {
 705         obd_unregister_type(LUSTRE_OST_NAME);
 706 }
 707
 708 MODULE_AUTHOR("Peter J. Braam <braam@clusterfs.com>");
 709 MODULE_DESCRIPTION("Lustre Object Storage Target (OST) v0.01");
 710 MODULE_LICENSE("GPL");
 711
 712 // for testing (maybe this stays)
 713 EXPORT_SYMBOL(ost_queue_req);
 714
 715 module_init(ost_init);
 716 module_exit(ost_exit);