lustre/mds/handler.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  lustre/mds/handler.c
   5  *  Lustre Metadata Server (mds) request handler
   6  *
   7  *  Copyright (c) 2001, 2002 Cluster File Systems, Inc.
   8  *   Author: Peter Braam <braam@clusterfs.com>
   9  *   Author: Andreas Dilger <adilger@clusterfs.com>
  10  *   Author: Phil Schwan <phil@clusterfs.com>
  11  *   Author: Mike Shaver <shaver@clusterfs.com>
  12  *
  13  *   This file is part of Lustre, http://www.lustre.org.
  14  *
  15  *   Lustre is free software; you can redistribute it and/or
  16  *   modify it under the terms of version 2 of the GNU General Public
  17  *   License as published by the Free Software Foundation.
  18  *
  19  *   Lustre is distributed in the hope that it will be useful,
  20  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  21  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  22  *   GNU General Public License for more details.
  23  *
  24  *   You should have received a copy of the GNU General Public License
  25  *   along with Lustre; if not, write to the Free Software
  26  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  27  */
  28
  29 #define EXPORT_SYMTAB
  30 #define DEBUG_SUBSYSTEM S_MDS
  31
  32 #include <linux/module.h>
  33 #include <linux/lustre_mds.h>
  34 #include <linux/lustre_dlm.h>
  35 #include <linux/init.h>
  36 #include <linux/obd_class.h>
  37 #include <linux/random.h>
  38 #include <linux/locks.h>
  39 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
  40 #include <linux/buffer_head.h>
  41 #endif
  42 #include <linux/obd_lov.h>
  43 #include <linux/lustre_mds.h>
  44 #include <linux/lustre_fsfilt.h>
  45 #include <linux/lprocfs_status.h>
  46
  47 static kmem_cache_t *mds_file_cache;
  48
  49 extern int mds_get_lovtgts(struct mds_obd *obd, int tgt_count,
  50                            obd_uuid_t *uuidarray);
  51 extern int mds_get_lovdesc(struct mds_obd  *obd, struct lov_desc *desc);
  52 extern void mds_start_transno(struct mds_obd *mds);
  53 extern int mds_finish_transno(struct mds_obd *mds, void *handle,
  54                               struct ptlrpc_request *req, int rc);
  55 static int mds_cleanup(struct obd_device * obddev);
  56
  57 extern struct lprocfs_vars status_var_nm_1[];
  58 extern struct lprocfs_vars status_class_var[];
  59
  60 inline struct mds_obd *mds_req2mds(struct ptlrpc_request *req)
  61 {
  62         return &req->rq_export->exp_obd->u.mds;
  63 }
  64
  65 static int mds_bulk_timeout(void *data)
  66 {
  67         struct ptlrpc_bulk_desc *desc = data;
  68
  69         ENTRY;
  70         recovd_conn_fail(desc->bd_connection);
  71         RETURN(1);
  72 }
  73
  74 /* Assumes caller has already pushed into the kernel filesystem context */
  75 static int mds_sendpage(struct ptlrpc_request *req, struct file *file,
  76                         __u64 offset)
  77 {
  78         struct ptlrpc_bulk_desc *desc;
  79         struct ptlrpc_bulk_page *bulk;
  80         struct l_wait_info lwi;
  81         char *buf;
  82         int rc = 0;
  83         ENTRY;
  84
  85         desc = ptlrpc_prep_bulk(req->rq_connection);
  86         if (desc == NULL)
  87                 GOTO(out, rc = -ENOMEM);
  88
  89         bulk = ptlrpc_prep_bulk_page(desc);
  90         if (bulk == NULL)
  91                 GOTO(cleanup_bulk, rc = -ENOMEM);
  92
  93         OBD_ALLOC(buf, PAGE_SIZE);
  94         if (buf == NULL)
  95                 GOTO(cleanup_bulk, rc = -ENOMEM);
  96
  97         rc = fsfilt_readpage(req->rq_export->exp_obd, file, buf, PAGE_SIZE,
  98                              (loff_t *)&offset);
  99
 100         if (rc != PAGE_SIZE)
 101                 GOTO(cleanup_buf, rc = -EIO);
 102
 103         bulk->bp_xid = req->rq_xid;
 104         bulk->bp_buf = buf;
 105         bulk->bp_buflen = PAGE_SIZE;
 106         desc->bd_ptl_ev_hdlr = NULL;
 107         desc->bd_portal = MDS_BULK_PORTAL;
 108
 109         rc = ptlrpc_send_bulk(desc);
 110         if (rc)
 111                 GOTO(cleanup_buf, rc);
 112
 113         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE)) {
 114                 CERROR("obd_fail_loc=%x, fail operation rc=%d\n",
 115                        OBD_FAIL_MDS_SENDPAGE, rc);
 116                 ptlrpc_abort_bulk(desc);
 117                 GOTO(cleanup_buf, rc);
 118         }
 119
 120         lwi = LWI_TIMEOUT(obd_timeout * HZ, mds_bulk_timeout, desc);
 121         rc = l_wait_event(desc->bd_waitq, desc->bd_flags & PTL_BULK_FL_SENT,
 122                           &lwi);
 123         if (rc) {
 124                 if (rc != -ETIMEDOUT)
 125                         LBUG();
 126                 GOTO(cleanup_buf, rc);
 127         }
 128
 129         EXIT;
 130  cleanup_buf:
 131         OBD_FREE(buf, PAGE_SIZE);
 132  cleanup_bulk:
 133         ptlrpc_bulk_decref(desc);
 134  out:
 135         return rc;
 136 }
 137
 138 /*
 139  * Look up a named entry in a directory, and get an LDLM lock on it.
 140  * 'dir' is a inode for which an LDLM lock has already been taken.
 141  *
 142  * If we do not need an exclusive or write lock on this entry (e.g.
 143  * a read lock for attribute lookup only) then we do not hold the
 144  * directory semaphore on return.  It is up to the caller to know what
 145  * type of lock it is getting, and clean up appropriately.
 146  */
 147 struct dentry *mds_name2locked_dentry(struct obd_device *obd,
 148                                       struct dentry *dir, struct vfsmount **mnt,
 149                                       char *name, int namelen, int lock_mode,
 150                                       struct lustre_handle *lockh,
 151                                       int dir_lock_mode)
 152 {
 153         struct dentry *dchild;
 154         int flags = 0, rc;
 155         __u64 res_id[3] = {0};
 156         ENTRY;
 157
 158         down(&dir->d_inode->i_sem);
 159         dchild = lookup_one_len(name, dir, namelen);
 160         if (IS_ERR(dchild)) {
 161                 CERROR("child lookup error %ld\n", PTR_ERR(dchild));
 162                 up(&dir->d_inode->i_sem);
 163                 LBUG();
 164                 RETURN(dchild);
 165         }
 166         if (dir_lock_mode != LCK_EX && dir_lock_mode != LCK_PW) {
 167                 up(&dir->d_inode->i_sem);
 168                 ldlm_lock_decref(lockh, dir_lock_mode);
 169         }
 170
 171         if (lock_mode == 0 || !dchild->d_inode)
 172                 RETURN(dchild);
 173
 174         res_id[0] = dchild->d_inode->i_ino;
 175         res_id[1] = dchild->d_inode->i_generation;
 176         rc = ldlm_match_or_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 177                                    res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 178                                    &flags, ldlm_completion_ast,
 179                                    mds_blocking_ast, NULL, 0, lockh);
 180         if (rc != ELDLM_OK) {
 181                 l_dput(dchild);
 182                 up(&dir->d_inode->i_sem);
 183                 RETURN(ERR_PTR(-ENOLCK)); /* XXX translate ldlm code */
 184         }
 185
 186         RETURN(dchild);
 187 }
 188
 189 struct dentry *mds_fid2locked_dentry(struct obd_device *obd, struct ll_fid *fid,
 190                                      struct vfsmount **mnt, int lock_mode,
 191                                      struct lustre_handle *lockh)
 192 {
 193         struct mds_obd *mds = &obd->u.mds;
 194         struct dentry *de = mds_fid2dentry(mds, fid, mnt), *retval = de;
 195         int flags = 0, rc;
 196         __u64 res_id[3] = {0};
 197         ENTRY;
 198
 199         if (IS_ERR(de))
 200                 RETURN(de);
 201
 202         res_id[0] = de->d_inode->i_ino;
 203         res_id[1] = de->d_inode->i_generation;
 204         rc = ldlm_match_or_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 205                                    res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 206                                    &flags, ldlm_completion_ast,
 207                                    mds_blocking_ast, NULL, 0, lockh);
 208         if (rc != ELDLM_OK) {
 209                 l_dput(de);
 210                 retval = ERR_PTR(-ENOLCK); /* XXX translate ldlm code */
 211         }
 212
 213         RETURN(retval);
 214 }
 215
 216 #ifndef DCACHE_DISCONNECTED
 217 #define DCACHE_DISCONNECTED DCACHE_NFSD_DISCONNECTED
 218 #endif
 219
 220 /* Look up an entry by inode number. */
 221 struct dentry *mds_fid2dentry(struct mds_obd *mds, struct ll_fid *fid,
 222                               struct vfsmount **mnt)
 223 {
 224         /* stolen from NFS */
 225         struct super_block *sb = mds->mds_sb;
 226         unsigned long ino = fid->id;
 227         __u32 generation = fid->generation;
 228         struct inode *inode;
 229         struct list_head *lp;
 230         struct dentry *result;
 231
 232         if (ino == 0)
 233                 RETURN(ERR_PTR(-ESTALE));
 234
 235         inode = iget(sb, ino);
 236         if (inode == NULL)
 237                 RETURN(ERR_PTR(-ENOMEM));
 238
 239         CDEBUG(D_DENTRY, "--> mds_fid2dentry: sb %p\n", inode->i_sb);
 240
 241         if (is_bad_inode(inode) ||
 242             (generation && inode->i_generation != generation)) {
 243                 /* we didn't find the right inode.. */
 244                 CERROR("bad inode %lu, link: %d ct: %d or version  %u/%u\n",
 245                        inode->i_ino, inode->i_nlink,
 246                        atomic_read(&inode->i_count), inode->i_generation,
 247                        generation);
 248                 iput(inode);
 249                 RETURN(ERR_PTR(-ENOENT));
 250         }
 251
 252         /* now to find a dentry. If possible, get a well-connected one */
 253         if (mnt)
 254                 *mnt = mds->mds_vfsmnt;
 255         spin_lock(&dcache_lock);
 256         list_for_each(lp, &inode->i_dentry) {
 257                 result = list_entry(lp, struct dentry, d_alias);
 258                 if (!(result->d_flags & DCACHE_DISCONNECTED)) {
 259                         dget_locked(result);
 260                         result->d_vfs_flags |= DCACHE_REFERENCED;
 261                         spin_unlock(&dcache_lock);
 262                         iput(inode);
 263                         if (mnt)
 264                                 mntget(*mnt);
 265                         return result;
 266                 }
 267         }
 268         spin_unlock(&dcache_lock);
 269         result = d_alloc_root(inode);
 270         if (result == NULL) {
 271                 iput(inode);
 272                 return ERR_PTR(-ENOMEM);
 273         }
 274         if (mnt)
 275                 mntget(*mnt);
 276         result->d_flags |= DCACHE_DISCONNECTED;
 277         return result;
 278 }
 279
 280 /* Establish a connection to the MDS.
 281  *
 282  * This will set up an export structure for the client to hold state data
 283  * about that client, like open files, the last operation number it did
 284  * on the server, etc.
 285  */
 286 static int mds_connect(struct lustre_handle *conn, struct obd_device *obd,
 287                        obd_uuid_t cluuid, struct recovd_obd *recovd,
 288                        ptlrpc_recovery_cb_t recover)
 289 {
 290         struct obd_export *exp;
 291         struct mds_export_data *med;
 292         struct mds_client_data *mcd;
 293         struct list_head *p;
 294         int rc;
 295         ENTRY;
 296
 297         if (!conn || !obd || !cluuid)
 298                 RETURN(-EINVAL);
 299
 300         MOD_INC_USE_COUNT;
 301
 302         spin_lock(&obd->obd_dev_lock);
 303         list_for_each(p, &obd->obd_exports) {
 304                 exp = list_entry(p, struct obd_export, exp_obd_chain);
 305                 mcd = exp->exp_mds_data.med_mcd;
 306                 if (!mcd) {
 307                         CERROR("FYI: NULL mcd - simultaneous connects\n");
 308                         continue;
 309                 }
 310                 if (!memcmp(cluuid, mcd->mcd_uuid, sizeof mcd->mcd_uuid)) {
 311                         /* XXX make handle-found-export a subroutine */
 312                         LASSERT(exp->exp_obd == obd);
 313
 314                         spin_unlock(&obd->obd_dev_lock);
 315                         if (exp->exp_connection) {
 316                                 struct lustre_handle *hdl;
 317                                 hdl = &exp->exp_ldlm_data.led_import.imp_handle;
 318                                 /* Might be a re-connect after a partition. */
 319                                 if (!memcmp(conn, hdl, sizeof *conn)) {
 320                                         CERROR("%s reconnecting\n", cluuid);
 321                                         conn->addr = (__u64) (unsigned long)exp;
 322                                         conn->cookie = exp->exp_cookie;
 323                                         rc = EALREADY;
 324                                 } else {
 325                                         CERROR("%s reconnecting from %s, "
 326                                                "handle mismatch (ours %Lx/%Lx, "
 327                                                "theirs %Lx/%Lx)\n", cluuid,
 328                                                exp->exp_connection->
 329                                                c_remote_uuid, hdl->addr,
 330                                                hdl->cookie, conn->addr,
 331                                                conn->cookie);
 332                                         /* XXX disconnect them here? */
 333                                         memset(conn, 0, sizeof *conn);
 334                                         rc = -EALREADY;
 335                                 }
 336                                 MOD_DEC_USE_COUNT;
 337                                 RETURN(rc);
 338                         }
 339                         conn->addr = (__u64) (unsigned long)exp;
 340                         conn->cookie = exp->exp_cookie;
 341                         CDEBUG(D_INFO, "existing export for UUID '%s' at %p\n",
 342                                cluuid, exp);
 343                         CDEBUG(D_IOCTL,"connect: addr %Lx cookie %Lx\n",
 344                                (long long)conn->addr, (long long)conn->cookie);
 345                         RETURN(0);
 346                 }
 347         }
 348         spin_unlock(&obd->obd_dev_lock);
 349
 350         if (obd->u.mds.mds_recoverable_clients != 0) {
 351                 CERROR("denying connection for new client %s: in recovery\n",
 352                        cluuid);
 353                 MOD_DEC_USE_COUNT;
 354                 RETURN(-EBUSY);
 355         }
 356
 357         /* XXX There is a small race between checking the list and adding a
 358          * new connection for the same UUID, but the real threat (list
 359          * corruption when multiple different clients connect) is solved.
 360          *
 361          * There is a second race between adding the export to the list,
 362          * and filling in the client data below.  Hence skipping the case
 363          * of NULL mcd above.  We should already be controlling multiple
 364          * connects at the client, and we can't hold the spinlock over
 365          * memory allocations without risk of deadlocking.
 366          */
 367         rc = class_connect(conn, obd, cluuid);
 368         if (rc)
 369                 GOTO(out_dec, rc);
 370         exp = class_conn2export(conn);
 371         LASSERT(exp);
 372         med = &exp->exp_mds_data;
 373
 374         OBD_ALLOC(mcd, sizeof(*mcd));
 375         if (!mcd) {
 376                 CERROR("mds: out of memory for client data\n");
 377                 GOTO(out_export, rc = -ENOMEM);
 378         }
 379
 380         memcpy(mcd->mcd_uuid, cluuid, sizeof(mcd->mcd_uuid));
 381         med->med_mcd = mcd;
 382
 383         INIT_LIST_HEAD(&med->med_open_head);
 384         spin_lock_init(&med->med_open_lock);
 385
 386         rc = mds_client_add(&obd->u.mds, med, -1);
 387         if (rc)
 388                 GOTO(out_mcd, rc);
 389
 390         RETURN(0);
 391
 392 out_mcd:
 393         OBD_FREE(mcd, sizeof(*mcd));
 394 out_export:
 395         class_disconnect(conn);
 396 out_dec:
 397         MOD_DEC_USE_COUNT;
 398
 399         return rc;
 400 }
 401
 402 /* Call with med->med_open_lock held, please. */
 403 inline int mds_close_mfd(struct mds_file_data *mfd, struct mds_export_data *med)
 404 {
 405         struct file *file = mfd->mfd_file;
 406         LASSERT(file->private_data == mfd);
 407
 408         list_del(&mfd->mfd_list);
 409         mfd->mfd_servercookie = DEAD_HANDLE_MAGIC;
 410         kmem_cache_free(mds_file_cache, mfd);
 411
 412         return filp_close(file, 0);
 413 }
 414
 415 static int mds_disconnect(struct lustre_handle *conn)
 416 {
 417         struct obd_export *export = class_conn2export(conn);
 418         struct list_head *tmp, *n;
 419         struct mds_export_data *med = &export->exp_mds_data;
 420         int rc;
 421         ENTRY;
 422
 423         /*
 424          * Close any open files.
 425          */
 426         spin_lock(&med->med_open_lock);
 427         list_for_each_safe(tmp, n, &med->med_open_head) {
 428                 struct mds_file_data *mfd =
 429                         list_entry(tmp, struct mds_file_data, mfd_list);
 430                 rc = mds_close_mfd(mfd, med);
 431                 if (rc) {
 432                         /* XXX better diagnostics, with file path and stuff */
 433                         CDEBUG(D_INODE, "Error %d closing mfd %p\n", rc, mfd);
 434                 }
 435         }
 436         spin_unlock(&med->med_open_lock);
 437
 438         ldlm_cancel_locks_for_export(export);
 439         mds_client_free(export);
 440
 441         rc = class_disconnect(conn);
 442         if (!rc)
 443                 MOD_DEC_USE_COUNT;
 444
 445         RETURN(rc);
 446 }
 447
 448 /*
 449  * XXX This is NOT guaranteed to flush all transactions to disk (even though
 450  *     it is equivalent to calling sync()) because it only _starts_ the flush
 451  *     and does not wait for completion.  It's better than nothing though.
 452  *     What we really want is a mild form of fsync_dev_lockfs(), but it is
 453  *     non-standard, or enabling do_sync_supers in ext3, just for this call.
 454  */
 455 static void mds_fsync_super(struct super_block *sb)
 456 {
 457         lock_kernel();
 458         lock_super(sb);
 459         if (sb->s_dirt && sb->s_op && sb->s_op->write_super)
 460                 sb->s_op->write_super(sb);
 461         unlock_super(sb);
 462         unlock_kernel();
 463 }
 464
 465 static int mds_getstatus(struct ptlrpc_request *req)
 466 {
 467         struct mds_obd *mds = mds_req2mds(req);
 468         struct mds_body *body;
 469         int rc, size = sizeof(*body);
 470         ENTRY;
 471
 472         rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen, &req->rq_repmsg);
 473         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_GETSTATUS_PACK)) {
 474                 CERROR("mds: out of memory for message: size=%d\n", size);
 475                 req->rq_status = -ENOMEM;
 476                 RETURN(0);
 477         }
 478
 479         /* Flush any outstanding transactions to disk so the client will
 480          * get the latest last_committed value and can drop their local
 481          * requests if they have any.  This would be fsync_super() if it
 482          * was exported.
 483          */
 484         mds_fsync_super(mds->mds_sb);
 485
 486         body = lustre_msg_buf(req->rq_repmsg, 0);
 487         memcpy(&body->fid1, &mds->mds_rootfid, sizeof(body->fid1));
 488
 489         /* the last_committed and last_xid fields are filled in for all
 490          * replies already - no need to do so here also.
 491          */
 492         RETURN(0);
 493 }
 494
 495 static int mds_getlovinfo(struct ptlrpc_request *req)
 496 {
 497         struct mds_obd *mds = mds_req2mds(req);
 498         struct mds_status_req *streq;
 499         struct lov_desc *desc;
 500         int tgt_count;
 501         int rc, size[2] = {sizeof(*desc)};
 502         ENTRY;
 503
 504         streq = lustre_msg_buf(req->rq_reqmsg, 0);
 505         streq->flags = NTOH__u32(streq->flags);
 506         streq->repbuf = NTOH__u32(streq->repbuf);
 507         size[1] = streq->repbuf;
 508
 509         rc = lustre_pack_msg(2, size, NULL, &req->rq_replen, &req->rq_repmsg);
 510         if (rc) {
 511                 CERROR("mds: out of memory for message: size=%d\n", size[1]);
 512                 req->rq_status = -ENOMEM;
 513                 RETURN(0);
 514         }
 515
 516         desc = lustre_msg_buf(req->rq_repmsg, 0);
 517         rc = mds_get_lovdesc(mds, desc);
 518         if (rc) {
 519                 req->rq_status = rc;
 520                 RETURN(0);
 521         }
 522
 523         tgt_count = le32_to_cpu(desc->ld_tgt_count);
 524         if (tgt_count * sizeof(obd_uuid_t) > streq->repbuf) {
 525                 CERROR("too many targets, enlarge client buffers\n");
 526                 req->rq_status = -ENOSPC;
 527                 RETURN(0);
 528         }
 529
 530         /* XXX the MDS should not really know about this */
 531         mds->mds_max_mdsize = lov_mds_md_size(tgt_count);
 532         rc = mds_get_lovtgts(mds, tgt_count,
 533                              lustre_msg_buf(req->rq_repmsg, 1));
 534         if (rc) {
 535                 CERROR("get_lovtgts error %d\n", rc);
 536                 req->rq_status = rc;
 537                 RETURN(0);
 538         }
 539         RETURN(0);
 540 }
 541
 542 int mds_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
 543                      void *data, __u32 data_len, int flag)
 544 {
 545         int do_ast;
 546         ENTRY;
 547
 548         if (flag == LDLM_CB_CANCELING) {
 549                 /* Don't need to do anything here. */
 550                 RETURN(0);
 551         }
 552
 553         /* XXX layering violation!  -phil */
 554         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 555         lock->l_flags |= LDLM_FL_CBPENDING;
 556         do_ast = (!lock->l_readers && !lock->l_writers);
 557         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 558
 559         if (do_ast) {
 560                 struct lustre_handle lockh;
 561                 int rc;
 562
 563                 LDLM_DEBUG(lock, "already unused, calling ldlm_cli_cancel");
 564                 ldlm_lock2handle(lock, &lockh);
 565                 rc = ldlm_cli_cancel(&lockh);
 566                 if (rc < 0)
 567                         CERROR("ldlm_cli_cancel: %d\n", rc);
 568         } else
 569                 LDLM_DEBUG(lock, "Lock still has references, will be"
 570                            "cancelled later");
 571         RETURN(0);
 572 }
 573
 574 int mds_pack_md(struct mds_obd *mds, struct ptlrpc_request *req,
 575                 int offset, struct mds_body *body, struct inode *inode)
 576 {
 577         struct lov_mds_md *lmm;
 578         int lmm_size = req->rq_repmsg->buflens[offset];
 579         int rc;
 580
 581         if (lmm_size == 0) {
 582                 CDEBUG(D_INFO, "no space reserved for inode %lu MD\n", inode->i_ino);
 583                 RETURN(0);
 584         }
 585
 586         lmm = lustre_msg_buf(req->rq_repmsg, offset);
 587
 588         /* I don't really like this, but it is a sanity check on the client
 589          * MD request.  However, if the client doesn't know how much space
 590          * to reserve for the MD, this shouldn't be fatal either...
 591          */
 592         if (lmm_size > mds->mds_max_mdsize) {
 593                 CERROR("Reading MD for inode %lu of %d bytes > max %d\n",
 594                        inode->i_ino, lmm_size, mds->mds_max_mdsize);
 595                 // RETURN(-EINVAL);
 596         }
 597
 598         /* We don't need to store the reply size, because this buffer is
 599          * discarded right after unpacking, and the LOV can figure out the
 600          * size itself from the ost count.
 601          */
 602         if ((rc = fsfilt_get_md(req->rq_export->exp_obd, inode,
 603                                 lmm, lmm_size)) < 0) {
 604                 CDEBUG(D_INFO, "No md for ino %lu: rc = %d\n", inode->i_ino,rc);
 605         } else if (rc > 0) {
 606                 body->valid |= OBD_MD_FLEASIZE;
 607                 rc = 0;
 608         }
 609
 610         return rc;
 611 }
 612
 613 static int mds_getattr_internal(struct mds_obd *mds, struct dentry *dentry,
 614                                 struct ptlrpc_request *req,
 615                                 struct mds_body *reqbody, int reply_off)
 616 {
 617         struct mds_body *body;
 618         struct inode *inode = dentry->d_inode;
 619         int rc = 0;
 620         ENTRY;
 621
 622         if (inode == NULL)
 623                 RETURN(-ENOENT);
 624
 625         body = lustre_msg_buf(req->rq_repmsg, reply_off);
 626
 627         mds_pack_inode2fid(&body->fid1, inode);
 628         mds_pack_inode2body(body, inode);
 629
 630         if (S_ISREG(inode->i_mode) /* && reqbody->valid & OBD_MD_FLEASIZE */) {
 631                 rc = mds_pack_md(mds, req, reply_off + 1, body, inode);
 632         } else if (S_ISLNK(inode->i_mode) && reqbody->valid & OBD_MD_LINKNAME) {
 633                 char *symname = lustre_msg_buf(req->rq_repmsg, reply_off + 1);
 634                 int len = req->rq_repmsg->buflens[reply_off + 1];
 635
 636                 rc = inode->i_op->readlink(dentry, symname, len);
 637                 if (rc < 0) {
 638                         CERROR("readlink failed: %d\n", rc);
 639                 } else {
 640                         CDEBUG(D_INODE, "read symlink dest %s\n", symname);
 641                         body->valid |= OBD_MD_LINKNAME;
 642                         rc = 0;
 643                 }
 644         }
 645         RETURN(rc);
 646 }
 647
 648 static int mds_getattr_name(int offset, struct ptlrpc_request *req)
 649 {
 650         struct mds_obd *mds = mds_req2mds(req);
 651         struct obd_device *obd = req->rq_export->exp_obd;
 652         struct obd_run_ctxt saved;
 653         struct mds_body *body;
 654         struct dentry *de = NULL, *dchild = NULL;
 655         struct inode *dir;
 656         struct lustre_handle lockh;
 657         char *name;
 658         int namelen, flags = 0, lock_mode, rc = 0;
 659         struct obd_ucred uc;
 660         __u64 res_id[3] = {0, 0, 0};
 661         ENTRY;
 662
 663         LASSERT(!strcmp(obd->obd_type->typ_name, "mds"));
 664
 665         if (req->rq_reqmsg->bufcount <= offset + 1) {
 666                 LBUG();
 667                 GOTO(out_pre_de, rc = -EINVAL);
 668         }
 669
 670         body = lustre_msg_buf(req->rq_reqmsg, offset);
 671         name = lustre_msg_buf(req->rq_reqmsg, offset + 1);
 672         namelen = req->rq_reqmsg->buflens[offset + 1];
 673         /* requests were at offset 2, replies go back at 1 */
 674         if (offset)
 675                 offset = 1;
 676
 677         uc.ouc_fsuid = body->fsuid;
 678         uc.ouc_fsgid = body->fsgid;
 679         uc.ouc_cap = body->capability;
 680         push_ctxt(&saved, &mds->mds_ctxt, &uc);
 681         de = mds_fid2dentry(mds, &body->fid1, NULL);
 682         if (IS_ERR(de)) {
 683                 GOTO(out_pre_de, rc = -ENOENT);
 684         }
 685
 686         dir = de->d_inode;
 687         CDEBUG(D_INODE, "parent ino %lu, name %*s\n", dir->i_ino,namelen,name);
 688
 689         lock_mode = LCK_PR;
 690         res_id[0] = dir->i_ino;
 691         res_id[1] = dir->i_generation;
 692
 693         rc = ldlm_lock_match(obd->obd_namespace, res_id, LDLM_PLAIN,
 694                              NULL, 0, lock_mode, &lockh);
 695         if (rc == 0) {
 696                 LDLM_DEBUG_NOLOCK("enqueue res "LPU64, res_id[0]);
 697                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 698                                       res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 699                                       &flags, ldlm_completion_ast,
 700                                       mds_blocking_ast, NULL, 0, &lockh);
 701                 if (rc != ELDLM_OK) {
 702                         CERROR("lock enqueue: err: %d\n", rc);
 703                         GOTO(out_create_de, rc = -EIO);
 704                 }
 705         }
 706         ldlm_lock_dump((void *)(unsigned long)lockh.addr);
 707
 708         down(&dir->i_sem);
 709         dchild = lookup_one_len(name, de, namelen - 1);
 710         if (IS_ERR(dchild)) {
 711                 CDEBUG(D_INODE, "child lookup error %ld\n", PTR_ERR(dchild));
 712                 up(&dir->i_sem);
 713                 GOTO(out_create_dchild, rc = PTR_ERR(dchild));
 714         }
 715
 716         rc = mds_getattr_internal(mds, dchild, req, body, offset);
 717
 718         EXIT;
 719 out_create_dchild:
 720         l_dput(dchild);
 721         up(&dir->i_sem);
 722         ldlm_lock_decref(&lockh, lock_mode);
 723 out_create_de:
 724         l_dput(de);
 725 out_pre_de:
 726         req->rq_status = rc;
 727         pop_ctxt(&saved, &mds->mds_ctxt, &uc);
 728         return 0;
 729 }
 730
 731 static int mds_getattr(int offset, struct ptlrpc_request *req)
 732 {
 733         struct mds_obd *mds = mds_req2mds(req);
 734         struct obd_run_ctxt saved;
 735         struct dentry *de;
 736         struct inode *inode;
 737         struct mds_body *body;
 738         struct obd_ucred uc;
 739         int rc = 0, size[2] = {sizeof(*body)}, bufcount = 1;
 740         ENTRY;
 741
 742         body = lustre_msg_buf(req->rq_reqmsg, offset);
 743         uc.ouc_fsuid = body->fsuid;
 744         uc.ouc_fsgid = body->fsgid;
 745         uc.ouc_cap = body->capability;
 746         push_ctxt(&saved, &mds->mds_ctxt, &uc);
 747         de = mds_fid2dentry(mds, &body->fid1, NULL);
 748         if (IS_ERR(de)) {
 749                 rc = req->rq_status = -ENOENT;
 750                 GOTO(out_pop, PTR_ERR(de));
 751         }
 752
 753         inode = de->d_inode;
 754         if (S_ISREG(body->fid1.f_type)) {
 755                 int rc = fsfilt_get_md(req->rq_export->exp_obd, inode, NULL, 0);
 756                 CDEBUG(D_INODE, "got %d bytes MD data for inode %lu\n",
 757                        rc, inode->i_ino);
 758                 if (rc < 0) {
 759                         if (rc != -ENODATA)
 760                                 CERROR("error getting inode %lu MD: rc = %d\n",
 761                                        inode->i_ino, rc);
 762                         size[bufcount] = 0;
 763                 } else if (rc > mds->mds_max_mdsize) {
 764                         size[bufcount] = 0;
 765                         CERROR("MD size %d larger than maximum possible %u\n",
 766                                rc, mds->mds_max_mdsize);
 767                 } else
 768                         size[bufcount] = rc;
 769                 bufcount++;
 770         } else if (body->valid & OBD_MD_LINKNAME) {
 771                 size[bufcount] = MIN(inode->i_size + 1, body->size);
 772                 bufcount++;
 773                 CDEBUG(D_INODE, "symlink size: %d, reply space: %d\n",
 774                        inode->i_size + 1, body->size);
 775         }
 776
 777         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) {
 778                 CERROR("failed MDS_GETATTR_PACK test\n");
 779                 req->rq_status = -ENOMEM;
 780                 GOTO(out, rc = -ENOMEM);
 781         }
 782
 783         rc = lustre_pack_msg(bufcount, size, NULL, &req->rq_replen,
 784                              &req->rq_repmsg);
 785         if (rc) {
 786                 CERROR("out of memoryK\n");
 787                 req->rq_status = rc;
 788                 GOTO(out, rc);
 789         }
 790
 791         req->rq_status = mds_getattr_internal(mds, de, req, body, 0);
 792
 793         EXIT;
 794 out:
 795         l_dput(de);
 796 out_pop:
 797         pop_ctxt(&saved, &mds->mds_ctxt, &uc);
 798         return rc;
 799 }
 800
 801 static int mds_statfs(struct ptlrpc_request *req)
 802 {
 803         struct obd_device *obd = req->rq_export->exp_obd;
 804         struct obd_statfs *osfs;
 805         int rc, size = sizeof(*osfs);
 806         ENTRY;
 807
 808         rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen, &req->rq_repmsg);
 809         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_STATFS_PACK)) {
 810                 CERROR("mds: statfs lustre_pack_msg failed: rc = %d\n", rc);
 811                 GOTO(out, rc);
 812         }
 813
 814         osfs = lustre_msg_buf(req->rq_repmsg, 0);
 815         rc = fsfilt_statfs(obd, obd->u.mds.mds_sb, osfs);
 816         if (rc) {
 817                 CERROR("mds: statfs failed: rc %d\n", rc);
 818                 GOTO(out, rc);
 819         }
 820         obd_statfs_pack(osfs, osfs);
 821
 822         EXIT;
 823 out:
 824         req->rq_status = rc;
 825         return 0;
 826 }
 827
 828 static struct mds_file_data *mds_handle2mfd(struct lustre_handle *handle)
 829 {
 830         struct mds_file_data *mfd = NULL;
 831         ENTRY;
 832
 833         if (!handle || !handle->addr)
 834                 RETURN(NULL);
 835
 836         mfd = (struct mds_file_data *)(unsigned long)(handle->addr);
 837         if (!kmem_cache_validate(mds_file_cache, mfd))
 838                 RETURN(NULL);
 839
 840         if (mfd->mfd_servercookie != handle->cookie)
 841                 RETURN(NULL);
 842
 843         RETURN(mfd);
 844 }
 845
 846 static int mds_store_md(struct mds_obd *mds, struct ptlrpc_request *req,
 847                         int offset, struct mds_body *body, struct inode *inode)
 848 {
 849         struct obd_device *obd = req->rq_export->exp_obd;
 850         struct lov_mds_md *lmm = lustre_msg_buf(req->rq_reqmsg, offset);
 851         int lmm_size = req->rq_reqmsg->buflens[offset];
 852         struct obd_run_ctxt saved;
 853         struct obd_ucred uc;
 854         void *handle;
 855         int rc, rc2;
 856         ENTRY;
 857
 858         /* I don't really like this, but it is a sanity check on the client
 859          * MD request.
 860          */
 861         if (lmm_size > mds->mds_max_mdsize) {
 862                 CERROR("Saving MD for inode %lu of %d bytes > max %d\n",
 863                        inode->i_ino, lmm_size, mds->mds_max_mdsize);
 864                 //RETURN(-EINVAL);
 865         }
 866
 867         CDEBUG(D_INODE, "storing %d bytes MD for inode %lu\n",
 868                lmm_size, inode->i_ino);
 869         uc.ouc_fsuid = body->fsuid;
 870         uc.ouc_fsgid = body->fsgid;
 871         uc.ouc_cap = body->capability;
 872         push_ctxt(&saved, &mds->mds_ctxt, &uc);
 873         mds_start_transno(mds);
 874         handle = fsfilt_start(obd, inode,FSFILT_OP_SETATTR);
 875         if (IS_ERR(handle)) {
 876                 rc = PTR_ERR(handle);
 877                 mds_finish_transno(mds, handle, req, rc);
 878                 GOTO(out_ea, rc);
 879         }
 880
 881         rc = fsfilt_set_md(obd, inode,handle,lmm,lmm_size);
 882         rc = mds_finish_transno(mds, handle, req, rc);
 883
 884         rc2 = fsfilt_commit(obd, inode, handle);
 885         if (rc2 && !rc)
 886                 rc = rc2;
 887 out_ea:
 888         pop_ctxt(&saved, &mds->mds_ctxt, &uc);
 889
 890         RETURN(rc);
 891 }
 892
 893 static int mds_open(struct ptlrpc_request *req)
 894 {
 895         struct mds_obd *mds = mds_req2mds(req);
 896         struct mds_body *body;
 897         struct mds_export_data *med;
 898         struct mds_file_data *mfd;
 899         struct dentry *de;
 900         struct file *file;
 901         struct vfsmount *mnt;
 902         __u32 flags;
 903         struct list_head *tmp;
 904         int rc, size = sizeof(*body);
 905         ENTRY;
 906
 907         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OPEN_PACK)) {
 908                 CERROR("test case OBD_FAIL_MDS_OPEN_PACK\n");
 909                 req->rq_status = -ENOMEM;
 910                 RETURN(-ENOMEM);
 911         }
 912
 913         rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen, &req->rq_repmsg);
 914         if (rc) {
 915                 CERROR("mds: pack error: rc = %d\n", rc);
 916                 req->rq_status = rc;
 917                 RETURN(rc);
 918         }
 919
 920         body = lustre_msg_buf(req->rq_reqmsg, 0);
 921
 922         /* was this animal open already and the client lost the reply? */
 923         /* XXX need some way to detect a reopen, to avoid locked list walks */
 924         med = &req->rq_export->exp_mds_data;
 925         spin_lock(&med->med_open_lock);
 926         list_for_each(tmp, &med->med_open_head) {
 927                 mfd = list_entry(tmp, typeof(*mfd), mfd_list);
 928                 if (!memcmp(&mfd->mfd_clienthandle, &body->handle,
 929                             sizeof(mfd->mfd_clienthandle)) &&
 930                     body->fid1.id == mfd->mfd_file->f_dentry->d_inode->i_ino) {
 931                         de = mfd->mfd_file->f_dentry;
 932                         spin_unlock(&med->med_open_lock);
 933                         CERROR("Re opening "LPD64"\n", body->fid1.id);
 934                         GOTO(out_pack, rc = 0);
 935                 }
 936         }
 937         spin_unlock(&med->med_open_lock);
 938
 939         mfd = kmem_cache_alloc(mds_file_cache, GFP_KERNEL);
 940         if (!mfd) {
 941                 CERROR("mds: out of memory\n");
 942                 req->rq_status = -ENOMEM;
 943                 RETURN(0);
 944         }
 945
 946         de = mds_fid2dentry(mds, &body->fid1, &mnt);
 947         if (IS_ERR(de))
 948                 GOTO(out_free, rc = PTR_ERR(de));
 949
 950         /* check if this inode has seen a delayed object creation */
 951         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MDS_OPEN_HAS_EA) {
 952                 rc = mds_store_md(mds, req, 1, body, de->d_inode);
 953                 if (rc) {
 954                         l_dput(de);
 955                         mntput(mnt);
 956                         GOTO(out_free, rc);
 957                 }
 958         }
 959
 960         flags = body->flags;
 961         /* dentry_open does a dput(de) and mntput(mnt) on error */
 962         file = dentry_open(de, mnt, flags & ~O_DIRECT);
 963         if (IS_ERR(file)) {
 964                 rc = PTR_ERR(file);
 965                 GOTO(out_free, 0);
 966         }
 967
 968         file->private_data = mfd;
 969         mfd->mfd_file = file;
 970         memcpy(&mfd->mfd_clienthandle, &body->handle, sizeof(body->handle));
 971         get_random_bytes(&mfd->mfd_servercookie, sizeof(mfd->mfd_servercookie));
 972         spin_lock(&med->med_open_lock);
 973         list_add(&mfd->mfd_list, &med->med_open_head);
 974         spin_unlock(&med->med_open_lock);
 975
 976 out_pack:
 977         body = lustre_msg_buf(req->rq_repmsg, 0);
 978         mds_pack_inode2fid(&body->fid1, de->d_inode);
 979         mds_pack_inode2body(body, de->d_inode);
 980         body->handle.addr = (__u64)(unsigned long)mfd;
 981         body->handle.cookie = mfd->mfd_servercookie;
 982         CDEBUG(D_INODE, "llite file "LPX64": addr %p, cookie "LPX64"\n",
 983                mfd->mfd_clienthandle.addr, mfd, mfd->mfd_servercookie);
 984         RETURN(0);
 985
 986 out_free:
 987         mfd->mfd_servercookie = DEAD_HANDLE_MAGIC;
 988         kmem_cache_free(mds_file_cache, mfd);
 989         req->rq_status = rc;
 990         RETURN(0);
 991 }
 992
 993 static int mds_close(struct ptlrpc_request *req)
 994 {
 995         struct mds_export_data *med = &req->rq_export->exp_mds_data;
 996         struct mds_body *body;
 997         struct mds_file_data *mfd;
 998         int rc;
 999         ENTRY;
1000
1001         body = lustre_msg_buf(req->rq_reqmsg, 0);
1002
1003         mfd = mds_handle2mfd(&body->handle);
1004         if (!mfd) {
1005                 CERROR("no handle for file close "LPD64
1006                        ": addr "LPX64", cookie "LPX64"\n",
1007                        body->fid1.id, body->handle.addr, body->handle.cookie);
1008                 RETURN(-ESTALE);
1009         }
1010
1011         spin_lock(&med->med_open_lock);
1012         req->rq_status = mds_close_mfd(mfd, med);
1013         spin_unlock(&med->med_open_lock);
1014
1015         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_CLOSE_PACK)) {
1016                 CERROR("test case OBD_FAIL_MDS_CLOSE_PACK\n");
1017                 req->rq_status = -ENOMEM;
1018                 RETURN(-ENOMEM);
1019         }
1020
1021         rc = lustre_pack_msg(0, NULL, NULL, &req->rq_replen, &req->rq_repmsg);
1022         if (rc) {
1023                 CERROR("mds: lustre_pack_msg: rc = %d\n", rc);
1024                 req->rq_status = rc;
1025         }
1026
1027         RETURN(0);
1028 }
1029
1030 static int mds_readpage(struct ptlrpc_request *req)
1031 {
1032         struct mds_obd *mds = mds_req2mds(req);
1033         struct vfsmount *mnt;
1034         struct dentry *de;
1035         struct file *file;
1036         struct mds_body *body, *repbody;
1037         struct obd_run_ctxt saved;
1038         int rc, size = sizeof(*body);
1039         struct obd_ucred uc;
1040         ENTRY;
1041
1042         rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen, &req->rq_repmsg);
1043         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_READPAGE_PACK)) {
1044                 CERROR("mds: out of memory\n");
1045                 GOTO(out, rc = -ENOMEM);
1046         }
1047
1048         body = lustre_msg_buf(req->rq_reqmsg, 0);
1049         uc.ouc_fsuid = body->fsuid;
1050         uc.ouc_fsgid = body->fsgid;
1051         uc.ouc_cap = body->capability;
1052         push_ctxt(&saved, &mds->mds_ctxt, &uc);
1053         de = mds_fid2dentry(mds, &body->fid1, &mnt);
1054         if (IS_ERR(de))
1055                 GOTO(out_pop, rc = PTR_ERR(de));
1056
1057         CDEBUG(D_INODE, "ino %lu\n", de->d_inode->i_ino);
1058
1059         file = dentry_open(de, mnt, O_RDONLY | O_LARGEFILE);
1060         /* note: in case of an error, dentry_open puts dentry */
1061         if (IS_ERR(file))
1062                 GOTO(out_pop, rc = PTR_ERR(file));
1063
1064         repbody = lustre_msg_buf(req->rq_repmsg, 0);
1065         repbody->size = file->f_dentry->d_inode->i_size;
1066         repbody->valid = OBD_MD_FLSIZE;
1067
1068         /* to make this asynchronous make sure that the handling function
1069            doesn't send a reply when this function completes. Instead a
1070            callback function would send the reply */
1071         /* note: in case of an error, dentry_open puts dentry */
1072         rc = mds_sendpage(req, file, body->size);
1073
1074         filp_close(file, 0);
1075 out_pop:
1076         pop_ctxt(&saved, &mds->mds_ctxt, &uc);
1077 out:
1078         req->rq_status = rc;
1079         RETURN(0);
1080 }
1081
1082 int mds_reint(struct ptlrpc_request *req, int offset)
1083 {
1084         int rc;
1085         struct mds_update_record rec;
1086
1087         rc = mds_update_unpack(req, offset, &rec);
1088         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNPACK)) {
1089                 CERROR("invalid record\n");
1090                 req->rq_status = -EINVAL;
1091                 RETURN(0);
1092         }
1093         /* rc will be used to interrupt a for loop over multiple records */
1094         rc = mds_reint_rec(&rec, offset, req);
1095         return rc;
1096 }
1097
1098 /* forward declaration */
1099 int mds_handle(struct ptlrpc_request *req);
1100
1101 static int check_for_next_transno(struct mds_obd *mds)
1102 {
1103         struct ptlrpc_request *req;
1104         req = list_entry(mds->mds_recovery_queue.next,
1105                          struct ptlrpc_request, rq_list);
1106         return req->rq_reqmsg->transno == mds->mds_next_recovery_transno;
1107 }
1108
1109 static void process_recovery_queue(struct mds_obd *mds)
1110 {
1111         struct ptlrpc_request *req;
1112
1113         for (;;) {
1114                 spin_lock(&mds->mds_processing_task_lock);
1115                 req = list_entry(mds->mds_recovery_queue.next,
1116                                  struct ptlrpc_request, rq_list);
1117
1118                 if (req->rq_reqmsg->transno != mds->mds_next_recovery_transno) {
1119                         spin_unlock(&mds->mds_processing_task_lock);
1120                         wait_event(mds->mds_next_transno_waitq,
1121                                    check_for_next_transno(mds));
1122                         continue;
1123                 }
1124                 list_del(&req->rq_list);
1125                 spin_unlock(&mds->mds_processing_task_lock);
1126
1127                 DEBUG_REQ(D_HA, req, "");
1128                 mds_handle(req);
1129
1130                 if (list_empty(&mds->mds_recovery_queue))
1131                         break;
1132         }
1133 }
1134
1135 static int queue_recovery_request(struct ptlrpc_request *req,
1136                                   struct mds_obd *mds)
1137 {
1138         struct list_head *tmp;
1139         int inserted = 0, transno = req->rq_reqmsg->transno;
1140
1141         if (!transno) {
1142                 DEBUG_REQ(D_HA, req, "not queueing");
1143                 return 1;
1144         }
1145
1146         spin_lock(&mds->mds_processing_task_lock);
1147
1148         if (mds->mds_processing_task == current->pid) {
1149                 /* Processing the queue right now, don't re-add. */
1150                 spin_unlock(&mds->mds_processing_task_lock);
1151                 return 1;
1152         }
1153
1154         /* XXX O(n^2) */
1155         list_for_each(tmp, &mds->mds_recovery_queue) {
1156                 struct ptlrpc_request *reqiter =
1157                         list_entry(tmp, struct ptlrpc_request, rq_list);
1158                 if (reqiter->rq_reqmsg->transno > transno) {
1159                         list_add_tail(&req->rq_list, &reqiter->rq_list);
1160                         inserted = 1;
1161                         break;
1162                 }
1163         }
1164
1165         if (!inserted)
1166                 list_add_tail(&req->rq_list, &mds->mds_recovery_queue);
1167
1168         if (mds->mds_processing_task != 0) {
1169                 /* Someone else is processing this queue, we'll leave it to
1170                  * them.
1171                  */
1172                 spin_unlock(&mds->mds_processing_task_lock);
1173                 if (transno == mds->mds_next_recovery_transno)
1174                         wake_up(&mds->mds_next_transno_waitq);
1175                 return 0;
1176         }
1177
1178         /* Nobody is processing, and we know there's (at least) one to process
1179          * now, so we'll do the honours.
1180          */
1181         mds->mds_processing_task = current->pid;
1182         spin_unlock(&mds->mds_processing_task_lock);
1183
1184         process_recovery_queue(mds);
1185         return 0;
1186 }
1187
1188 static int filter_recovery_request(struct ptlrpc_request *req,
1189                                    struct mds_obd *mds, int *process)
1190 {
1191         switch (req->rq_reqmsg->opc) {
1192         case MDS_CONNECT:
1193         case MDS_DISCONNECT:
1194         case MDS_OPEN:
1195                *process = 1;
1196                RETURN(0);
1197
1198         case MDS_GETSTATUS: /* used in unmounting */
1199         case MDS_REINT:
1200         case LDLM_ENQUEUE:
1201                 *process = queue_recovery_request(req, mds);
1202                 RETURN(0);
1203
1204         default:
1205                 DEBUG_REQ(D_ERROR, req, "not permitted during recovery");
1206                 *process = 0;
1207                 RETURN(ptlrpc_error(req->rq_svc, req));
1208         }
1209 }
1210
1211 static int mds_queue_final_reply(struct ptlrpc_request *req, int rc)
1212 {
1213         struct mds_obd *mds = mds_req2mds(req);
1214
1215         if (rc) {
1216                 /* Just like ptlrpc_error, but without the sending. */
1217                 lustre_pack_msg(0, NULL, NULL, &req->rq_replen,
1218                                 &req->rq_repmsg);
1219                 req->rq_type = PTL_RPC_MSG_ERR;
1220         }
1221
1222         list_add(&req->rq_list, &mds->mds_delayed_reply_queue);
1223         if (--mds->mds_recoverable_clients == 0) {
1224                 struct list_head *tmp, *n;
1225
1226                 CDEBUG(D_HA,
1227                        "all clients recovered, sending delayed replies\n");
1228                 list_for_each_safe(tmp, n, &mds->mds_delayed_reply_queue) {
1229                         req = list_entry(tmp, struct ptlrpc_request, rq_list);
1230                         DEBUG_REQ(D_HA, req, "delayed:");
1231                         ptlrpc_reply(req->rq_svc, req);
1232                 }
1233         } else {
1234                 CDEBUG(D_HA, "%d recoverable clients remain\n",
1235                        mds->mds_recoverable_clients);
1236         }
1237
1238         return 1;
1239 }
1240
1241 static char *reint_names[] = {
1242         [REINT_SETATTR] "setattr",
1243         [REINT_CREATE]  "create",
1244         [REINT_LINK]    "link",
1245         [REINT_UNLINK]  "unlink",
1246         [REINT_RENAME]  "rename"
1247 };
1248
1249 int mds_handle(struct ptlrpc_request *req)
1250 {
1251         int rc;
1252         int should_process;
1253         struct mds_obd *mds = NULL; /* quell gcc overwarning */
1254         ENTRY;
1255
1256         rc = lustre_unpack_msg(req->rq_reqmsg, req->rq_reqlen);
1257         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_HANDLE_UNPACK)) {
1258                 CERROR("lustre_mds: Invalid request\n");
1259                 GOTO(out, rc);
1260         }
1261
1262         LASSERT(!strcmp(req->rq_obd->obd_type->typ_name, LUSTRE_MDT_NAME));
1263
1264         if (req->rq_reqmsg->opc != MDS_CONNECT) {
1265                 if (req->rq_export == NULL)
1266                         GOTO(out, rc = -ENOTCONN);
1267
1268                 mds = mds_req2mds(req);
1269                 if (mds->mds_recoverable_clients != 0) {
1270                         rc = filter_recovery_request(req, mds, &should_process);
1271                         if (rc || !should_process)
1272                                 RETURN(rc);
1273                 }
1274         }
1275
1276         switch (req->rq_reqmsg->opc) {
1277         case MDS_CONNECT:
1278                 DEBUG_REQ(D_INODE, req, "connect");
1279                 OBD_FAIL_RETURN(OBD_FAIL_MDS_CONNECT_NET, 0);
1280                 rc = target_handle_connect(req);
1281                 /* Make sure that last_rcvd is correct. */
1282                 if (!rc) {
1283                         /* Now that we have an export, set mds. */
1284                         mds = mds_req2mds(req);
1285                         mds_fsync_super(mds->mds_sb);
1286                 }
1287                 break;
1288
1289         case MDS_DISCONNECT:
1290                 DEBUG_REQ(D_INODE, req, "disconnect");
1291                 OBD_FAIL_RETURN(OBD_FAIL_MDS_DISCONNECT_NET, 0);
1292                 rc = target_handle_disconnect(req);
1293                 /* Make sure that last_rcvd is correct. */
1294                 if (!rc)
1295                         mds_fsync_super(mds->mds_sb);
1296                 goto out;
1297
1298         case MDS_GETSTATUS:
1299                 DEBUG_REQ(D_INODE, req, "getstatus");
1300                 OBD_FAIL_RETURN(OBD_FAIL_MDS_GETSTATUS_NET, 0);
1301                 rc = mds_getstatus(req);
1302                 break;
1303
1304         case MDS_GETLOVINFO:
1305                 DEBUG_REQ(D_INODE, req, "getlovinfo");
1306                 rc = mds_getlovinfo(req);
1307                 break;
1308
1309         case MDS_GETATTR:
1310                 DEBUG_REQ(D_INODE, req, "getattr");
1311                 OBD_FAIL_RETURN(OBD_FAIL_MDS_GETATTR_NET, 0);
1312                 rc = mds_getattr(0, req);
1313                 break;
1314
1315         case MDS_STATFS:
1316                 DEBUG_REQ(D_INODE, req, "statfs");
1317                 OBD_FAIL_RETURN(OBD_FAIL_MDS_STATFS_NET, 0);
1318                 rc = mds_statfs(req);
1319                 break;
1320
1321         case MDS_READPAGE:
1322                 DEBUG_REQ(D_INODE, req, "readpage\n");
1323                 OBD_FAIL_RETURN(OBD_FAIL_MDS_READPAGE_NET, 0);
1324                 rc = mds_readpage(req);
1325
1326                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE))
1327                         return 0;
1328                 break;
1329
1330         case MDS_REINT: {
1331                 int size = sizeof(struct mds_body);
1332                 int opc = *(u32 *)lustre_msg_buf(req->rq_reqmsg, 0),
1333                         realopc = opc & REINT_OPCODE_MASK;
1334
1335                 DEBUG_REQ(D_INODE, req, "reint (%s%s)",
1336                           reint_names[realopc],
1337                           opc & REINT_REPLAYING ? "|REPLAYING" : "");
1338
1339                 OBD_FAIL_RETURN(OBD_FAIL_MDS_REINT_NET, 0);
1340
1341                 rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen,
1342                                      &req->rq_repmsg);
1343                 if (rc) {
1344                         req->rq_status = rc;
1345                         break;
1346                 }
1347                 rc = mds_reint(req, 0);
1348                 OBD_FAIL_RETURN(OBD_FAIL_MDS_REINT_NET_REP, 0);
1349                 break;
1350                 }
1351
1352         case MDS_OPEN:
1353                 DEBUG_REQ(D_INODE, req, "open");
1354                 OBD_FAIL_RETURN(OBD_FAIL_MDS_OPEN_NET, 0);
1355                 rc = mds_open(req);
1356                 break;
1357
1358         case MDS_CLOSE:
1359                 DEBUG_REQ(D_INODE, req, "close");
1360                 OBD_FAIL_RETURN(OBD_FAIL_MDS_CLOSE_NET, 0);
1361                 rc = mds_close(req);
1362                 break;
1363
1364         case LDLM_ENQUEUE:
1365                 DEBUG_REQ(D_INODE, req, "enqueue");
1366                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_ENQUEUE, 0);
1367                 rc = ldlm_handle_enqueue(req);
1368                 break;
1369         case LDLM_CONVERT:
1370                 DEBUG_REQ(D_INODE, req, "convert");
1371                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CONVERT, 0);
1372                 rc = ldlm_handle_convert(req);
1373                 break;
1374         case LDLM_BL_CALLBACK:
1375         case LDLM_CP_CALLBACK:
1376                 DEBUG_REQ(D_INODE, req, "callback");
1377                 CERROR("callbacks should not happen on MDS\n");
1378                 LBUG();
1379                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_BL_CALLBACK, 0);
1380                 break;
1381         default:
1382                 rc = ptlrpc_error(req->rq_svc, req);
1383                 RETURN(rc);
1384         }
1385
1386         EXIT;
1387
1388         if (!rc) {
1389                 struct mds_export_data *med = &req->rq_export->exp_mds_data;
1390
1391                 req->rq_repmsg->last_xid =
1392                         HTON__u64(le64_to_cpu(med->med_mcd->mcd_last_xid));
1393                 req->rq_repmsg->last_committed =
1394                         HTON__u64(mds->mds_last_committed);
1395                 CDEBUG(D_INFO, "last_rcvd ~%Lu, last_committed %Lu, xid %d\n",
1396                        (unsigned long long)mds->mds_last_rcvd,
1397                        (unsigned long long)mds->mds_last_committed,
1398                        cpu_to_le32(req->rq_xid));
1399         }
1400  out:
1401
1402         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_LAST_REPLAY) {
1403                 struct mds_obd *mds = mds_req2mds(req);
1404                 LASSERT(mds->mds_recoverable_clients);
1405                 DEBUG_REQ(D_HA, req, "LAST_REPLAY, queuing reply");
1406                 return mds_queue_final_reply(req, rc);
1407         }
1408
1409         /* MDS_CONNECT / EALREADY (note: not -EALREADY!) isn't an error */
1410         if (rc && (req->rq_reqmsg->opc != MDS_CONNECT ||
1411                    rc != EALREADY)) {
1412                 CERROR("mds: processing error (opcode %d): %d\n",
1413                        req->rq_reqmsg->opc, rc);
1414                 ptlrpc_error(req->rq_svc, req);
1415         } else {
1416                 CDEBUG(D_NET, "sending reply\n");
1417                 ptlrpc_reply(req->rq_svc, req);
1418         }
1419         return 0;
1420 }
1421
1422 /* Update the server data on disk.  This stores the new mount_count and
1423  * also the last_rcvd value to disk.  If we don't have a clean shutdown,
1424  * then the server last_rcvd value may be less than that of the clients.
1425  * This will alert us that we may need to do client recovery.
1426  *
1427  * Assumes we are already in the server filesystem context.
1428  *
1429  * Also assumes for mds_last_rcvd that we are not modifying it (no locking).
1430  */
1431 int mds_update_server_data(struct mds_obd *mds)
1432 {
1433         struct mds_server_data *msd = mds->mds_server_data;
1434         struct file *filp = mds->mds_rcvd_filp;
1435         loff_t off = 0;
1436         int rc;
1437
1438         msd->msd_last_rcvd = cpu_to_le64(mds->mds_last_rcvd);
1439         msd->msd_mount_count = cpu_to_le64(mds->mds_mount_count);
1440
1441         CDEBUG(D_SUPER, "MDS mount_count is %Lu, last_rcvd is %Lu\n",
1442                (unsigned long long)mds->mds_mount_count,
1443                (unsigned long long)mds->mds_last_rcvd);
1444         rc = lustre_fwrite(filp, (char *)msd, sizeof(*msd), &off);
1445         if (rc != sizeof(*msd)) {
1446                 CERROR("error writing MDS server data: rc = %d\n", rc);
1447                 if (rc > 0)
1448                         RETURN(-EIO);
1449                 RETURN(rc);
1450         }
1451 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
1452         rc = fsync_dev(filp->f_dentry->d_inode->i_rdev);
1453 #else
1454         rc = file_fsync(filp,  filp->f_dentry, 1);
1455 #endif
1456         if (rc)
1457                 CERROR("error flushing MDS server data: rc = %d\n", rc);
1458
1459         return 0;
1460 }
1461
1462 /* Do recovery actions for the MDS */
1463 static int mds_recovery_complete(struct obd_device *obddev)
1464 {
1465         struct mds_obd *mds = &obddev->u.mds;
1466         struct obd_run_ctxt saved;
1467         int rc;
1468
1469         LASSERT(mds->mds_recoverable_clients == 0);
1470
1471         /* This happens at the end when recovery is complete */
1472         ++mds->mds_mount_count;
1473         push_ctxt(&saved, &mds->mds_ctxt, NULL);
1474         rc = mds_update_server_data(mds);
1475         pop_ctxt(&saved, &mds->mds_ctxt, NULL);
1476
1477         return rc;
1478 }
1479
1480 /* mount the file system (secretly) */
1481 static int mds_setup(struct obd_device *obddev, obd_count len, void *buf)
1482 {
1483         struct obd_ioctl_data* data = buf;
1484         struct mds_obd *mds = &obddev->u.mds;
1485         struct vfsmount *mnt;
1486         int rc = 0;
1487         ENTRY;
1488
1489         MOD_INC_USE_COUNT;
1490 #ifdef CONFIG_DEV_RDONLY
1491         dev_clear_rdonly(2);
1492 #endif
1493         if (!data->ioc_inlbuf1 || !data->ioc_inlbuf2)
1494                 GOTO(err_dec, rc = -EINVAL);
1495
1496         obddev->obd_fsops = fsfilt_get_ops(data->ioc_inlbuf2);
1497         if (IS_ERR(obddev->obd_fsops))
1498                 GOTO(err_dec, rc = PTR_ERR(obddev->obd_fsops));
1499
1500         mnt = do_kern_mount(data->ioc_inlbuf2, 0, data->ioc_inlbuf1, NULL);
1501         if (IS_ERR(mnt)) {
1502                 rc = PTR_ERR(mnt);
1503                 CERROR("do_kern_mount failed: rc = %d\n", rc);
1504                 GOTO(err_ops, rc);
1505         }
1506
1507         CDEBUG(D_SUPER, "%s: mnt = %p\n", data->ioc_inlbuf1, mnt);
1508         mds->mds_sb = mnt->mnt_root->d_inode->i_sb;
1509         if (!mds->mds_sb)
1510                 GOTO(err_put, rc = -ENODEV);
1511
1512         init_MUTEX(&mds->mds_transno_sem);
1513         mds->mds_max_mdsize = sizeof(struct lov_mds_md);
1514         rc = mds_fs_setup(obddev, mnt);
1515         if (rc) {
1516                 CERROR("MDS filesystem method init failed: rc = %d\n", rc);
1517                 GOTO(err_put, rc);
1518         }
1519
1520         obddev->obd_namespace =
1521                 ldlm_namespace_new("mds_server", LDLM_NAMESPACE_SERVER);
1522         if (obddev->obd_namespace == NULL) {
1523                 mds_cleanup(obddev);
1524                 GOTO(err_fs, rc = -ENOMEM);
1525         }
1526
1527         ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL,
1528                            "mds_ldlm_client", &obddev->obd_ldlm_client);
1529
1530         spin_lock_init(&mds->mds_processing_task_lock);
1531         mds->mds_processing_task = 0;
1532         INIT_LIST_HEAD(&mds->mds_recovery_queue);
1533         INIT_LIST_HEAD(&mds->mds_delayed_reply_queue);
1534
1535         RETURN(0);
1536
1537 err_fs:
1538         mds_fs_cleanup(obddev);
1539 err_put:
1540         unlock_kernel();
1541         mntput(mds->mds_vfsmnt);
1542         mds->mds_sb = 0;
1543         lock_kernel();
1544 err_ops:
1545         fsfilt_put_ops(obddev->obd_fsops);
1546 err_dec:
1547         MOD_DEC_USE_COUNT;
1548         RETURN(rc);
1549 }
1550
1551 static int mds_cleanup(struct obd_device *obddev)
1552 {
1553         struct super_block *sb;
1554         struct mds_obd *mds = &obddev->u.mds;
1555         struct obd_run_ctxt saved;
1556         ENTRY;
1557
1558         sb = mds->mds_sb;
1559         if (!mds->mds_sb)
1560                 RETURN(0);
1561
1562         push_ctxt(&saved, &mds->mds_ctxt, NULL);
1563         mds_update_server_data(mds);
1564
1565         if (mds->mds_rcvd_filp) {
1566                 int rc = filp_close(mds->mds_rcvd_filp, 0);
1567                 mds->mds_rcvd_filp = NULL;
1568
1569                 if (rc)
1570                         CERROR("last_rcvd file won't close, rc=%d\n", rc);
1571         }
1572         pop_ctxt(&saved, &mds->mds_ctxt, NULL);
1573
1574         unlock_kernel();
1575         mntput(mds->mds_vfsmnt);
1576         mds->mds_sb = 0;
1577
1578         ldlm_namespace_free(obddev->obd_namespace);
1579
1580         lock_kernel();
1581 #ifdef CONFIG_DEV_RDONLY
1582         dev_clear_rdonly(2);
1583 #endif
1584         mds_fs_cleanup(obddev);
1585         fsfilt_put_ops(obddev->obd_fsops);
1586
1587         MOD_DEC_USE_COUNT;
1588         RETURN(0);
1589 }
1590
1591 static int ldlm_intent_policy(struct ldlm_lock *lock, void *req_cookie,
1592                               ldlm_mode_t mode, int flags, void *data)
1593 {
1594         struct ptlrpc_request *req = req_cookie;
1595         int rc = 0;
1596         ENTRY;
1597
1598         if (!req_cookie)
1599                 RETURN(0);
1600
1601         if (req->rq_reqmsg->bufcount > 1) {
1602                 /* an intent needs to be considered */
1603                 struct ldlm_intent *it = lustre_msg_buf(req->rq_reqmsg, 1);
1604                 struct mds_obd *mds= &req->rq_export->exp_obd->u.mds;
1605                 struct mds_body *mds_rep;
1606                 struct ldlm_reply *rep;
1607                 __u64 new_resid[3] = {0, 0, 0}, old_res;
1608                 int rc, size[3] = {sizeof(struct ldlm_reply),
1609                                                   sizeof(struct mds_body),
1610                                                   mds->mds_max_mdsize};
1611
1612                 it->opc = NTOH__u64(it->opc);
1613
1614                 LDLM_DEBUG(lock, "intent policy, opc: %s",
1615                            ldlm_it2str(it->opc));
1616
1617                 rc = lustre_pack_msg(3, size, NULL, &req->rq_replen,
1618                                      &req->rq_repmsg);
1619                 if (rc) {
1620                         rc = req->rq_status = -ENOMEM;
1621                         RETURN(rc);
1622                 }
1623
1624                 rep = lustre_msg_buf(req->rq_repmsg, 0);
1625                 rep->lock_policy_res1 = 1;
1626
1627                 /* execute policy */
1628                 switch ((long)it->opc) {
1629                 case IT_CREAT|IT_OPEN:
1630                         rc = mds_reint(req, 2);
1631                         if (rc || (req->rq_status != 0 &&
1632                                    req->rq_status != -EEXIST)) {
1633                                 rep->lock_policy_res2 = req->rq_status;
1634                                 RETURN(ELDLM_LOCK_ABORTED);
1635                         }
1636                         break;
1637                 case IT_CREAT:
1638                 case IT_MKDIR:
1639                 case IT_MKNOD:
1640                 case IT_RENAME2:
1641                 case IT_LINK2:
1642                 case IT_RMDIR:
1643                 case IT_SYMLINK:
1644                 case IT_UNLINK:
1645                         rc = mds_reint(req, 2);
1646                         if (rc || (req->rq_status != 0 &&
1647                                    req->rq_status != -EISDIR &&
1648                                    req->rq_status != -ENOTDIR)) {
1649                                 rep->lock_policy_res2 = req->rq_status;
1650                                 RETURN(ELDLM_LOCK_ABORTED);
1651                         }
1652                         break;
1653                 case IT_GETATTR:
1654                 case IT_LOOKUP:
1655                 case IT_OPEN:
1656                 case IT_READDIR:
1657                 case IT_READLINK:
1658                 case IT_RENAME:
1659                 case IT_LINK:
1660                 case IT_SETATTR:
1661                         rc = mds_getattr_name(2, req);
1662                         /* FIXME: we need to sit down and decide on who should
1663                          * set req->rq_status, who should return negative and
1664                          * positive return values, and what they all mean. */
1665                         if (rc || req->rq_status != 0) {
1666                                 rep->lock_policy_res2 = req->rq_status;
1667                                 RETURN(ELDLM_LOCK_ABORTED);
1668                         }
1669                         break;
1670                 case IT_READDIR|IT_OPEN:
1671                         LBUG();
1672                         break;
1673                 default:
1674                         CERROR("Unhandled intent "LPD64"\n", it->opc);
1675                         LBUG();
1676                 }
1677
1678                 /* We don't bother returning a lock to the client for a file
1679                  * or directory we are removing.
1680                  *
1681                  * As for link and rename, there is no reason for the client
1682                  * to get a lock on the target at this point.  If they are
1683                  * going to modify the file/directory later they will get a
1684                  * lock at that time.
1685                  */
1686                 if (it->opc & (IT_UNLINK | IT_RMDIR | IT_LINK | IT_LINK2 |
1687                                IT_RENAME | IT_RENAME2))
1688                         RETURN(ELDLM_LOCK_ABORTED);
1689
1690                 rep->lock_policy_res2 = req->rq_status;
1691                 mds_rep = lustre_msg_buf(req->rq_repmsg, 1);
1692
1693                 /* If the client is about to open a file that doesn't have an MD
1694                  * stripe record, it's going to need a write lock. */
1695                 if (it->opc & IT_OPEN && !(mds_rep->valid & OBD_MD_FLEASIZE)) {
1696                         LDLM_DEBUG(lock, "open with no EA; returning PW lock");
1697                         lock->l_req_mode = LCK_PW;
1698                 }
1699
1700                 if (flags & LDLM_FL_INTENT_ONLY) {
1701                         LDLM_DEBUG(lock, "INTENT_ONLY, aborting lock");
1702                         RETURN(ELDLM_LOCK_ABORTED);
1703                 }
1704                 /* Give the client a lock on the child object, instead of the
1705                  * parent that it requested. */
1706                 new_resid[0] = NTOH__u32(mds_rep->ino);
1707                 new_resid[1] = NTOH__u32(mds_rep->generation);
1708                 if (new_resid[0] == 0)
1709                         LBUG();
1710                 old_res = lock->l_resource->lr_name[0];
1711
1712                 ldlm_lock_change_resource(lock, new_resid);
1713                 if (lock->l_resource == NULL) {
1714                         LBUG();
1715                         RETURN(-ENOMEM);
1716                 }
1717                 LDLM_DEBUG(lock, "intent policy, old res %ld",
1718                            (long)old_res);
1719                 RETURN(ELDLM_LOCK_CHANGED);
1720         } else {
1721                 int size = sizeof(struct ldlm_reply);
1722                 rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen,
1723                                      &req->rq_repmsg);
1724                 if (rc) {
1725                         LBUG();
1726                         RETURN(-ENOMEM);
1727                 }
1728         }
1729         RETURN(rc);
1730 }
1731
1732 int mds_attach(struct obd_device *dev, obd_count len, void *data)
1733 {
1734         return lprocfs_reg_obd(dev, status_var_nm_1, dev);
1735 }
1736
1737 int mds_detach(struct obd_device *dev)
1738 {
1739         return lprocfs_dereg_obd(dev);
1740 }
1741
1742 static int mdt_setup(struct obd_device *obddev, obd_count len, void *buf)
1743 {
1744         int i;
1745         //        struct obd_ioctl_data* data = buf;
1746         struct mds_obd *mds = &obddev->u.mds;
1747         int rc = 0;
1748         ENTRY;
1749
1750         MOD_INC_USE_COUNT;
1751
1752         mds->mds_service = ptlrpc_init_svc(MDS_NEVENTS, MDS_NBUFS,
1753                                            MDS_BUFSIZE, MDS_MAXREQSIZE,
1754                                            MDS_REQUEST_PORTAL, MDC_REPLY_PORTAL,
1755                                            "self", mds_handle, "mds");
1756         if (!mds->mds_service) {
1757                 CERROR("failed to start service\n");
1758                 GOTO(err_dec, rc = -ENOMEM);
1759         }
1760
1761         for (i = 0; i < MDT_NUM_THREADS; i++) {
1762                 char name[32];
1763                 sprintf(name, "ll_mdt_%02d", i);
1764                 rc = ptlrpc_start_thread(obddev, mds->mds_service, name);
1765                 if (rc) {
1766                         CERROR("cannot start MDT thread #%d: rc %d\n", i, rc);
1767                         GOTO(err_thread, rc);
1768                 }
1769         }
1770
1771         RETURN(0);
1772
1773 err_thread:
1774         ptlrpc_stop_all_threads(mds->mds_service);
1775         ptlrpc_unregister_service(mds->mds_service);
1776 err_dec:
1777         MOD_DEC_USE_COUNT;
1778         RETURN(rc);
1779 }
1780
1781
1782 static int mdt_cleanup(struct obd_device *obddev)
1783 {
1784         struct mds_obd *mds = &obddev->u.mds;
1785         ENTRY;
1786
1787         ptlrpc_stop_all_threads(mds->mds_service);
1788         ptlrpc_unregister_service(mds->mds_service);
1789
1790         MOD_DEC_USE_COUNT;
1791         RETURN(0);
1792 }
1793
1794 extern int mds_iocontrol(unsigned int cmd, struct lustre_handle *conn,
1795                          int len, void *karg, void *uarg);
1796
1797 /* use obd ops to offer management infrastructure */
1798 static struct obd_ops mds_obd_ops = {
1799         o_attach:      mds_attach,
1800         o_detach:      mds_detach,
1801         o_connect:     mds_connect,
1802         o_disconnect:  mds_disconnect,
1803         o_setup:       mds_setup,
1804         o_cleanup:     mds_cleanup,
1805         o_iocontrol:   mds_iocontrol
1806 };
1807
1808 static struct obd_ops mdt_obd_ops = {
1809         o_setup:       mdt_setup,
1810         o_cleanup:     mdt_cleanup,
1811 };
1812
1813
1814 static int __init mds_init(void)
1815 {
1816         mds_file_cache = kmem_cache_create("ll_mds_file_data",
1817                                            sizeof(struct mds_file_data),
1818                                            0, 0, NULL, NULL);
1819         if (mds_file_cache == NULL)
1820                 return -ENOMEM;
1821
1822         class_register_type(&mds_obd_ops, status_class_var, LUSTRE_MDS_NAME);
1823         class_register_type(&mdt_obd_ops, 0, LUSTRE_MDT_NAME);
1824         ldlm_register_intent(ldlm_intent_policy);
1825
1826         return 0;
1827 }
1828
1829 static void __exit mds_exit(void)
1830 {
1831         ldlm_unregister_intent();
1832         class_unregister_type(LUSTRE_MDS_NAME);
1833         class_unregister_type(LUSTRE_MDT_NAME);
1834         if (kmem_cache_destroy(mds_file_cache))
1835                 CERROR("couldn't free MDS file cache\n");
1836 }
1837
1838 MODULE_AUTHOR("Cluster File Systems <info@clusterfs.com>");
1839 MODULE_DESCRIPTION("Lustre Metadata Server (MDS) v0.01");
1840 MODULE_LICENSE("GPL");
1841
1842 module_init(mds_init);
1843 module_exit(mds_exit);