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