lustre/mds/handler.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 2001, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/mds/handler.c
  37  *
  38  * Author: Peter Braam <braam@clusterfs.com>
  39  * Author: Andreas Dilger <adilger@clusterfs.com>
  40  * Author: Phil Schwan <phil@clusterfs.com>
  41  * Author: Mike Shaver <shaver@clusterfs.com>
  42  */
  43
  44 #ifndef EXPORT_SYMTAB
  45 # define EXPORT_SYMTAB
  46 #endif
  47 #define DEBUG_SUBSYSTEM S_MDS
  48
  49 #include <lustre_mds.h>
  50 #include <linux/module.h>
  51 #include <linux/init.h>
  52 #include <linux/random.h>
  53 #include <linux/fs.h>
  54 #include <linux/jbd.h>
  55 # include <linux/smp_lock.h>
  56 # include <linux/buffer_head.h>
  57 # include <linux/workqueue.h>
  58 # include <linux/mount.h>
  59
  60 #include <obd_class.h>
  61 #include <lustre_dlm.h>
  62 #include <obd_lov.h>
  63 #include <lustre_fsfilt.h>
  64 #include <lprocfs_status.h>
  65 #include <lustre_quota.h>
  66 #include <lustre_disk.h>
  67 #include <lustre_param.h>
  68
  69 #include "mds_internal.h"
  70
  71 int mds_num_threads;
  72 CFS_MODULE_PARM(mds_num_threads, "i", int, 0444,
  73                 "number of MDS service threads to start");
  74
  75 static int mds_intent_policy(struct ldlm_namespace *ns,
  76                              struct ldlm_lock **lockp, void *req_cookie,
  77                              ldlm_mode_t mode, int flags, void *data);
  78 static int mds_postsetup(struct obd_device *obd);
  79 static int mds_cleanup(struct obd_device *obd);
  80
  81 /* Assumes caller has already pushed into the kernel filesystem context */
  82 static int mds_sendpage(struct ptlrpc_request *req, struct file *file,
  83                         loff_t offset, int count)
  84 {
  85         struct ptlrpc_bulk_desc *desc;
  86         struct obd_export *exp = req->rq_export;
  87         struct l_wait_info lwi;
  88         struct page **pages;
  89         int timeout;
  90         int rc = 0, npages, i, tmpcount, tmpsize = 0;
  91         ENTRY;
  92
  93         LASSERT((offset & ~CFS_PAGE_MASK) == 0); /* I'm dubious about this */
  94
  95         npages = (count + CFS_PAGE_SIZE - 1) >> CFS_PAGE_SHIFT;
  96         OBD_ALLOC(pages, sizeof(*pages) * npages);
  97         if (!pages)
  98                 GOTO(out, rc = -ENOMEM);
  99
 100         desc = ptlrpc_prep_bulk_exp(req, npages, BULK_PUT_SOURCE,
 101                                     MDS_BULK_PORTAL);
 102         if (desc == NULL)
 103                 GOTO(out_free, rc = -ENOMEM);
 104
 105         for (i = 0, tmpcount = count; i < npages; i++, tmpcount -= tmpsize) {
 106                 tmpsize = tmpcount > CFS_PAGE_SIZE ? CFS_PAGE_SIZE : tmpcount;
 107
 108                 OBD_PAGE_ALLOC(pages[i], CFS_ALLOC_STD);
 109                 if (pages[i] == NULL)
 110                         GOTO(cleanup_buf, rc = -ENOMEM);
 111
 112                 ptlrpc_prep_bulk_page(desc, pages[i], 0, tmpsize);
 113         }
 114
 115         for (i = 0, tmpcount = count; i < npages; i++, tmpcount -= tmpsize) {
 116                 tmpsize = tmpcount > CFS_PAGE_SIZE ? CFS_PAGE_SIZE : tmpcount;
 117                 CDEBUG(D_EXT2, "reading %u@%llu from dir %lu (size %llu)\n",
 118                        tmpsize, offset, file->f_dentry->d_inode->i_ino,
 119                        i_size_read(file->f_dentry->d_inode));
 120
 121                 rc = fsfilt_readpage(exp->exp_obd, file,
 122                                      kmap(pages[i]), tmpsize, &offset);
 123                 kunmap(pages[i]);
 124
 125                 if (rc != tmpsize)
 126                         GOTO(cleanup_buf, rc = -EIO);
 127         }
 128
 129         LASSERT(desc->bd_nob == count);
 130
 131         rc = ptlrpc_start_bulk_transfer(desc);
 132         if (rc)
 133                 GOTO(cleanup_buf, rc);
 134
 135         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE)) {
 136                 CERROR("obd_fail_loc=%x, fail operation rc=%d\n",
 137                        OBD_FAIL_MDS_SENDPAGE, rc);
 138                 GOTO(abort_bulk, rc);
 139         }
 140
 141         timeout = (int)req->rq_deadline - (int)cfs_time_current_sec();
 142         if (timeout < 0) {
 143                 CERROR("Req deadline already passed %lu (now: %lu)\n",
 144                        req->rq_deadline, cfs_time_current_sec());
 145         }
 146         lwi = LWI_TIMEOUT_INTERVAL(cfs_time_seconds(max(timeout, 1)),
 147                                    cfs_time_seconds(1), NULL, NULL);
 148         rc = l_wait_event(desc->bd_waitq, !ptlrpc_server_bulk_active(desc) ||
 149                           exp->exp_failed || exp->exp_abort_active_req, &lwi);
 150         LASSERT (rc == 0 || rc == -ETIMEDOUT);
 151
 152         if (rc == 0) {
 153                 if (desc->bd_success &&
 154                     desc->bd_nob_transferred == count)
 155                         GOTO(cleanup_buf, rc);
 156                 rc = -ETIMEDOUT;
 157                 if (exp->exp_abort_active_req || exp->exp_failed)
 158                         GOTO(abort_bulk, rc);
 159         }
 160
 161         DEBUG_REQ(D_ERROR, req, "bulk failed: %s %d(%d), evicting %s@%s\n",
 162                   (rc == -ETIMEDOUT) ? "timeout" : "network error",
 163                   desc->bd_nob_transferred, count,
 164                   req->rq_export->exp_client_uuid.uuid,
 165                   req->rq_export->exp_connection->c_remote_uuid.uuid);
 166
 167         class_fail_export(req->rq_export);
 168
 169         EXIT;
 170  abort_bulk:
 171         ptlrpc_abort_bulk(desc);
 172  cleanup_buf:
 173         for (i = 0; i < npages; i++)
 174                 if (pages[i])
 175                         OBD_PAGE_FREE(pages[i]);
 176
 177         ptlrpc_free_bulk(desc);
 178  out_free:
 179         OBD_FREE(pages, sizeof(*pages) * npages);
 180  out:
 181         return rc;
 182 }
 183
 184 /* only valid locked dentries or errors should be returned */
 185 struct dentry *mds_fid2locked_dentry(struct obd_device *obd, struct ll_fid *fid,
 186                                      struct vfsmount **mnt, int lock_mode,
 187                                      struct lustre_handle *lockh,
 188                                      char *name, int namelen, __u64 lockpart)
 189 {
 190         struct mds_obd *mds = &obd->u.mds;
 191         struct dentry *de = mds_fid2dentry(mds, fid, mnt), *retval = de;
 192         struct ldlm_res_id res_id = { .name = {0} };
 193         int flags = LDLM_FL_ATOMIC_CB, rc;
 194         ldlm_policy_data_t policy = { .l_inodebits = { lockpart} };
 195         ENTRY;
 196
 197         if (IS_ERR(de))
 198                 RETURN(de);
 199
 200         res_id.name[0] = de->d_inode->i_ino;
 201         res_id.name[1] = de->d_inode->i_generation;
 202         rc = ldlm_cli_enqueue_local(obd->obd_namespace, &res_id,
 203                                     LDLM_IBITS, &policy, lock_mode, &flags,
 204                                     ldlm_blocking_ast, ldlm_completion_ast,
 205                                     NULL, NULL, 0, NULL, lockh);
 206         if (rc != ELDLM_OK) {
 207                 l_dput(de);
 208                 retval = ERR_PTR(-EIO); /* XXX translate ldlm code */
 209         }
 210
 211         RETURN(retval);
 212 }
 213
 214 /* Look up an entry by inode number. */
 215 /* this function ONLY returns valid dget'd dentries with an initialized inode
 216    or errors */
 217 struct dentry *mds_fid2dentry(struct mds_obd *mds, struct ll_fid *fid,
 218                               struct vfsmount **mnt)
 219 {
 220         struct obd_device *obd = container_of(mds, struct obd_device, u.mds);
 221         char fid_name[32];
 222         unsigned long ino = fid->id;
 223         __u32 generation = fid->generation;
 224         struct inode *inode;
 225         struct dentry *result;
 226
 227         if (ino == 0)
 228                 RETURN(ERR_PTR(-ESTALE));
 229
 230         snprintf(fid_name, sizeof(fid_name), "0x%lx", ino);
 231
 232         CDEBUG(D_DENTRY, "--> mds_fid2dentry: ino/gen %lu/%u, sb %p\n",
 233                ino, generation, mds->mds_obt.obt_sb);
 234
 235         /* under ext3 this is neither supposed to return bad inodes
 236            nor NULL inodes. */
 237         result = mds_lookup(obd, fid_name, mds->mds_fid_de, strlen(fid_name));
 238         if (IS_ERR(result))
 239                 RETURN(result);
 240
 241         inode = result->d_inode;
 242         if (!inode)
 243                 RETURN(ERR_PTR(-ENOENT));
 244
 245        if (inode->i_nlink == 0) {
 246                 if (inode->i_mode == 0 &&
 247                     LTIME_S(inode->i_ctime) == 0 ) {
 248                         LCONSOLE_WARN("Found inode with zero nlink, mode and "
 249                                       "ctime -- this may indicate disk"
 250                                       "corruption (device %s, inode %lu, link:"
 251                                       " %lu, count: %d)\n", obd->obd_name, inode->i_ino,
 252                                       (unsigned long)inode->i_nlink,
 253                                       atomic_read(&inode->i_count));
 254                 }
 255                 dput(result);
 256                 RETURN(ERR_PTR(-ENOENT));
 257         }
 258
 259         if (generation && inode->i_generation != generation) {
 260                 /* we didn't find the right inode.. */
 261                 CDEBUG(D_INODE, "found wrong generation: inode %lu, link: %lu, "
 262                        "count: %d, generation %u/%u\n", inode->i_ino,
 263                        (unsigned long)inode->i_nlink,
 264                        atomic_read(&inode->i_count), inode->i_generation,
 265                        generation);
 266                 dput(result);
 267                 RETURN(ERR_PTR(-ENOENT));
 268         }
 269
 270         if (mnt) {
 271                 *mnt = mds->mds_vfsmnt;
 272                 mntget(*mnt);
 273         }
 274
 275         RETURN(result);
 276 }
 277
 278 static int mds_connect_internal(struct obd_export *exp,
 279                                 struct obd_connect_data *data)
 280 {
 281         struct obd_device *obd = exp->exp_obd;
 282         if (data != NULL) {
 283                 data->ocd_connect_flags &= MDS_CONNECT_SUPPORTED;
 284                 data->ocd_ibits_known &= MDS_INODELOCK_FULL;
 285
 286                 /* If no known bits (which should not happen, probably,
 287                    as everybody should support LOOKUP and UPDATE bits at least)
 288                    revert to compat mode with plain locks. */
 289                 if (!data->ocd_ibits_known &&
 290                     data->ocd_connect_flags & OBD_CONNECT_IBITS)
 291                         data->ocd_connect_flags &= ~OBD_CONNECT_IBITS;
 292
 293                 if (!obd->u.mds.mds_fl_acl)
 294                         data->ocd_connect_flags &= ~OBD_CONNECT_ACL;
 295
 296                 if (!obd->u.mds.mds_fl_user_xattr)
 297                         data->ocd_connect_flags &= ~OBD_CONNECT_XATTR;
 298
 299                 exp->exp_connect_flags = data->ocd_connect_flags;
 300                 data->ocd_version = LUSTRE_VERSION_CODE;
 301                 exp->exp_mds_data.med_ibits_known = data->ocd_ibits_known;
 302         }
 303
 304         if (obd->u.mds.mds_fl_acl &&
 305             ((exp->exp_connect_flags & OBD_CONNECT_ACL) == 0)) {
 306                 CWARN("%s: MDS requires ACL support but client does not\n",
 307                       obd->obd_name);
 308                 return -EBADE;
 309         }
 310         return 0;
 311 }
 312
 313 static int mds_reconnect(struct obd_export *exp, struct obd_device *obd,
 314                          struct obd_uuid *cluuid,
 315                          struct obd_connect_data *data,
 316                          void *localdata)
 317 {
 318         int rc;
 319         ENTRY;
 320
 321         if (exp == NULL || obd == NULL || cluuid == NULL)
 322                 RETURN(-EINVAL);
 323
 324         rc = mds_connect_internal(exp, data);
 325         if (rc == 0)
 326                 mds_export_stats_init(obd, exp, 1, localdata);
 327
 328         RETURN(rc);
 329 }
 330
 331 /* Establish a connection to the MDS.
 332  *
 333  * This will set up an export structure for the client to hold state data
 334  * about that client, like open files, the last operation number it did
 335  * on the server, etc.
 336  */
 337 static int mds_connect(struct lustre_handle *conn, struct obd_device *obd,
 338                        struct obd_uuid *cluuid, struct obd_connect_data *data,
 339                        void *localdata)
 340 {
 341         struct obd_export *exp;
 342         struct mds_export_data *med;
 343         struct lsd_client_data *lcd = NULL;
 344         int rc;
 345         ENTRY;
 346
 347         if (!conn || !obd || !cluuid)
 348                 RETURN(-EINVAL);
 349
 350         /* Check for aborted recovery. */
 351         target_recovery_check_and_stop(obd);
 352
 353         /* XXX There is a small race between checking the list and adding a
 354          * new connection for the same UUID, but the real threat (list
 355          * corruption when multiple different clients connect) is solved.
 356          *
 357          * There is a second race between adding the export to the list,
 358          * and filling in the client data below.  Hence skipping the case
 359          * of NULL lcd above.  We should already be controlling multiple
 360          * connects at the client, and we can't hold the spinlock over
 361          * memory allocations without risk of deadlocking.
 362          */
 363         rc = class_connect(conn, obd, cluuid);
 364         if (rc)
 365                 RETURN(rc);
 366         exp = class_conn2export(conn);
 367         LASSERT(exp);
 368         med = &exp->exp_mds_data;
 369
 370         rc = mds_connect_internal(exp, data);
 371         if (rc)
 372                 GOTO(out, rc);
 373
 374         OBD_ALLOC_PTR(lcd);
 375         if (!lcd)
 376                 GOTO(out, rc = -ENOMEM);
 377
 378         memcpy(lcd->lcd_uuid, cluuid, sizeof(lcd->lcd_uuid));
 379         med->med_lcd = lcd;
 380
 381         rc = mds_client_add(obd, exp, -1, localdata);
 382         GOTO(out, rc);
 383
 384 out:
 385         if (rc) {
 386                 if (lcd) {
 387                         OBD_FREE_PTR(lcd);
 388                         med->med_lcd = NULL;
 389                 }
 390                 class_disconnect(exp);
 391         } else {
 392                 class_export_put(exp);
 393         }
 394
 395         RETURN(rc);
 396 }
 397
 398 int mds_init_export(struct obd_export *exp)
 399 {
 400         struct mds_export_data *med = &exp->exp_mds_data;
 401         ENTRY;
 402
 403         INIT_LIST_HEAD(&med->med_open_head);
 404         spin_lock_init(&med->med_open_lock);
 405
 406         spin_lock(&exp->exp_lock);
 407         exp->exp_connecting = 1;
 408         spin_unlock(&exp->exp_lock);
 409
 410         RETURN(ldlm_init_export(exp));
 411 }
 412
 413 static int mds_destroy_export(struct obd_export *exp)
 414 {
 415         ENTRY;
 416
 417         target_destroy_export(exp);
 418         ldlm_destroy_export(exp);
 419
 420         LASSERT(list_empty(&exp->exp_mds_data.med_open_head));
 421         mds_client_free(exp);
 422
 423         RETURN(0);
 424 }
 425
 426 static int mds_cleanup_mfd(struct obd_export *exp)
 427 {
 428         struct mds_export_data *med;
 429         struct obd_device *obd = exp->exp_obd;
 430         struct mds_obd *mds = &obd->u.mds;
 431         struct lvfs_run_ctxt saved;
 432         struct lov_mds_md *lmm;
 433         __u32 lmm_sz, cookie_sz;
 434         struct llog_cookie *logcookies;
 435         struct list_head closing_list;
 436         struct mds_file_data *mfd, *n;
 437         int rc = 0;
 438         ENTRY;
 439
 440         med = &exp->exp_mds_data;
 441
 442         spin_lock(&med->med_open_lock);
 443         if (list_empty(&med->med_open_head)) {
 444                 spin_unlock(&med->med_open_lock);
 445                 RETURN(0);
 446         }
 447
 448         CFS_INIT_LIST_HEAD(&closing_list);
 449         while (!list_empty(&med->med_open_head)) {
 450                 struct list_head *tmp = med->med_open_head.next;
 451                 struct mds_file_data *mfd =
 452                         list_entry(tmp, struct mds_file_data, mfd_list);
 453
 454                 /* Remove mfd handle so it can't be found again.
 455                  * We are consuming the mfd_list reference here. */
 456                 mds_mfd_unlink(mfd, 0);
 457                 list_add_tail(&mfd->mfd_list, &closing_list);
 458         }
 459         spin_unlock(&med->med_open_lock);
 460
 461         lmm_sz = mds->mds_max_mdsize;
 462         OBD_ALLOC(lmm, lmm_sz);
 463         if (lmm == NULL) {
 464                 CWARN("%s: allocation failure during cleanup; can not force "
 465                       "close file handles on this service.\n", obd->obd_name);
 466                 GOTO(out, rc = -ENOMEM);
 467         }
 468
 469         cookie_sz = mds->mds_max_cookiesize;
 470         OBD_ALLOC(logcookies, cookie_sz);
 471         if (logcookies == NULL) {
 472                 CWARN("%s: allocation failure during cleanup; can not force "
 473                       "close file handles on this service.\n", obd->obd_name);
 474                 OBD_FREE(lmm, lmm_sz);
 475                 GOTO(out, rc = -ENOMEM);
 476         }
 477
 478         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 479         /* Close any open files (which may also cause orphan unlinking). */
 480         list_for_each_entry_safe(mfd, n, &closing_list, mfd_list) {
 481                 int lmm_size = lmm_sz;
 482                 umode_t mode = mfd->mfd_dentry->d_inode->i_mode;
 483                 __u64 valid = 0;
 484
 485                 /* If you change this message, be sure to update
 486                  * replay_single:test_46 */
 487                 CDEBUG(D_INODE|D_IOCTL, "%s: force closing file handle for "
 488                        "%.*s (ino %lu)\n", obd->obd_name,
 489                        mfd->mfd_dentry->d_name.len,mfd->mfd_dentry->d_name.name,
 490                        mfd->mfd_dentry->d_inode->i_ino);
 491
 492                 rc = mds_get_md(obd, mfd->mfd_dentry->d_inode, lmm,
 493                                 &lmm_size, 1, 0, 0);
 494                 if (rc < 0)
 495                         CWARN("mds_get_md failure, rc=%d\n", rc);
 496                 else
 497                         valid |= OBD_MD_FLEASIZE;
 498
 499                 /* child orphan sem protects orphan_dec_test and
 500                  * is_orphan race, mds_mfd_close drops it */
 501                 MDS_DOWN_WRITE_ORPHAN_SEM(mfd->mfd_dentry->d_inode);
 502
 503                 list_del_init(&mfd->mfd_list);
 504                 rc = mds_mfd_close(NULL, REQ_REC_OFF, obd, mfd,
 505                                    !(exp->exp_flags & OBD_OPT_FAILOVER),
 506                                    lmm, lmm_size, logcookies,
 507                                    mds->mds_max_cookiesize,
 508                                    &valid);
 509
 510                 if (rc)
 511                         CDEBUG(D_INODE|D_IOCTL, "Error closing file: %d\n", rc);
 512
 513                 if (valid & OBD_MD_FLCOOKIE) {
 514                         rc = mds_osc_destroy_orphan(obd, mode, lmm,
 515                                                     lmm_size, logcookies, 1);
 516                         if (rc < 0) {
 517                                 CDEBUG(D_INODE, "%s: destroy of orphan failed,"
 518                                        " rc = %d\n", obd->obd_name, rc);
 519                                 rc = 0;
 520                         }
 521                         valid &= ~OBD_MD_FLCOOKIE;
 522                 }
 523
 524         }
 525         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 526         OBD_FREE(logcookies, cookie_sz);
 527         OBD_FREE(lmm, lmm_sz);
 528 out:
 529         RETURN(rc);
 530 }
 531
 532 static int mds_disconnect(struct obd_export *exp)
 533 {
 534         int rc = 0;
 535         ENTRY;
 536
 537         LASSERT(exp);
 538         class_export_get(exp);
 539
 540         rc = server_disconnect_export(exp);
 541
 542         rc = mds_cleanup_mfd(exp);
 543
 544         class_export_put(exp);
 545         RETURN(rc);
 546 }
 547
 548 static int mds_getstatus(struct ptlrpc_request *req)
 549 {
 550         struct mds_obd *mds = mds_req2mds(req);
 551         struct mds_body *body;
 552         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
 553         ENTRY;
 554
 555         OBD_FAIL_RETURN(OBD_FAIL_MDS_GETSTATUS_PACK, req->rq_status = -ENOMEM);
 556         rc = lustre_pack_reply(req, 2, size, NULL);
 557         if (rc)
 558                 RETURN(req->rq_status = rc);
 559
 560         body = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof(*body));
 561         memcpy(&body->fid1, &mds->mds_rootfid, sizeof(body->fid1));
 562
 563         /* the last_committed and last_xid fields are filled in for all
 564          * replies already - no need to do so here also.
 565          */
 566         RETURN(0);
 567 }
 568
 569 /* get the LOV EA from @inode and store it into @md.  It can be at most
 570  * @size bytes, and @size is updated with the actual EA size.
 571  * The EA size is also returned on success, and -ve errno on failure.
 572  * If there is no EA then 0 is returned. */
 573 int mds_get_md(struct obd_device *obd, struct inode *inode, void *md,
 574                int *size, int lock, int flags,
 575                __u64 connect_flags)
 576 {
 577         int rc = 0;
 578         int lmm_size = 0;
 579
 580         if (lock)
 581                 LOCK_INODE_MUTEX(inode);
 582         rc = fsfilt_get_md(obd, inode, md, *size, "lov");
 583
 584         if (rc == 0 && flags == MDS_GETATTR)
 585                 rc = mds_get_default_md(obd, md);
 586
 587         if (rc < 0) {
 588                 CERROR("Error %d reading eadata for ino %lu\n",
 589                        rc, inode->i_ino);
 590         } else if (rc > 0) {
 591                 lmm_size = rc;
 592                 rc = mds_convert_lov_ea(obd, inode, md, lmm_size,
 593                                         connect_flags);
 594
 595                 if (rc == 0) {
 596                         *size = lmm_size;
 597                         rc = lmm_size;
 598                 } else if (rc > 0) {
 599                         *size = rc;
 600                 }
 601         } else {
 602                 *size = 0;
 603         }
 604         if (lock)
 605                 UNLOCK_INODE_MUTEX(inode);
 606
 607         RETURN (rc);
 608 }
 609
 610
 611 /* Call with lock=1 if you want mds_pack_md to take the i_mutex.
 612  * Call with lock=0 if the caller has already taken the i_mutex. */
 613 int mds_pack_md(struct obd_device *obd, struct lustre_msg *msg, int offset,
 614                 struct mds_body *body, struct inode *inode, int lock, int flags,
 615                 __u64 connect_flags)
 616 {
 617         struct mds_obd *mds = &obd->u.mds;
 618         void *lmm;
 619         int lmm_size;
 620         int rc;
 621         ENTRY;
 622
 623         lmm = lustre_msg_buf(msg, offset, 0);
 624         if (lmm == NULL) {
 625                 /* Some problem with getting eadata when I sized the reply
 626                  * buffer... */
 627                 CDEBUG(D_INFO, "no space reserved for inode %lu MD\n",
 628                        inode->i_ino);
 629                 RETURN(0);
 630         }
 631         /* if this replay request we should be silencely exist without fill md*/
 632         lmm_size = lustre_msg_buflen(msg, offset);
 633         if (lmm_size == 0)
 634                 RETURN(0);
 635
 636         /* I don't really like this, but it is a sanity check on the client
 637          * MD request.  However, if the client doesn't know how much space
 638          * to reserve for the MD, it shouldn't be bad to have too much space.
 639          */
 640         if (lmm_size > mds->mds_max_mdsize) {
 641                 CWARN("Reading MD for inode %lu of %d bytes > max %d\n",
 642                        inode->i_ino, lmm_size, mds->mds_max_mdsize);
 643                 // RETURN(-EINVAL);
 644         }
 645
 646         rc = mds_get_md(obd, inode, lmm, &lmm_size, lock, flags,
 647                         connect_flags);
 648         if (rc > 0) {
 649                 if (S_ISDIR(inode->i_mode))
 650                         body->valid |= OBD_MD_FLDIREA;
 651                 else
 652                         body->valid |= OBD_MD_FLEASIZE;
 653                 body->eadatasize = lmm_size;
 654                 rc = 0;
 655         }
 656
 657         RETURN(rc);
 658 }
 659
 660 #ifdef CONFIG_FS_POSIX_ACL
 661 static
 662 int mds_pack_posix_acl(struct inode *inode, struct lustre_msg *repmsg,
 663                        struct mds_body *repbody, int repoff)
 664 {
 665         struct dentry de = { .d_inode = inode };
 666         int buflen, rc;
 667         ENTRY;
 668
 669         LASSERT(repbody->aclsize == 0);
 670         LASSERT(lustre_msg_bufcount(repmsg) > repoff);
 671
 672         buflen = lustre_msg_buflen(repmsg, repoff);
 673         if (!buflen)
 674                 GOTO(out, 0);
 675
 676         if (!inode->i_op || !inode->i_op->getxattr)
 677                 GOTO(out, 0);
 678
 679         rc = inode->i_op->getxattr(&de, MDS_XATTR_NAME_ACL_ACCESS,
 680                                    lustre_msg_buf(repmsg, repoff, buflen),
 681                                    buflen);
 682         if (rc >= 0) {
 683                 repbody->aclsize = rc;
 684         } else if (rc != -ENODATA) {
 685                 CERROR("buflen %d, get acl: %d\n", buflen, rc);
 686                 RETURN(rc);
 687         }
 688         EXIT;
 689 out:
 690         repbody->valid |= OBD_MD_FLACL;
 691         return 0;
 692 }
 693 #else
 694 #define mds_pack_posix_acl(inode, repmsg, repbody, repoff) 0
 695 #endif
 696
 697 int mds_pack_acl(struct mds_export_data *med, struct inode *inode,
 698                  struct lustre_msg *repmsg, struct mds_body *repbody,
 699                  int repoff)
 700 {
 701         return mds_pack_posix_acl(inode, repmsg, repbody, repoff);
 702 }
 703
 704 static int mds_getattr_internal(struct obd_device *obd, struct dentry *dentry,
 705                                 struct ptlrpc_request *req,
 706                                 struct mds_body *reqbody, int reply_off)
 707 {
 708         struct mds_body *body;
 709         struct inode *inode = dentry->d_inode;
 710         int rc = 0;
 711         int flags = 0;
 712         ENTRY;
 713
 714         if (inode == NULL)
 715                 RETURN(-ENOENT);
 716
 717         body = lustre_msg_buf(req->rq_repmsg, reply_off, sizeof(*body));
 718         LASSERT(body != NULL);                 /* caller prepped reply */
 719
 720         body->flags = reqbody->flags; /* copy MDS_BFLAG_EXT_FLAGS if present */
 721         mds_pack_inode2body(body, inode);
 722         reply_off++;
 723
 724         if ((S_ISREG(inode->i_mode) && (reqbody->valid & OBD_MD_FLEASIZE)) ||
 725             (S_ISDIR(inode->i_mode) && (reqbody->valid & OBD_MD_FLDIREA))) {
 726                 if (lustre_msg_get_opc(req->rq_reqmsg) == MDS_GETATTR &&
 727                    ((S_ISDIR(inode->i_mode) && (reqbody->valid & OBD_MD_FLDIREA))))
 728                         flags = MDS_GETATTR;
 729
 730                 rc = mds_pack_md(obd, req->rq_repmsg, reply_off, body,
 731                                  inode, 1, flags,
 732                                  req->rq_export->exp_connect_flags);
 733
 734                 /* If we have LOV EA data, the OST holds size, atime, mtime */
 735                 if (!(body->valid & OBD_MD_FLEASIZE) &&
 736                     !(body->valid & OBD_MD_FLDIREA))
 737                         body->valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
 738                                         OBD_MD_FLATIME | OBD_MD_FLMTIME);
 739
 740                 reply_off++;
 741         } else if (S_ISLNK(inode->i_mode) &&
 742                    (reqbody->valid & OBD_MD_LINKNAME) != 0) {
 743                 char *symname = lustre_msg_buf(req->rq_repmsg, reply_off, 0);
 744                 int len;
 745
 746                 LASSERT (symname != NULL);       /* caller prepped reply */
 747                 len = lustre_msg_buflen(req->rq_repmsg, reply_off);
 748
 749                 rc = inode->i_op->readlink(dentry, symname, len);
 750                 if (rc < 0) {
 751                         CERROR("readlink failed: %d\n", rc);
 752                 } else if (rc != len - 1) {
 753                         CERROR ("Unexpected readlink rc %d: expecting %d\n",
 754                                 rc, len - 1);
 755                         rc = -EINVAL;
 756                 } else {
 757                         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_READLINK_EPROTO))
 758                                 rc -= 2;
 759
 760                         CDEBUG(D_INODE, "read symlink dest %s\n", symname);
 761                         body->valid |= OBD_MD_LINKNAME;
 762                         body->eadatasize = rc + 1;
 763                         symname[rc] = 0;        /* NULL terminate */
 764                         rc = 0;
 765                 }
 766                 reply_off++;
 767         } else if (reqbody->valid == OBD_MD_FLFLAGS &&
 768                    reqbody->flags & MDS_BFLAG_EXT_FLAGS) {
 769                 int flags;
 770
 771                 /* We only return the full set of flags on ioctl, otherwise we
 772                  * get enough flags from the inode in mds_pack_inode2body(). */
 773                 rc = fsfilt_iocontrol(obd, dentry, FSFILT_IOC_GETFLAGS,
 774                                       (long)&flags);
 775                 if (rc == 0)
 776                         body->flags = flags | MDS_BFLAG_EXT_FLAGS;
 777         }
 778
 779         if (reqbody->valid & OBD_MD_FLMODEASIZE) {
 780                 struct mds_obd *mds = mds_req2mds(req);
 781                 body->max_cookiesize = mds->mds_max_cookiesize;
 782                 body->max_mdsize = mds->mds_max_mdsize;
 783                 body->valid |= OBD_MD_FLMODEASIZE;
 784         }
 785
 786         if (rc)
 787                 RETURN(rc);
 788
 789 #ifdef CONFIG_FS_POSIX_ACL
 790         if ((req->rq_export->exp_connect_flags & OBD_CONNECT_ACL) &&
 791             (reqbody->valid & OBD_MD_FLACL)) {
 792                 rc = mds_pack_acl(&req->rq_export->exp_mds_data,
 793                                   inode, req->rq_repmsg,
 794                                   body, reply_off);
 795
 796                 if (body->aclsize)
 797                         reply_off++;
 798         }
 799 #endif
 800
 801         RETURN(rc);
 802 }
 803
 804 static int mds_getattr_pack_msg(struct ptlrpc_request *req, struct inode *inode,
 805                                 int offset)
 806 {
 807         struct mds_obd *mds = mds_req2mds(req);
 808         struct mds_body *body;
 809         int rc, bufcount = REPLY_REC_OFF + 1;
 810         int size[4] = { sizeof(struct ptlrpc_body),
 811                         sizeof(*body) };
 812         ENTRY;
 813
 814         LASSERT(offset == REQ_REC_OFF); /* non-intent */
 815
 816         body = lustre_msg_buf(req->rq_reqmsg, offset, sizeof(*body));
 817         LASSERT(body != NULL);                    /* checked by caller */
 818         LASSERT(lustre_req_swabbed(req, offset)); /* swabbed by caller */
 819
 820         if (body->valid & (OBD_MD_FLEASIZE | OBD_MD_FLDIREA)) {
 821                 /* this will be shrinked to actual size before size */
 822                 if (S_ISREG(inode->i_mode) || (S_ISDIR(inode->i_mode)))
 823                         size[bufcount ++] = mds->mds_max_mdsize;
 824                 else
 825                         /* we not want LSM for specfial files */
 826                         body->valid &= ~(OBD_MD_FLEASIZE | OBD_MD_FLDIREA);
 827         } else if (S_ISLNK(inode->i_mode) && (body->valid & OBD_MD_LINKNAME)) {
 828                 if (i_size_read(inode) > body->eadatasize)
 829                         CERROR("symlink size: %Lu, reply space: %d\n",
 830                                i_size_read(inode) + 1, body->eadatasize);
 831                 size[bufcount ++] = min_t(int, i_size_read(inode) + 1,
 832                                           body->eadatasize);
 833                 CDEBUG(D_INODE, "symlink size: %Lu, reply space: %d\n",
 834                        i_size_read(inode) + 1, body->eadatasize);
 835         }
 836 #ifdef CONFIG_FS_POSIX_ACL
 837         if ((req->rq_export->exp_connect_flags & OBD_CONNECT_ACL) &&
 838             (body->valid & OBD_MD_FLACL)) {
 839                 size[bufcount ++] = LUSTRE_POSIX_ACL_MAX_SIZE;
 840         }
 841 #endif
 842
 843         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) {
 844                 CERROR("failed MDS_GETATTR_PACK test\n");
 845                 req->rq_status = -ENOMEM;
 846                 RETURN(-ENOMEM);
 847         }
 848
 849         rc = lustre_pack_reply(req, bufcount, size, NULL);
 850         if (rc) {
 851                 req->rq_status = rc;
 852                 RETURN(rc);
 853         }
 854
 855         RETURN(0);
 856 }
 857
 858 static int mds_getattr_lock(struct ptlrpc_request *req, int offset,
 859                             int child_part, struct lustre_handle *child_lockh)
 860 {
 861         struct obd_device *obd = req->rq_export->exp_obd;
 862         struct mds_obd *mds = &obd->u.mds;
 863         struct ldlm_reply *rep = NULL;
 864         struct lvfs_run_ctxt saved;
 865         struct mds_body *body;
 866         struct dentry *dparent = NULL, *dchild = NULL;
 867         struct lvfs_ucred uc = {NULL,};
 868         struct lustre_handle parent_lockh;
 869         int namesize;
 870         int rc = 0, cleanup_phase = 0, resent_req = 0;
 871         int rq_offset = offset;
 872         char *name;
 873         ENTRY;
 874
 875         LASSERT(!strcmp(obd->obd_type->typ_name, LUSTRE_MDS_NAME));
 876         LASSERT(offset == REQ_REC_OFF || offset == DLM_INTENT_REC_OFF);
 877         /* if requests were at offset 2, the getattr reply goes back at 1 */
 878         if (offset == DLM_INTENT_REC_OFF) {
 879                 rep = lustre_msg_buf(req->rq_repmsg, DLM_LOCKREPLY_OFF,
 880                                      sizeof(*rep));
 881                 offset = DLM_REPLY_REC_OFF;
 882         }
 883
 884         /* Swab now, before anyone looks inside the request */
 885         body = lustre_swab_reqbuf(req, rq_offset, sizeof(*body),
 886                                   lustre_swab_mds_body);
 887         if (body == NULL) {
 888                 CERROR("Can't swab mds_body\n");
 889                 GOTO(cleanup_exit, rc = -EFAULT);
 890         }
 891
 892         lustre_set_req_swabbed(req, rq_offset + 1);
 893         name = lustre_msg_string(req->rq_reqmsg, rq_offset + 1, 0);
 894         if (name == NULL) {
 895                 CERROR("Can't unpack name\n");
 896                 GOTO(cleanup_exit, rc = -EFAULT);
 897         }
 898         namesize = lustre_msg_buflen(req->rq_reqmsg, rq_offset + 1);
 899         /* namesize less than 2 means we have empty name, probably came from
 900            revalidate by cfid, so no point in having name to be set */
 901         if (namesize <= 1)
 902                 name = NULL;
 903
 904         rc = mds_init_ucred(&uc, req, rq_offset);
 905         if (rc)
 906                 GOTO(cleanup, rc);
 907
 908
 909         push_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
 910         cleanup_phase = 1; /* kernel context */
 911         ldlm_reply_set_disposition(rep, DISP_LOOKUP_EXECD);
 912
 913         /* FIXME: handle raw lookup */
 914 #if 0
 915         if (body->valid == OBD_MD_FLID) {
 916                 struct mds_body *mds_reply;
 917                 int size = sizeof(*mds_reply);
 918                 ino_t inum;
 919                 // The user requested ONLY the inode number, so do a raw lookup
 920                 rc = lustre_pack_reply(req, 1, &size, NULL);
 921                 if (rc) {
 922                         CERROR("out of memory\n");
 923                         GOTO(cleanup, rc);
 924                 }
 925
 926                 rc = dir->i_op->lookup_raw(dir, name, namesize - 1, &inum);
 927
 928                 mds_reply = lustre_msg_buf(req->rq_repmsg, offset,
 929                                            sizeof(*mds_reply));
 930                 mds_reply->fid1.id = inum;
 931                 mds_reply->valid = OBD_MD_FLID;
 932                 GOTO(cleanup, rc);
 933         }
 934 #endif
 935
 936         /* child_lockh() is only set in fixup_handle_for_resent_req()
 937          * if MSG_RESENT is set */
 938         if (lustre_handle_is_used(child_lockh)) {
 939                 LASSERT(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT);
 940                 resent_req = 1;
 941         }
 942
 943         if (resent_req == 0) {
 944                 if (name) {
 945                         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RESEND, obd_timeout*2);
 946                         rc = mds_get_parent_child_locked(obd, &obd->u.mds,
 947                                                          &body->fid1,
 948                                                          &parent_lockh,
 949                                                          &dparent, LCK_CR,
 950                                                          MDS_INODELOCK_UPDATE,
 951                                                          name, namesize,
 952                                                          child_lockh, &dchild,
 953                                                          LCK_CR, child_part);
 954                 } else {
 955                         /* For revalidate by fid we always take UPDATE lock */
 956                         dchild = mds_fid2locked_dentry(obd, &body->fid2, NULL,
 957                                                        LCK_CR, child_lockh,
 958                                                        NULL, 0, child_part);
 959                         LASSERT(dchild);
 960                         if (IS_ERR(dchild))
 961                                 rc = PTR_ERR(dchild);
 962                 }
 963                 if (rc)
 964                         GOTO(cleanup, rc);
 965         } else {
 966                 struct ldlm_lock *granted_lock;
 967                 struct ll_fid child_fid;
 968                 struct ldlm_resource *res;
 969                 DEBUG_REQ(D_DLMTRACE, req, "resent, not enqueuing new locks");
 970                 granted_lock = ldlm_handle2lock(child_lockh);
 971                 /* lock was granted in fixup_handle_for_resent_req() if
 972                  * MSG_RESENT is set */
 973                 LASSERTF(granted_lock != NULL, LPU64"/%u lockh "LPX64"\n",
 974                          body->fid1.id, body->fid1.generation,
 975                          child_lockh->cookie);
 976
 977
 978                 res = granted_lock->l_resource;
 979                 child_fid.id = res->lr_name.name[0];
 980                 child_fid.generation = res->lr_name.name[1];
 981                 dchild = mds_fid2dentry(&obd->u.mds, &child_fid, NULL);
 982                 if (IS_ERR(dchild)) {
 983                         rc = PTR_ERR(dchild);
 984                         LCONSOLE_WARN("Child "LPU64"/%u lookup error %d.",
 985                                       child_fid.id, child_fid.generation, rc);
 986                         GOTO(cleanup, rc);
 987                 }
 988                 LDLM_LOCK_PUT(granted_lock);
 989         }
 990
 991         cleanup_phase = 2; /* dchild, dparent, locks */
 992
 993         if (dchild->d_inode == NULL) {
 994                 ldlm_reply_set_disposition(rep, DISP_LOOKUP_NEG);
 995                 /* in the intent case, the policy clears this error:
 996                    the disposition is enough */
 997                 GOTO(cleanup, rc = -ENOENT);
 998         } else {
 999                 ldlm_reply_set_disposition(rep, DISP_LOOKUP_POS);
1000         }
1001
1002         if (req->rq_repmsg == NULL) {
1003                 rc = mds_getattr_pack_msg(req, dchild->d_inode, offset);
1004                 if (rc != 0) {
1005                         CERROR ("mds_getattr_pack_msg: %d\n", rc);
1006                         GOTO (cleanup, rc);
1007                 }
1008         }
1009
1010         rc = mds_getattr_internal(obd, dchild, req, body, offset);
1011         GOTO(cleanup, rc); /* returns the lock to the client */
1012
1013  cleanup:
1014         switch (cleanup_phase) {
1015         case 2:
1016                 if (resent_req == 0) {
1017                         if (rc && dchild->d_inode)
1018                                 ldlm_lock_decref(child_lockh, LCK_CR);
1019                         if (name) {
1020                                 ldlm_lock_decref(&parent_lockh, LCK_CR);
1021                                 l_dput(dparent);
1022                         }
1023                 }
1024                 l_dput(dchild);
1025         case 1:
1026                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
1027         default:
1028                 mds_exit_ucred(&uc, mds);
1029                 if (!req->rq_packed_final) {
1030                         int rc2 = lustre_pack_reply(req, 1, NULL, NULL);
1031                         if (rc == 0)
1032                                 rc = rc2;
1033                         req->rq_status = rc;
1034                 }
1035         }
1036 cleanup_exit:
1037         return rc;
1038 }
1039
1040 static int mds_getattr(struct ptlrpc_request *req, int offset)
1041 {
1042         struct mds_obd *mds = mds_req2mds(req);
1043         struct obd_device *obd = req->rq_export->exp_obd;
1044         struct lvfs_run_ctxt saved;
1045         struct dentry *de;
1046         struct mds_body *body;
1047         struct lvfs_ucred uc = { NULL, };
1048         int rc = 0;
1049         ENTRY;
1050
1051         OBD_COUNTER_INCREMENT(obd, getattr);
1052
1053         body = lustre_swab_reqbuf(req, offset, sizeof(*body),
1054                                   lustre_swab_mds_body);
1055         if (body == NULL)
1056                 GOTO(cleanup_exit, rc = -EFAULT);
1057
1058         rc = mds_init_ucred(&uc, req, offset);
1059         if (rc)
1060                 GOTO(out_ucred, rc);
1061
1062         push_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
1063         de = mds_fid2dentry(mds, &body->fid1, NULL);
1064         if (IS_ERR(de)) {
1065                 req->rq_status = PTR_ERR(de);
1066                 GOTO(out_pop, rc);
1067         }
1068
1069         rc = mds_getattr_pack_msg(req, de->d_inode, offset);
1070         if (rc != 0) {
1071                 CERROR("mds_getattr_pack_msg: %d\n", rc);
1072                 GOTO(out_dput, rc);
1073         }
1074
1075         req->rq_status = mds_getattr_internal(obd, de, req, body,REPLY_REC_OFF);
1076 out_dput:
1077         l_dput(de);
1078         GOTO(out_pop, rc);
1079 out_pop:
1080         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
1081 out_ucred:
1082         if (!req->rq_packed_final) {
1083                 int rc2 = lustre_pack_reply(req, 1, NULL, NULL);
1084                 if (rc == 0)
1085                         rc = rc2;
1086                 if (rc != 0)
1087                         req->rq_status = rc;
1088         } else {
1089                 mds_shrink_body_reply(req, offset, REPLY_REC_OFF);
1090         }
1091         mds_exit_ucred(&uc, mds);
1092
1093 cleanup_exit:
1094         return rc;
1095 }
1096
1097 static int mds_obd_statfs(struct obd_device *obd, struct obd_statfs *osfs,
1098                           __u64 max_age, __u32 flags)
1099 {
1100         int rc;
1101
1102         spin_lock(&obd->obd_osfs_lock);
1103         rc = fsfilt_statfs(obd, obd->u.obt.obt_sb, max_age);
1104         if (rc == 0)
1105                 memcpy(osfs, &obd->obd_osfs, sizeof(*osfs));
1106         spin_unlock(&obd->obd_osfs_lock);
1107
1108         return rc;
1109 }
1110
1111 static int mds_statfs(struct ptlrpc_request *req)
1112 {
1113         struct obd_device *obd = req->rq_export->exp_obd;
1114         struct ptlrpc_service *svc = req->rq_rqbd->rqbd_service;
1115         int rc, size[2] = { sizeof(struct ptlrpc_body),
1116                             sizeof(struct obd_statfs) };
1117         ENTRY;
1118
1119         /* This will trigger a watchdog timeout */
1120         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_STATFS_LCW_SLEEP,
1121                          (MDS_SERVICE_WATCHDOG_FACTOR *
1122                           at_get(&svc->srv_at_estimate)) + 1);
1123         OBD_COUNTER_INCREMENT(obd, statfs);
1124
1125         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_STATFS_PACK))
1126                 GOTO(out, rc = -ENOMEM);
1127         rc = lustre_pack_reply(req, 2, size, NULL);
1128         if (rc)
1129                 GOTO(out, rc);
1130
1131         /* We call this so that we can cache a bit - 1 jiffie worth */
1132         rc = mds_obd_statfs(obd, lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
1133                                                 size[REPLY_REC_OFF]),
1134                             cfs_time_shift_64(-OBD_STATFS_CACHE_SECONDS), 0);
1135         if (rc) {
1136                 CERROR("mds_obd_statfs failed: rc %d\n", rc);
1137                 GOTO(out, rc);
1138         }
1139
1140         EXIT;
1141 out:
1142         req->rq_status = rc;
1143         return 0;
1144 }
1145
1146 static int mds_sync(struct ptlrpc_request *req, int offset)
1147 {
1148         struct obd_device *obd = req->rq_export->exp_obd;
1149         struct mds_obd *mds = &obd->u.mds;
1150         struct mds_body *body;
1151         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
1152         ENTRY;
1153
1154         body = lustre_swab_reqbuf(req, offset, sizeof(*body),
1155                                   lustre_swab_mds_body);
1156         if (body == NULL)
1157                 GOTO(out, rc = -EFAULT);
1158
1159         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SYNC_PACK))
1160                 GOTO(out, rc = -ENOMEM);
1161         rc = lustre_pack_reply(req, 2, size, NULL);
1162         if (rc)
1163                 GOTO(out, rc);
1164
1165         rc = fsfilt_sync(obd, obd->u.obt.obt_sb);
1166         if (rc == 0 && body->fid1.id != 0) {
1167                 struct dentry *de;
1168
1169                 de = mds_fid2dentry(mds, &body->fid1, NULL);
1170                 if (IS_ERR(de))
1171                         GOTO(out, rc = PTR_ERR(de));
1172
1173                 body = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
1174                                       sizeof(*body));
1175                 mds_pack_inode2body(body, de->d_inode);
1176
1177                 l_dput(de);
1178         }
1179         GOTO(out, rc);
1180 out:
1181         req->rq_status = rc;
1182         return 0;
1183 }
1184
1185 /* mds_readpage does not take a DLM lock on the inode, because the client must
1186  * already have a PR lock.
1187  *
1188  * If we were to take another one here, a deadlock will result, if another
1189  * thread is already waiting for a PW lock. */
1190 static int mds_readpage(struct ptlrpc_request *req, int offset)
1191 {
1192         struct obd_device *obd = req->rq_export->exp_obd;
1193         struct mds_obd *mds = &obd->u.mds;
1194         struct vfsmount *mnt;
1195         struct dentry *de;
1196         struct file *file;
1197         struct mds_body *body, *repbody;
1198         struct lvfs_run_ctxt saved;
1199         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*repbody) };
1200         struct lvfs_ucred uc = {NULL,};
1201         ENTRY;
1202
1203         OBD_FAIL_RETURN(OBD_FAIL_MDS_READPAGE_PACK, -ENOMEM);
1204         rc = lustre_pack_reply(req, 2, size, NULL);
1205         if (rc)
1206                 GOTO(out, rc);
1207
1208         body = lustre_swab_reqbuf(req, offset, sizeof(*body),
1209                                   lustre_swab_mds_body);
1210         if (body == NULL)
1211                 GOTO (out, rc = -EFAULT);
1212
1213         rc = mds_init_ucred(&uc, req, offset);
1214         if (rc)
1215                 GOTO(out, rc);
1216
1217         push_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
1218         de = mds_fid2dentry(&obd->u.mds, &body->fid1, &mnt);
1219         if (IS_ERR(de))
1220                 GOTO(out_pop, rc = PTR_ERR(de));
1221
1222         CDEBUG(D_INODE, "ino %lu\n", de->d_inode->i_ino);
1223
1224         file = ll_dentry_open(de, mnt, O_RDONLY | O_LARGEFILE, current_cred());
1225         /* note: in case of an error, dentry_open puts dentry */
1226         if (IS_ERR(file))
1227                 GOTO(out_pop, rc = PTR_ERR(file));
1228
1229         /* body->size is actually the offset -eeb */
1230         if ((body->size & (de->d_inode->i_sb->s_blocksize - 1)) != 0) {
1231                 CERROR("offset "LPU64" not on a block boundary of %lu\n",
1232                        body->size, de->d_inode->i_sb->s_blocksize);
1233                 GOTO(out_file, rc = -EFAULT);
1234         }
1235
1236         /* body->nlink is actually the #bytes to read -eeb */
1237         if (body->nlink & (de->d_inode->i_sb->s_blocksize - 1)) {
1238                 CERROR("size %u is not multiple of blocksize %lu\n",
1239                        body->nlink, de->d_inode->i_sb->s_blocksize);
1240                 GOTO(out_file, rc = -EFAULT);
1241         }
1242
1243         repbody = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
1244                                  sizeof(*repbody));
1245         repbody->size = i_size_read(file->f_dentry->d_inode);
1246         repbody->valid = OBD_MD_FLSIZE;
1247
1248         /* to make this asynchronous make sure that the handling function
1249            doesn't send a reply when this function completes. Instead a
1250            callback function would send the reply */
1251         /* body->size is actually the offset -eeb */
1252         rc = mds_sendpage(req, file, body->size, body->nlink);
1253
1254 out_file:
1255         filp_close(file, 0);
1256 out_pop:
1257         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
1258 out:
1259         mds_exit_ucred(&uc, mds);
1260         req->rq_status = rc;
1261         RETURN(0);
1262 }
1263
1264 int mds_reint(struct ptlrpc_request *req, int offset,
1265               struct lustre_handle *lockh)
1266 {
1267         struct mds_update_record *rec; /* 116 bytes on the stack?  no sir! */
1268         int rc;
1269
1270         OBD_ALLOC(rec, sizeof(*rec));
1271         if (rec == NULL)
1272                 RETURN(-ENOMEM);
1273
1274         rc = mds_update_unpack(req, offset, rec);
1275         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNPACK)) {
1276                 CERROR("invalid record\n");
1277                 GOTO(out, req->rq_status = -EINVAL);
1278         }
1279
1280         mds_root_squash(&req->rq_export->exp_obd->u.mds, &req->rq_peer.nid,
1281                         &rec->ur_uc.luc_fsuid, &rec->ur_uc.luc_fsgid,
1282                         &rec->ur_uc.luc_cap, &rec->ur_uc.luc_suppgid1,
1283                         &rec->ur_uc.luc_suppgid2);
1284
1285         /* rc will be used to interrupt a for loop over multiple records */
1286         rc = mds_reint_rec(rec, offset, req, lockh);
1287  out:
1288         OBD_FREE(rec, sizeof(*rec));
1289         return rc;
1290 }
1291
1292 static int mds_filter_recovery_request(struct ptlrpc_request *req,
1293                                        struct obd_device *obd, int *process)
1294 {
1295         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1296         case MDS_CONNECT: /* This will never get here, but for completeness. */
1297         case OST_CONNECT: /* This will never get here, but for completeness. */
1298         case MDS_DISCONNECT:
1299         case OST_DISCONNECT:
1300                *process = 1;
1301                RETURN(0);
1302
1303         case MDS_CLOSE:
1304         case MDS_SYNC: /* used in unmounting */
1305         case OBD_PING:
1306         case MDS_SETXATTR:
1307         case MDS_REINT:
1308         case LDLM_ENQUEUE:
1309                 *process = target_queue_recovery_request(req, obd);
1310                 RETURN(0);
1311
1312         default:
1313                 DEBUG_REQ(D_ERROR, req, "not permitted during recovery");
1314                 *process = 0;
1315                 /* XXX what should we set rq_status to here? */
1316                 req->rq_status = -EAGAIN;
1317                 RETURN(ptlrpc_error(req));
1318         }
1319 }
1320
1321 static char *reint_names[] = {
1322         [REINT_SETATTR] "setattr",
1323         [REINT_CREATE]  "create",
1324         [REINT_LINK]    "link",
1325         [REINT_UNLINK]  "unlink",
1326         [REINT_RENAME]  "rename",
1327         [REINT_OPEN]    "open",
1328 };
1329
1330 static int mds_set_info_rpc(struct obd_export *exp, struct ptlrpc_request *req)
1331 {
1332         void *key, *val;
1333         int keylen, vallen, rc = 0;
1334         ENTRY;
1335
1336         key = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF, 1);
1337         if (key == NULL) {
1338                 DEBUG_REQ(D_HA, req, "no set_info key");
1339                 RETURN(-EFAULT);
1340         }
1341         keylen = lustre_msg_buflen(req->rq_reqmsg, REQ_REC_OFF);
1342
1343         val = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 1, 0);
1344         vallen = lustre_msg_buflen(req->rq_reqmsg, REQ_REC_OFF + 1);
1345
1346         rc = lustre_pack_reply(req, 1, NULL, NULL);
1347         if (rc)
1348                 RETURN(rc);
1349
1350         lustre_msg_set_status(req->rq_repmsg, 0);
1351
1352         /* Accept the broken "read-only" key from 1.6.6 clients. b=17493 */
1353         if (KEY_IS(KEY_READONLY) || KEY_IS(KEY_READONLY_166COMPAT)) {
1354                 if (val == NULL || vallen < sizeof(__u32)) {
1355                         DEBUG_REQ(D_HA, req, "no set_info val");
1356                         RETURN(-EFAULT);
1357                 }
1358
1359                 if (*(__u32 *)val)
1360                         exp->exp_connect_flags |= OBD_CONNECT_RDONLY;
1361                 else
1362                         exp->exp_connect_flags &= ~OBD_CONNECT_RDONLY;
1363         } else {
1364                 RETURN(-EINVAL);
1365         }
1366
1367         RETURN(0);
1368 }
1369
1370 #ifdef HAVE_QUOTA_SUPPORT
1371 static int mds_handle_quotacheck(struct ptlrpc_request *req)
1372 {
1373         struct obd_quotactl *oqctl;
1374         int rc;
1375         ENTRY;
1376
1377         oqctl = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*oqctl),
1378                                    lustre_swab_obd_quotactl);
1379         if (oqctl == NULL)
1380                 RETURN(-EPROTO);
1381
1382         rc = lustre_pack_reply(req, 1, NULL, NULL);
1383         if (rc)
1384                 RETURN(rc);
1385
1386         req->rq_status = obd_quotacheck(req->rq_export, oqctl);
1387         RETURN(0);
1388 }
1389
1390 static int mds_handle_quotactl(struct ptlrpc_request *req)
1391 {
1392         struct obd_quotactl *oqctl, *repoqc;
1393         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*repoqc) };
1394         ENTRY;
1395
1396         oqctl = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*oqctl),
1397                                    lustre_swab_obd_quotactl);
1398         if (oqctl == NULL)
1399                 RETURN(-EPROTO);
1400
1401         rc = lustre_pack_reply(req, 2, size, NULL);
1402         if (rc)
1403                 RETURN(rc);
1404
1405         repoqc = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof(*repoqc));
1406
1407         req->rq_status = obd_quotactl(req->rq_export, oqctl);
1408         *repoqc = *oqctl;
1409         RETURN(0);
1410 }
1411 #endif
1412
1413 static int mds_msg_check_version(struct lustre_msg *msg)
1414 {
1415         int rc;
1416
1417         switch (lustre_msg_get_opc(msg)) {
1418         case MDS_CONNECT:
1419         case MDS_DISCONNECT:
1420         case OBD_PING:
1421                 rc = lustre_msg_check_version(msg, LUSTRE_OBD_VERSION);
1422                 if (rc)
1423                         CERROR("bad opc %u version %08x, expecting %08x\n",
1424                                lustre_msg_get_opc(msg),
1425                                lustre_msg_get_version(msg),
1426                                LUSTRE_OBD_VERSION);
1427                 break;
1428         case MDS_GETSTATUS:
1429         case MDS_GETATTR:
1430         case MDS_GETATTR_NAME:
1431         case MDS_STATFS:
1432         case MDS_READPAGE:
1433         case MDS_REINT:
1434         case MDS_CLOSE:
1435         case MDS_DONE_WRITING:
1436         case MDS_PIN:
1437         case MDS_SYNC:
1438         case MDS_GETXATTR:
1439         case MDS_SETXATTR:
1440         case MDS_SET_INFO:
1441         case MDS_QUOTACHECK:
1442         case MDS_QUOTACTL:
1443         case QUOTA_DQACQ:
1444         case QUOTA_DQREL:
1445                 rc = lustre_msg_check_version(msg, LUSTRE_MDS_VERSION);
1446                 if (rc)
1447                         CERROR("bad opc %u version %08x, expecting %08x\n",
1448                                lustre_msg_get_opc(msg),
1449                                lustre_msg_get_version(msg),
1450                                LUSTRE_MDS_VERSION);
1451                 break;
1452         case LDLM_ENQUEUE:
1453         case LDLM_CONVERT:
1454         case LDLM_BL_CALLBACK:
1455         case LDLM_CP_CALLBACK:
1456                 rc = lustre_msg_check_version(msg, LUSTRE_DLM_VERSION);
1457                 if (rc)
1458                         CERROR("bad opc %u version %08x, expecting %08x\n",
1459                                lustre_msg_get_opc(msg),
1460                                lustre_msg_get_version(msg),
1461                                LUSTRE_DLM_VERSION);
1462                 break;
1463         case OBD_LOG_CANCEL:
1464         case LLOG_ORIGIN_HANDLE_CREATE:
1465         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
1466         case LLOG_ORIGIN_HANDLE_READ_HEADER:
1467         case LLOG_ORIGIN_HANDLE_CLOSE:
1468         case LLOG_ORIGIN_HANDLE_DESTROY:
1469         case LLOG_ORIGIN_HANDLE_PREV_BLOCK:
1470         case LLOG_CATINFO:
1471                 rc = lustre_msg_check_version(msg, LUSTRE_LOG_VERSION);
1472                 if (rc)
1473                         CERROR("bad opc %u version %08x, expecting %08x\n",
1474                                lustre_msg_get_opc(msg),
1475                                lustre_msg_get_version(msg),
1476                                LUSTRE_LOG_VERSION);
1477                 break;
1478         default:
1479                 CERROR("MDS unknown opcode %d\n", lustre_msg_get_opc(msg));
1480                 rc = -ENOTSUPP;
1481         }
1482         return rc;
1483 }
1484
1485 int mds_handle(struct ptlrpc_request *req)
1486 {
1487         int should_process, fail = OBD_FAIL_MDS_ALL_REPLY_NET;
1488         int rc = 0;
1489         struct mds_obd *mds = NULL; /* quell gcc overwarning */
1490         struct obd_device *obd = NULL;
1491         ENTRY;
1492
1493         OBD_FAIL_RETURN(OBD_FAIL_MDS_ALL_REQUEST_NET | OBD_FAIL_ONCE, 0);
1494
1495         LASSERT(current->journal_info == NULL);
1496
1497         rc = mds_msg_check_version(req->rq_reqmsg);
1498         if (rc) {
1499                 CERROR("MDS drop mal-formed request\n");
1500                 RETURN(rc);
1501         }
1502
1503         if (lustre_msg_get_opc(req->rq_reqmsg) != MDS_CONNECT) {
1504                 struct mds_export_data *med;
1505                 int recovering;
1506
1507                 if (!class_connected_export(req->rq_export)) {
1508                         CERROR("operation %d on unconnected MDS from %s\n",
1509                                lustre_msg_get_opc(req->rq_reqmsg),
1510                                libcfs_id2str(req->rq_peer));
1511                         req->rq_status = -ENOTCONN;
1512                         GOTO(out, rc = -ENOTCONN);
1513                 }
1514
1515                 med = &req->rq_export->exp_mds_data;
1516                 obd = req->rq_export->exp_obd;
1517                 mds = &obd->u.mds;
1518
1519                 /* sanity check: if the xid matches, the request must
1520                  * be marked as a resent or replayed */
1521                 if (req->rq_xid == le64_to_cpu(med->med_lcd->lcd_last_xid) ||
1522                     req->rq_xid == le64_to_cpu(med->med_lcd->lcd_last_close_xid))
1523                         if (!(lustre_msg_get_flags(req->rq_reqmsg) &
1524                                  (MSG_RESENT | MSG_REPLAY))) {
1525                                 CERROR("rq_xid "LPU64" matches last_xid, "
1526                                        "expected RESENT flag\n",
1527                                         req->rq_xid);
1528                                 req->rq_status = -ENOTCONN;
1529                                 GOTO(out, rc = -EFAULT);
1530                         }
1531                 /* else: note the opposite is not always true; a
1532                  * RESENT req after a failover will usually not match
1533                  * the last_xid, since it was likely never
1534                  * committed. A REPLAYed request will almost never
1535                  * match the last xid, however it could for a
1536                  * committed, but still retained, open. */
1537
1538                 /* Check for aborted recovery. */
1539                 spin_lock_bh(&obd->obd_processing_task_lock);
1540                 recovering = obd->obd_recovering;
1541                 spin_unlock_bh(&obd->obd_processing_task_lock);
1542                 if (recovering &&
1543                     target_recovery_check_and_stop(obd) == 0) {
1544                         rc = mds_filter_recovery_request(req, obd,
1545                                                          &should_process);
1546                         if (rc || !should_process)
1547                                 RETURN(rc);
1548                 }
1549         }
1550
1551         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1552         case MDS_CONNECT:
1553                 DEBUG_REQ(D_INODE, req, "connect");
1554                 OBD_FAIL_RETURN(OBD_FAIL_MDS_CONNECT_NET, 0);
1555                 rc = target_handle_connect(req, mds_handle);
1556                 if (!rc) {
1557                         /* Now that we have an export, set obd. */
1558                         obd = req->rq_export->exp_obd;
1559                 }
1560                 break;
1561
1562         case MDS_DISCONNECT:
1563                 DEBUG_REQ(D_INODE, req, "disconnect");
1564                 OBD_FAIL_RETURN(OBD_FAIL_MDS_DISCONNECT_NET, 0);
1565                 rc = target_handle_disconnect(req);
1566                 req->rq_status = rc;            /* superfluous? */
1567                 break;
1568
1569         case MDS_GETSTATUS:
1570                 DEBUG_REQ(D_INODE, req, "getstatus");
1571                 OBD_FAIL_RETURN(OBD_FAIL_MDS_GETSTATUS_NET, 0);
1572                 rc = mds_getstatus(req);
1573                 break;
1574
1575         case MDS_GETATTR:
1576                 DEBUG_REQ(D_INODE, req, "getattr");
1577                 OBD_FAIL_RETURN(OBD_FAIL_MDS_GETATTR_NET, 0);
1578                 rc = mds_getattr(req, REQ_REC_OFF);
1579                 break;
1580
1581         case MDS_SETXATTR:
1582                 DEBUG_REQ(D_INODE, req, "setxattr");
1583                 OBD_FAIL_RETURN(OBD_FAIL_MDS_SETXATTR_NET, 0);
1584                 rc = mds_setxattr(req);
1585                 break;
1586
1587         case MDS_GETXATTR:
1588                 DEBUG_REQ(D_INODE, req, "getxattr");
1589                 OBD_FAIL_RETURN(OBD_FAIL_MDS_GETXATTR_NET, 0);
1590                 rc = mds_getxattr(req);
1591                 break;
1592
1593         case MDS_GETATTR_NAME: {
1594                 struct lustre_handle lockh = { 0 };
1595                 DEBUG_REQ(D_INODE, req, "getattr_name");
1596                 OBD_FAIL_RETURN(OBD_FAIL_MDS_GETATTR_NAME_NET, 0);
1597
1598                 /* If this request gets a reconstructed reply, we won't be
1599                  * acquiring any new locks in mds_getattr_lock, so we don't
1600                  * want to cancel.
1601                  */
1602                 rc = mds_getattr_lock(req, REQ_REC_OFF, MDS_INODELOCK_UPDATE,
1603                                       &lockh);
1604                 mds_shrink_body_reply(req, REQ_REC_OFF, REPLY_REC_OFF);
1605                 /* this non-intent call (from an ioctl) is special */
1606                 req->rq_status = rc;
1607                 if (rc == 0 && lustre_handle_is_used(&lockh))
1608                         ldlm_lock_decref(&lockh, LCK_CR);
1609                 break;
1610         }
1611         case MDS_STATFS:
1612                 DEBUG_REQ(D_INODE, req, "statfs");
1613                 OBD_FAIL_RETURN(OBD_FAIL_MDS_STATFS_NET, 0);
1614                 rc = mds_statfs(req);
1615                 break;
1616
1617         case MDS_READPAGE:
1618                 DEBUG_REQ(D_INODE, req, "readpage");
1619                 OBD_FAIL_RETURN(OBD_FAIL_MDS_READPAGE_NET, 0);
1620                 rc = mds_readpage(req, REQ_REC_OFF);
1621
1622                 if (OBD_FAIL_CHECK_ONCE(OBD_FAIL_MDS_SENDPAGE)) {
1623                         RETURN(0);
1624                 }
1625
1626                 break;
1627
1628         case MDS_REINT: {
1629                 __u32 *opcp = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF,
1630                                              sizeof(*opcp));
1631                 __u32  opc;
1632                 int op = 0;
1633                 int size[4] = { sizeof(struct ptlrpc_body),
1634                                sizeof(struct mds_body),
1635                                mds->mds_max_mdsize,
1636                                mds->mds_max_cookiesize };
1637                 int bufcount;
1638
1639                 /* NB only peek inside req now; mds_reint() will swab it */
1640                 if (opcp == NULL) {
1641                         CERROR ("Can't inspect opcode\n");
1642                         rc = -EINVAL;
1643                         break;
1644                 }
1645                 opc = *opcp;
1646                 if (lustre_req_need_swab(req))
1647                         __swab32s(&opc);
1648
1649                 DEBUG_REQ(D_INODE, req, "reint %d (%s)", opc,
1650                           (opc < REINT_MAX) ? reint_names[opc] :
1651                           "unknown opcode");
1652
1653                 switch (opc) {
1654                 case REINT_CREATE:
1655                         op = PTLRPC_LAST_CNTR + MDS_REINT_CREATE;
1656                         break;
1657                 case REINT_LINK:
1658                         op = PTLRPC_LAST_CNTR + MDS_REINT_LINK;
1659                         break;
1660                 case REINT_OPEN:
1661                         op = PTLRPC_LAST_CNTR + MDS_REINT_OPEN;
1662                         break;
1663                 case REINT_SETATTR:
1664                         op = PTLRPC_LAST_CNTR + MDS_REINT_SETATTR;
1665                         break;
1666                 case REINT_RENAME:
1667                         op = PTLRPC_LAST_CNTR + MDS_REINT_RENAME;
1668                         break;
1669                 case REINT_UNLINK:
1670                         op = PTLRPC_LAST_CNTR + MDS_REINT_UNLINK;
1671                         break;
1672                 default:
1673                         op = 0;
1674                         break;
1675                 }
1676
1677                 if (op && req->rq_rqbd->rqbd_service->srv_stats)
1678                         lprocfs_counter_incr(
1679                                 req->rq_rqbd->rqbd_service->srv_stats, op);
1680
1681                 OBD_FAIL_RETURN(OBD_FAIL_MDS_REINT_NET, 0);
1682
1683                 if (opc == REINT_UNLINK || opc == REINT_RENAME)
1684                         bufcount = 4;
1685                 else if (opc == REINT_OPEN)
1686                         bufcount = 3;
1687                 else
1688                         bufcount = 2;
1689
1690                 /* if we do recovery we isn't send reply mds state is restored */
1691                 if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) {
1692                         size[DLM_REPLY_REC_OFF] = 0;
1693                         if (opc == REINT_UNLINK || opc == REINT_RENAME)
1694                                 size[DLM_REPLY_REC_OFF + 1] = 0;
1695                 }
1696
1697                 rc = lustre_pack_reply(req, bufcount, size, NULL);
1698                 if (rc)
1699                         break;
1700
1701                 rc = mds_reint(req, REQ_REC_OFF, NULL);
1702                 mds_shrink_intent_reply(req, opc, REPLY_REC_OFF);
1703                 fail = OBD_FAIL_MDS_REINT_NET_REP;
1704                 break;
1705         }
1706
1707         case MDS_CLOSE:
1708                 DEBUG_REQ(D_INODE, req, "close");
1709                 OBD_FAIL_RETURN(OBD_FAIL_MDS_CLOSE_NET, 0);
1710                 rc = mds_close(req, REQ_REC_OFF);
1711                 mds_shrink_body_reply(req, REQ_REC_OFF, REPLY_REC_OFF);
1712                 fail = OBD_FAIL_MDS_CLOSE_NET_REP;
1713                 break;
1714
1715         case MDS_DONE_WRITING:
1716                 DEBUG_REQ(D_INODE, req, "done_writing");
1717                 OBD_FAIL_RETURN(OBD_FAIL_MDS_DONE_WRITING_NET, 0);
1718                 rc = mds_done_writing(req, REQ_REC_OFF);
1719                 break;
1720
1721         case MDS_PIN:
1722                 DEBUG_REQ(D_INODE, req, "pin");
1723                 OBD_FAIL_RETURN(OBD_FAIL_MDS_PIN_NET, 0);
1724                 rc = mds_pin(req, REQ_REC_OFF);
1725                 break;
1726
1727         case MDS_SYNC:
1728                 DEBUG_REQ(D_INODE, req, "sync");
1729                 OBD_FAIL_RETURN(OBD_FAIL_MDS_SYNC_NET, 0);
1730                 rc = mds_sync(req, REQ_REC_OFF);
1731                 break;
1732
1733         case MDS_SET_INFO:
1734                 DEBUG_REQ(D_INODE, req, "set_info");
1735                 rc = mds_set_info_rpc(req->rq_export, req);
1736                 break;
1737 #ifdef HAVE_QUOTA_SUPPORT
1738         case MDS_QUOTACHECK:
1739                 DEBUG_REQ(D_INODE, req, "quotacheck");
1740                 OBD_FAIL_RETURN(OBD_FAIL_MDS_QUOTACHECK_NET, 0);
1741                 rc = mds_handle_quotacheck(req);
1742                 break;
1743
1744         case MDS_QUOTACTL:
1745                 DEBUG_REQ(D_INODE, req, "quotactl");
1746                 OBD_FAIL_RETURN(OBD_FAIL_MDS_QUOTACTL_NET, 0);
1747                 rc = mds_handle_quotactl(req);
1748                 break;
1749 #endif
1750         case OBD_PING:
1751                 DEBUG_REQ(D_INODE, req, "ping");
1752                 rc = target_handle_ping(req);
1753                 if (req->rq_export->exp_delayed)
1754                         mds_update_client_epoch(req->rq_export);
1755                 break;
1756
1757         case OBD_LOG_CANCEL:
1758                 CDEBUG(D_INODE, "log cancel\n");
1759                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOG_CANCEL_NET, 0);
1760                 rc = -ENOTSUPP; /* la la la */
1761                 break;
1762
1763         case LDLM_ENQUEUE:
1764                 DEBUG_REQ(D_INODE, req, "enqueue");
1765                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_ENQUEUE, 0);
1766                 rc = ldlm_handle_enqueue(req, ldlm_server_completion_ast,
1767                                          ldlm_server_blocking_ast, NULL);
1768                 fail = OBD_FAIL_LDLM_REPLY;
1769                 break;
1770         case LDLM_CONVERT:
1771                 DEBUG_REQ(D_INODE, req, "convert");
1772                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CONVERT, 0);
1773                 rc = ldlm_handle_convert(req);
1774                 break;
1775         case LDLM_BL_CALLBACK:
1776         case LDLM_CP_CALLBACK:
1777                 DEBUG_REQ(D_INODE, req, "callback");
1778                 CERROR("callbacks should not happen on MDS\n");
1779                 LBUG();
1780                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_BL_CALLBACK, 0);
1781                 break;
1782         case LLOG_ORIGIN_HANDLE_CREATE:
1783                 DEBUG_REQ(D_INODE, req, "llog_init");
1784                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1785                 rc = llog_origin_handle_create(req);
1786                 break;
1787         case LLOG_ORIGIN_HANDLE_DESTROY:
1788                 DEBUG_REQ(D_INODE, req, "llog_init");
1789                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1790                 rc = llog_origin_handle_destroy(req);
1791                 break;
1792         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
1793                 DEBUG_REQ(D_INODE, req, "llog next block");
1794                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1795                 rc = llog_origin_handle_next_block(req);
1796                 break;
1797         case LLOG_ORIGIN_HANDLE_PREV_BLOCK:
1798                 DEBUG_REQ(D_INODE, req, "llog prev block");
1799                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1800                 rc = llog_origin_handle_prev_block(req);
1801                 break;
1802         case LLOG_ORIGIN_HANDLE_READ_HEADER:
1803                 DEBUG_REQ(D_INODE, req, "llog read header");
1804                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1805                 rc = llog_origin_handle_read_header(req);
1806                 break;
1807         case LLOG_ORIGIN_HANDLE_CLOSE:
1808                 DEBUG_REQ(D_INODE, req, "llog close");
1809                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1810                 rc = llog_origin_handle_close(req);
1811                 break;
1812         case LLOG_CATINFO:
1813                 DEBUG_REQ(D_INODE, req, "llog catinfo");
1814                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1815                 rc = llog_catinfo(req);
1816                 break;
1817         default:
1818                 req->rq_status = -ENOTSUPP;
1819                 rc = ptlrpc_error(req);
1820                 RETURN(rc);
1821         }
1822
1823         LASSERT(current->journal_info == NULL);
1824
1825         /* If we're DISCONNECTing, the mds_export_data is already freed */
1826         if (!rc && lustre_msg_get_opc(req->rq_reqmsg) != MDS_DISCONNECT) {
1827                 struct mds_export_data *med = &req->rq_export->exp_mds_data;
1828
1829                 /* I don't think last_xid is used for anyway, so I'm not sure
1830                    if we need to care about last_close_xid here.*/
1831                 lustre_msg_set_last_xid(req->rq_repmsg,
1832                                         le64_to_cpu(med->med_lcd->lcd_last_xid));
1833                 target_committed_to_req(req);
1834         }
1835
1836         EXIT;
1837 out:
1838         return target_handle_reply(req, rc, fail);
1839 }
1840
1841 /* Update the server data on disk.  This stores the new mount_count and
1842  * also the last_rcvd value to disk.  If we don't have a clean shutdown,
1843  * then the server last_rcvd value may be less than that of the clients.
1844  * This will alert us that we may need to do client recovery.
1845  *
1846  * Also assumes for mds_last_transno that we are not modifying it (no locking).
1847  */
1848 int mds_update_server_data(struct obd_device *obd, int force_sync)
1849 {
1850         struct mds_obd *mds = &obd->u.mds;
1851         struct lr_server_data *lsd = mds->mds_server_data;
1852         struct file *filp = mds->mds_rcvd_filp;
1853         struct lvfs_run_ctxt saved;
1854         loff_t off = 0;
1855         int rc;
1856         ENTRY;
1857
1858         CDEBUG(D_SUPER, "MDS mount_count is "LPU64", last_transno is "LPU64"\n",
1859                mds->mds_mount_count, mds->mds_last_transno);
1860
1861         spin_lock(&mds->mds_transno_lock);
1862         lsd->lsd_last_transno = cpu_to_le64(mds->mds_last_transno);
1863         spin_unlock(&mds->mds_transno_lock);
1864
1865         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1866         rc = fsfilt_write_record(obd, filp, lsd, sizeof(*lsd), &off,force_sync);
1867         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1868         if (rc)
1869                 CERROR("error writing MDS server data: rc = %d\n", rc);
1870
1871         RETURN(rc);
1872 }
1873
1874 static void fsoptions_to_mds_flags(struct mds_obd *mds, char *options)
1875 {
1876         char *p = options;
1877
1878         if (!options)
1879                 return;
1880
1881         while (*options) {
1882                 int len;
1883
1884                 while (*p && *p != ',')
1885                         p++;
1886
1887                 len = p - options;
1888                 if (len == sizeof("user_xattr") - 1 &&
1889                     memcmp(options, "user_xattr", len) == 0) {
1890                         mds->mds_fl_user_xattr = 1;
1891                         LCONSOLE_INFO("Enabling user_xattr\n");
1892                 } else if (len == sizeof("nouser_xattr") - 1 &&
1893                            memcmp(options, "nouser_xattr", len) == 0) {
1894                         mds->mds_fl_user_xattr = 0;
1895                         LCONSOLE_INFO("Disabling user_xattr\n");
1896                 } else if (len == sizeof("acl") - 1 &&
1897                            memcmp(options, "acl", len) == 0) {
1898 #ifdef CONFIG_FS_POSIX_ACL
1899                         mds->mds_fl_acl = 1;
1900                         LCONSOLE_INFO("Enabling ACL\n");
1901 #else
1902                         CWARN("ignoring unsupported acl mount option\n");
1903 #endif
1904                 } else if (len == sizeof("noacl") - 1 &&
1905                            memcmp(options, "noacl", len) == 0) {
1906 #ifdef CONFIG_FS_POSIX_ACL
1907                         mds->mds_fl_acl = 0;
1908                         LCONSOLE_INFO("Disabling ACL\n");
1909 #endif
1910                 }
1911
1912                 options = ++p;
1913         }
1914 }
1915
1916 /* mount the file system (secretly).  lustre_cfg parameters are:
1917  * 1 = device
1918  * 2 = fstype
1919  * 3 = config name
1920  * 4 = mount options
1921  */
1922 static int mds_setup(struct obd_device *obd, obd_count len, void *buf)
1923 {
1924         struct lprocfs_static_vars lvars;
1925         struct lustre_cfg* lcfg = buf;
1926         struct mds_obd *mds = &obd->u.mds;
1927         struct lustre_sb_info *lsi;
1928         struct lustre_mount_info *lmi;
1929         struct vfsmount *mnt;
1930         struct obd_uuid uuid;
1931         __u8 *uuid_ptr;
1932         char *str, *label;
1933         char ns_name[48];
1934         int rc = 0;
1935         ENTRY;
1936
1937         /* setup 1:/dev/loop/0 2:ext3 3:mdsA 4:errors=remount-ro,iopen_nopriv */
1938
1939         CLASSERT(offsetof(struct obd_device, u.obt) ==
1940                  offsetof(struct obd_device, u.mds.mds_obt));
1941
1942         if (lcfg->lcfg_bufcount < 3)
1943                 RETURN(-EINVAL);
1944
1945         if (LUSTRE_CFG_BUFLEN(lcfg, 1) == 0 || LUSTRE_CFG_BUFLEN(lcfg, 2) == 0)
1946                 RETURN(-EINVAL);
1947
1948         lmi = server_get_mount(obd->obd_name);
1949         if (!lmi) {
1950                 CERROR("Not mounted in lustre_fill_super?\n");
1951                 RETURN(-EINVAL);
1952         }
1953
1954         /* We mounted in lustre_fill_super.
1955            lcfg bufs 1, 2, 4 (device, fstype, mount opts) are ignored.*/
1956         lsi = s2lsi(lmi->lmi_sb);
1957         fsoptions_to_mds_flags(mds, lsi->lsi_ldd->ldd_mount_opts);
1958         fsoptions_to_mds_flags(mds, lsi->lsi_lmd->lmd_opts);
1959         mnt = lmi->lmi_mnt;
1960         obd->obd_fsops = fsfilt_get_ops(MT_STR(lsi->lsi_ldd));
1961         if (IS_ERR(obd->obd_fsops))
1962                 GOTO(err_put, rc = PTR_ERR(obd->obd_fsops));
1963
1964         CDEBUG(D_SUPER, "%s: mnt = %p\n", lustre_cfg_string(lcfg, 1), mnt);
1965
1966         if (lvfs_check_rdonly(lvfs_sbdev(mnt->mnt_sb))) {
1967                 CERROR("%s: Underlying device is marked as read-only. "
1968                        "Setup failed\n", obd->obd_name);
1969                 GOTO(err_ops, rc = -EROFS);
1970         }
1971
1972         sema_init(&mds->mds_epoch_sem, 1);
1973         spin_lock_init(&mds->mds_transno_lock);
1974         mds->mds_max_mdsize = sizeof(struct lov_mds_md_v3);
1975         mds->mds_max_cookiesize = sizeof(struct llog_cookie);
1976         mds->mds_atime_diff = MAX_ATIME_DIFF;
1977         mds->mds_evict_ost_nids = 1;
1978         /* sync permission changes */
1979         mds->mds_sync_permission = 0;
1980
1981         sprintf(ns_name, "mds-%s", obd->obd_uuid.uuid);
1982         obd->obd_namespace = ldlm_namespace_new(obd, ns_name, LDLM_NAMESPACE_SERVER,
1983                                                 LDLM_NAMESPACE_GREEDY);
1984         if (obd->obd_namespace == NULL) {
1985                 mds_cleanup(obd);
1986                 GOTO(err_ops, rc = -ENOMEM);
1987         }
1988         ldlm_register_intent(obd->obd_namespace, mds_intent_policy);
1989
1990         lprocfs_mds_init_vars(&lvars);
1991         if (lprocfs_obd_setup(obd, lvars.obd_vars) == 0 &&
1992             lprocfs_alloc_obd_stats(obd, LPROC_MDS_LAST) == 0) {
1993                 /* Init private stats here */
1994                 mds_stats_counter_init(obd->obd_stats);
1995 #ifdef HAVE_DELAYED_RECOVERY
1996                 lprocfs_obd_attach_stale_exports(obd);
1997 #endif
1998                 obd->obd_proc_exports_entry = proc_mkdir("exports",
1999                                                          obd->obd_proc_entry);
2000         }
2001
2002         rc = mds_fs_setup(obd, mnt);
2003         if (rc) {
2004                 CERROR("%s: MDS filesystem method init failed: rc = %d\n",
2005                        obd->obd_name, rc);
2006                 GOTO(err_ns, rc);
2007         }
2008
2009         if (obd->obd_proc_exports_entry)
2010                 lprocfs_add_simple(obd->obd_proc_exports_entry,
2011                                    "clear", lprocfs_nid_stats_clear_read,
2012                                    lprocfs_nid_stats_clear_write, obd, NULL);
2013
2014         if (lcfg->lcfg_bufcount >= 4 && LUSTRE_CFG_BUFLEN(lcfg, 3) > 0) {
2015                 class_uuid_t uuid;
2016
2017                 ll_generate_random_uuid(uuid);
2018                 class_uuid_unparse(uuid, &mds->mds_lov_uuid);
2019
2020                 OBD_ALLOC(mds->mds_profile, LUSTRE_CFG_BUFLEN(lcfg, 3));
2021                 if (mds->mds_profile == NULL)
2022                         GOTO(err_fs, rc = -ENOMEM);
2023
2024                 strncpy(mds->mds_profile, lustre_cfg_string(lcfg, 3),
2025                         LUSTRE_CFG_BUFLEN(lcfg, 3));
2026         }
2027
2028         ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL,
2029                            "mds_ldlm_client", &obd->obd_ldlm_client);
2030         obd->obd_replayable = 1;
2031
2032         rc = lquota_setup(mds_quota_interface_ref, obd);
2033         if (rc)
2034                 GOTO(err_fs, rc);
2035
2036         mds->mds_group_hash = upcall_cache_init(obd->obd_name);
2037         if (IS_ERR(mds->mds_group_hash)) {
2038                 rc = PTR_ERR(mds->mds_group_hash);
2039                 mds->mds_group_hash = NULL;
2040                 GOTO(err_qctxt, rc);
2041         }
2042
2043         /* Don't wait for mds_postrecov trying to clear orphans */
2044         obd->obd_async_recov = 1;
2045         rc = mds_postsetup(obd);
2046         /* Bug 11557 - allow async abort_recov start
2047            FIXME can remove most of this obd_async_recov plumbing
2048         obd->obd_async_recov = 0;
2049         */
2050         if (rc)
2051                 GOTO(err_qctxt, rc);
2052
2053         uuid_ptr = fsfilt_uuid(obd, obd->u.obt.obt_sb);
2054         if (uuid_ptr != NULL) {
2055                 class_uuid_unparse(uuid_ptr, &uuid);
2056                 str = uuid.uuid;
2057         } else {
2058                 str = "no UUID";
2059         }
2060
2061         label = fsfilt_get_label(obd, obd->u.obt.obt_sb);
2062         LCONSOLE_INFO("%s: Now serving %s on %s with recovery %s\n",
2063                       obd->obd_name, label ?: str, lsi->lsi_lmd->lmd_dev,
2064                       obd->obd_replayable ? "enabled" : "disabled");
2065
2066         if (obd->obd_recovering)
2067                 LCONSOLE_WARN("%s: Will be in recovery for at least %d:%.02d, "
2068                               "or until %d client%s reconnect%s\n",
2069                               obd->obd_name,
2070                               obd->obd_recovery_timeout / 60,
2071                               obd->obd_recovery_timeout % 60,
2072                               obd->obd_recoverable_clients,
2073                               obd->obd_recoverable_clients == 1 ? "" : "s",
2074                               obd->obd_recoverable_clients == 1 ? "s": "");
2075
2076         /* Reduce the initial timeout on an MDS because it doesn't need such
2077          * a long timeout as an OST does. Adaptive timeouts will adjust this
2078          * value appropriately. */
2079         if (ldlm_timeout == LDLM_TIMEOUT_DEFAULT)
2080                 ldlm_timeout = MDS_LDLM_TIMEOUT_DEFAULT;
2081
2082         RETURN(0);
2083
2084 err_qctxt:
2085         lquota_cleanup(mds_quota_interface_ref, obd);
2086 err_fs:
2087         /* No extra cleanup needed for llog_init_commit_thread() */
2088         mds_fs_cleanup(obd);
2089         upcall_cache_cleanup(mds->mds_group_hash);
2090         mds->mds_group_hash = NULL;
2091         remove_proc_entry("clear", obd->obd_proc_exports_entry);
2092 err_ns:
2093         lprocfs_free_per_client_stats(obd);
2094         lprocfs_free_obd_stats(obd);
2095         lprocfs_obd_cleanup(obd);
2096         ldlm_namespace_free(obd->obd_namespace, NULL, 0);
2097         obd->obd_namespace = NULL;
2098 err_ops:
2099         fsfilt_put_ops(obd->obd_fsops);
2100 err_put:
2101         server_put_mount(obd->obd_name, mnt);
2102         obd->u.obt.obt_sb = NULL;
2103         return rc;
2104 }
2105
2106 static int mds_lov_clean(struct obd_device *obd)
2107 {
2108         struct mds_obd *mds = &obd->u.mds;
2109         struct obd_device *lov = mds->mds_lov_obd;
2110         ENTRY;
2111
2112         if (mds->mds_profile) {
2113                 class_del_profile(mds->mds_profile);
2114                 OBD_FREE(mds->mds_profile, strlen(mds->mds_profile) + 1);
2115                 mds->mds_profile = NULL;
2116         }
2117
2118         /* There better be a lov */
2119         if (!lov)
2120                 RETURN(0);
2121         if (IS_ERR(lov))
2122                 RETURN(PTR_ERR(lov));
2123
2124         obd_register_observer(lov, NULL);
2125
2126         /* Give lov our same shutdown flags */
2127         lov->obd_force = obd->obd_force;
2128         lov->obd_fail = obd->obd_fail;
2129
2130         /* Cleanup the lov */
2131         obd_disconnect(mds->mds_lov_exp);
2132         class_manual_cleanup(lov);
2133         mds->mds_lov_exp = NULL;
2134
2135         RETURN(0);
2136 }
2137
2138 static int mds_postsetup(struct obd_device *obd)
2139 {
2140         struct mds_obd *mds = &obd->u.mds;
2141         struct llog_ctxt *ctxt;
2142         int rc = 0;
2143         ENTRY;
2144
2145         rc = llog_setup(obd, LLOG_CONFIG_ORIG_CTXT, obd, 0, NULL,
2146                         &llog_lvfs_ops);
2147         if (rc)
2148                 RETURN(rc);
2149
2150         rc = llog_setup(obd, LLOG_LOVEA_ORIG_CTXT, obd, 0, NULL,
2151                         &llog_lvfs_ops);
2152         if (rc)
2153                 GOTO(err_llog, rc);
2154
2155         if (mds->mds_profile) {
2156                 struct lustre_profile *lprof;
2157                 /* The profile defines which osc and mdc to connect to, for a
2158                    client.  We reuse that here to figure out the name of the
2159                    lov to use (and ignore lprof->lp_mdc).
2160                    The profile was set in the config log with
2161                    LCFG_MOUNTOPT profilenm oscnm mdcnm */
2162                 lprof = class_get_profile(mds->mds_profile);
2163                 if (lprof == NULL) {
2164                         CERROR("No profile found: %s\n", mds->mds_profile);
2165                         GOTO(err_cleanup, rc = -ENOENT);
2166                 }
2167                 rc = mds_lov_connect(obd, lprof->lp_osc);
2168                 if (rc)
2169                         GOTO(err_cleanup, rc);
2170         }
2171
2172         RETURN(rc);
2173
2174 err_cleanup:
2175         mds_lov_clean(obd);
2176         ctxt = llog_get_context(obd, LLOG_LOVEA_ORIG_CTXT);
2177         if (ctxt)
2178                 llog_cleanup(ctxt);
2179 err_llog:
2180         ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
2181         if (ctxt)
2182                 llog_cleanup(ctxt);
2183         return rc;
2184 }
2185
2186 int mds_postrecov(struct obd_device *obd)
2187 {
2188         int rc;
2189         ENTRY;
2190
2191         if (obd->obd_fail)
2192                 RETURN(0);
2193
2194         LASSERT(!obd->obd_recovering);
2195
2196         /* VBR: update boot epoch after recovery */
2197         mds_update_last_epoch(obd);
2198
2199         /* clean PENDING dir */
2200         rc = mds_cleanup_pending(obd);
2201         if (rc < 0)
2202                 GOTO(out, rc);
2203         /* FIXME Does target_finish_recovery really need this to block? */
2204         /* Notify the LOV, which will in turn call mds_notify for each tgt */
2205         /* This means that we have to hack obd_notify to think we're obd_set_up
2206            during mds_lov_connect. */
2207         obd_notify(obd->u.mds.mds_lov_obd, NULL,
2208                    obd->obd_async_recov ? OBD_NOTIFY_SYNC_NONBLOCK :
2209                    OBD_NOTIFY_SYNC, NULL);
2210
2211         /* quota recovery */
2212         if (likely(obd->obd_stopping == 0))
2213                 lquota_recovery(mds_quota_interface_ref, obd);
2214
2215 out:
2216         RETURN(rc);
2217 }
2218
2219 /* We need to be able to stop an mds_lov_synchronize */
2220 static int mds_lov_early_clean(struct obd_device *obd)
2221 {
2222         struct mds_obd *mds = &obd->u.mds;
2223         struct obd_device *lov = mds->mds_lov_obd;
2224
2225         if (!lov || (!obd->obd_force && !obd->obd_fail))
2226                 return(0);
2227
2228         CDEBUG(D_HA, "abort inflight\n");
2229         return (obd_precleanup(lov, OBD_CLEANUP_EARLY));
2230 }
2231
2232 static int mds_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
2233 {
2234         int rc = 0;
2235         ENTRY;
2236
2237         switch (stage) {
2238         case OBD_CLEANUP_EARLY:
2239                 break;
2240         case OBD_CLEANUP_EXPORTS:
2241                 target_cleanup_recovery(obd);
2242                 mds_lov_early_clean(obd);
2243                 break;
2244         case OBD_CLEANUP_SELF_EXP:
2245                 mds_lov_disconnect(obd);
2246                 mds_lov_clean(obd);
2247                 llog_cleanup(llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT));
2248                 llog_cleanup(llog_get_context(obd, LLOG_LOVEA_ORIG_CTXT));
2249                 rc = obd_llog_finish(obd, 0);
2250                 break;
2251         case OBD_CLEANUP_OBD:
2252                 break;
2253         }
2254         RETURN(rc);
2255 }
2256
2257 static int mds_cleanup(struct obd_device *obd)
2258 {
2259         struct mds_obd *mds = &obd->u.mds;
2260         lvfs_sbdev_type save_dev;
2261         ENTRY;
2262
2263         if (obd->u.obt.obt_sb == NULL)
2264                 RETURN(0);
2265         save_dev = lvfs_sbdev(obd->u.obt.obt_sb);
2266
2267         if (mds->mds_lov_exp)
2268                 /* lov export was disconnected by mds_lov_clean;
2269                    we just need to drop our ref */
2270                 class_export_put(mds->mds_lov_exp);
2271
2272         remove_proc_entry("clear", obd->obd_proc_exports_entry);
2273         lprocfs_free_per_client_stats(obd);
2274         lprocfs_free_obd_stats(obd);
2275         lprocfs_obd_cleanup(obd);
2276
2277         lquota_cleanup(mds_quota_interface_ref, obd);
2278
2279         mds_update_server_data(obd, 1);
2280         mds_fs_cleanup(obd);
2281
2282         upcall_cache_cleanup(mds->mds_group_hash);
2283         mds->mds_group_hash = NULL;
2284
2285         server_put_mount(obd->obd_name, mds->mds_vfsmnt);
2286         obd->u.obt.obt_sb = NULL;
2287
2288         ldlm_namespace_free(obd->obd_namespace, NULL, obd->obd_force);
2289         obd->obd_namespace = NULL;
2290
2291         spin_lock_bh(&obd->obd_processing_task_lock);
2292         if (obd->obd_recovering) {
2293                 target_cancel_recovery_timer(obd);
2294                 obd->obd_recovering = 0;
2295         }
2296         spin_unlock_bh(&obd->obd_processing_task_lock);
2297
2298         fsfilt_put_ops(obd->obd_fsops);
2299
2300         LCONSOLE_INFO("MDT %s has stopped.\n", obd->obd_name);
2301
2302         RETURN(0);
2303 }
2304
2305 static void fixup_handle_for_resent_req(struct ptlrpc_request *req, int offset,
2306                                         struct ldlm_lock *new_lock,
2307                                         struct ldlm_lock **old_lock,
2308                                         struct lustre_handle *lockh)
2309 {
2310         struct obd_export *exp = req->rq_export;
2311         struct ldlm_request *dlmreq =
2312                 lustre_msg_buf(req->rq_reqmsg, offset, sizeof(*dlmreq));
2313         struct lustre_handle remote_hdl = dlmreq->lock_handle[0];
2314         struct ldlm_lock *lock;
2315
2316         if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT))
2317                 return;
2318
2319         lock = lustre_hash_lookup(exp->exp_lock_hash, &remote_hdl);
2320         if (lock) {
2321                 if (lock != new_lock) {
2322                         lockh->cookie = lock->l_handle.h_cookie;
2323                         LDLM_DEBUG(lock, "restoring lock cookie");
2324                         DEBUG_REQ(D_DLMTRACE, req, "restoring lock cookie "
2325                                   LPX64, lockh->cookie);
2326                         if (old_lock)
2327                                 *old_lock = LDLM_LOCK_GET(lock);
2328
2329                         lh_put(exp->exp_lock_hash, &lock->l_exp_hash);
2330                         return;
2331                 }
2332                 lh_put(exp->exp_lock_hash, &lock->l_exp_hash);
2333         }
2334
2335         /* If the xid matches, then we know this is a resent request,
2336          * and allow it. (It's probably an OPEN, for which we don't
2337          * send a lock */
2338         if (req->rq_xid <=
2339             le64_to_cpu(exp->exp_mds_data.med_lcd->lcd_last_xid))
2340                 return;
2341
2342         if (req->rq_xid <=
2343             le64_to_cpu(exp->exp_mds_data.med_lcd->lcd_last_close_xid))
2344                 return;
2345
2346         /* This remote handle isn't enqueued, so we never received or
2347          * processed this request.  Clear MSG_RESENT, because it can
2348          * be handled like any normal request now. */
2349
2350         lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
2351
2352         DEBUG_REQ(D_DLMTRACE, req, "no existing lock with rhandle "LPX64,
2353                   remote_hdl.cookie);
2354 }
2355
2356 #define IS_CLIENT_DISCONNECT_ERROR(error) \
2357                 (error == -ENOTCONN || error == -ENODEV)
2358
2359 static int mds_intent_policy(struct ldlm_namespace *ns,
2360                              struct ldlm_lock **lockp, void *req_cookie,
2361                              ldlm_mode_t mode, int flags, void *data)
2362 {
2363         struct ptlrpc_request *req = req_cookie;
2364         struct ldlm_lock *lock = *lockp;
2365         struct ldlm_intent *it;
2366         struct mds_obd *mds = &req->rq_export->exp_obd->u.mds;
2367         struct ldlm_reply *rep;
2368         struct lustre_handle lockh = { 0 };
2369         struct ldlm_lock *new_lock = NULL;
2370         int getattr_part = MDS_INODELOCK_UPDATE;
2371         int repsize[5] = { [MSG_PTLRPC_BODY_OFF] = sizeof(struct ptlrpc_body),
2372                            [DLM_LOCKREPLY_OFF]   = sizeof(struct ldlm_reply),
2373                            [DLM_REPLY_REC_OFF]   = sizeof(struct mds_body),
2374                            [DLM_REPLY_REC_OFF+1] = mds->mds_max_mdsize };
2375         int repbufcnt = 4, rc;
2376         ENTRY;
2377
2378         LASSERT(req != NULL);
2379
2380         if (lustre_msg_bufcount(req->rq_reqmsg) <= DLM_INTENT_IT_OFF) {
2381                 /* No intent was provided */
2382                 rc = lustre_pack_reply(req, 2, repsize, NULL);
2383                 if (rc)
2384                         RETURN(rc);
2385                 RETURN(0);
2386         }
2387
2388         it = lustre_swab_reqbuf(req, DLM_INTENT_IT_OFF, sizeof(*it),
2389                                 lustre_swab_ldlm_intent);
2390         if (it == NULL) {
2391                 CERROR("Intent missing\n");
2392                 RETURN(req->rq_status = -EFAULT);
2393         }
2394
2395         LDLM_DEBUG(lock, "intent policy, opc: %s", ldlm_it2str(it->opc));
2396
2397         if ((req->rq_export->exp_connect_flags & OBD_CONNECT_ACL) &&
2398             (it->opc & (IT_OPEN | IT_GETATTR | IT_LOOKUP | IT_READDIR)))
2399                 /* we should never allow OBD_CONNECT_ACL if not configured */
2400                 repsize[repbufcnt++] = LUSTRE_POSIX_ACL_MAX_SIZE;
2401         else if (it->opc & IT_UNLINK)
2402                 repsize[repbufcnt++] = mds->mds_max_cookiesize;
2403
2404         /* if we do recovery we isn't send reply mds state is restored */
2405         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) {
2406                 repsize[DLM_REPLY_REC_OFF+1] = 0;
2407                 if (it->opc & IT_UNLINK)
2408                         repsize[DLM_REPLY_REC_OFF+2] = 0;
2409         }
2410
2411         rc = lustre_pack_reply(req, repbufcnt, repsize, NULL);
2412         if (rc)
2413                 RETURN(req->rq_status = rc);
2414
2415         rep = lustre_msg_buf(req->rq_repmsg, DLM_LOCKREPLY_OFF, sizeof(*rep));
2416         ldlm_reply_set_disposition(rep, DISP_IT_EXECD);
2417
2418         /* execute policy */
2419         switch ((long)it->opc) {
2420         case IT_OPEN:
2421         case IT_CREAT|IT_OPEN:
2422                 mds_counter_incr(req->rq_export, LPROC_MDS_OPEN);
2423                 fixup_handle_for_resent_req(req, DLM_LOCKREQ_OFF, lock, NULL,
2424                                             &lockh);
2425                 /* XXX swab here to assert that an mds_open reint
2426                  * packet is following */
2427                 rep->lock_policy_res2 = mds_reint(req, DLM_INTENT_REC_OFF,
2428                                                   &lockh);
2429                 mds_shrink_intent_reply(req, REINT_OPEN, DLM_REPLY_REC_OFF);
2430 #if 0
2431                 /* We abort the lock if the lookup was negative and
2432                  * we did not make it to the OPEN portion */
2433                 if (!ldlm_reply_disposition(rep, DISP_LOOKUP_EXECD))
2434                         RETURN(ELDLM_LOCK_ABORTED);
2435                 if (ldlm_reply_disposition(rep, DISP_LOOKUP_NEG) &&
2436                     !ldlm_reply_disposition(rep, DISP_OPEN_OPEN))
2437 #endif
2438
2439                 /* If there was an error of some sort or if we are not
2440                  * returning any locks */
2441                  if (rep->lock_policy_res2 ||
2442                      !ldlm_reply_disposition(rep, DISP_OPEN_LOCK)) {
2443                         /* If it is the disconnect error (ENODEV & ENOCONN)
2444                          * ptlrpc layer should know this imediately, it should
2445                          * be replied by rq_stats, otherwise, return it by
2446                          * intent here
2447                          */
2448                          /* if VBR failure then return error in rq_stats too */
2449                         if (IS_CLIENT_DISCONNECT_ERROR(rep->lock_policy_res2) ||
2450                             rep->lock_policy_res2 == -EOVERFLOW)
2451                                 RETURN(rep->lock_policy_res2);
2452                         else
2453                                 RETURN(ELDLM_LOCK_ABORTED);
2454                  }
2455                 break;
2456         case IT_LOOKUP:
2457                         getattr_part = MDS_INODELOCK_LOOKUP;
2458         case IT_GETATTR:
2459                         getattr_part |= MDS_INODELOCK_LOOKUP;
2460                         OBD_COUNTER_INCREMENT(req->rq_export->exp_obd, getattr);
2461         case IT_READDIR:
2462                 fixup_handle_for_resent_req(req, DLM_LOCKREQ_OFF, lock,
2463                                             &new_lock, &lockh);
2464
2465                 /* INODEBITS_INTEROP: if this lock was converted from a
2466                  * plain lock (client does not support inodebits), then
2467                  * child lock must be taken with both lookup and update
2468                  * bits set for all operations.
2469                  */
2470                 if (!(req->rq_export->exp_connect_flags & OBD_CONNECT_IBITS))
2471                         getattr_part = MDS_INODELOCK_LOOKUP |
2472                                        MDS_INODELOCK_UPDATE;
2473
2474                 rep->lock_policy_res2 = mds_getattr_lock(req,DLM_INTENT_REC_OFF,
2475                                                          getattr_part, &lockh);
2476                 mds_shrink_body_reply(req,DLM_INTENT_REC_OFF, DLM_REPLY_REC_OFF);
2477                 /* FIXME: LDLM can set req->rq_status. MDS sets
2478                    policy_res{1,2} with disposition and status.
2479                    - replay: returns 0 & req->status is old status
2480                    - otherwise: returns req->status */
2481                 if (ldlm_reply_disposition(rep, DISP_LOOKUP_NEG))
2482                         rep->lock_policy_res2 = 0;
2483                 if (!ldlm_reply_disposition(rep, DISP_LOOKUP_POS) ||
2484                     rep->lock_policy_res2)
2485                         RETURN(ELDLM_LOCK_ABORTED);
2486                 if (req->rq_status != 0) {
2487                         LBUG();
2488                         rep->lock_policy_res2 = req->rq_status;
2489                         RETURN(ELDLM_LOCK_ABORTED);
2490                 }
2491                 break;
2492         default:
2493                 CERROR("Unhandled intent "LPD64"\n", it->opc);
2494                 RETURN(-EFAULT);
2495         }
2496
2497         /* By this point, whatever function we called above must have either
2498          * filled in 'lockh', been an intent replay, or returned an error.  We
2499          * want to allow replayed RPCs to not get a lock, since we would just
2500          * drop it below anyways because lock replay is done separately by the
2501          * client afterwards.  For regular RPCs we want to give the new lock to
2502          * the client instead of whatever lock it was about to get. */
2503         if (new_lock == NULL)
2504                 new_lock = ldlm_handle2lock(&lockh);
2505         if (new_lock == NULL && (flags & LDLM_FL_INTENT_ONLY))
2506                 RETURN(0);
2507
2508         LASSERTF(new_lock != NULL, "op "LPX64" lockh "LPX64"\n",
2509                  it->opc, lockh.cookie);
2510
2511         /* If we've already given this lock to a client once, then we should
2512          * have no readers or writers.  Otherwise, we should have one reader
2513          * _or_ writer ref (which will be zeroed below) before returning the
2514          * lock to a client. */
2515         if (new_lock->l_export == req->rq_export) {
2516                 LASSERT(new_lock->l_readers + new_lock->l_writers == 0);
2517         } else {
2518                 LASSERT(new_lock->l_export == NULL);
2519                 LASSERT(new_lock->l_readers + new_lock->l_writers == 1);
2520         }
2521
2522         *lockp = new_lock;
2523
2524         if (new_lock->l_export == req->rq_export) {
2525                 /* Already gave this to the client, which means that we
2526                  * reconstructed a reply. */
2527                 LASSERT(lustre_msg_get_flags(req->rq_reqmsg) &
2528                         MSG_RESENT);
2529                 RETURN(ELDLM_LOCK_REPLACED);
2530         }
2531
2532         /* Fixup the lock to be given to the client */
2533         lock_res_and_lock(new_lock);
2534         new_lock->l_readers = 0;
2535         new_lock->l_writers = 0;
2536
2537         new_lock->l_export = class_export_get(req->rq_export);
2538         new_lock->l_blocking_ast = lock->l_blocking_ast;
2539         new_lock->l_completion_ast = lock->l_completion_ast;
2540         new_lock->l_flags &= ~LDLM_FL_LOCAL;
2541
2542         memcpy(&new_lock->l_remote_handle, &lock->l_remote_handle,
2543                sizeof(lock->l_remote_handle));
2544
2545         unlock_res_and_lock(new_lock);
2546
2547         lustre_hash_add(new_lock->l_export->exp_lock_hash,
2548                         &new_lock->l_remote_handle,
2549                         &new_lock->l_exp_hash);
2550         LDLM_LOCK_PUT(new_lock);
2551
2552         RETURN(ELDLM_LOCK_REPLACED);
2553 }
2554
2555 static int mdt_setup(struct obd_device *obd, obd_count len, void *buf)
2556 {
2557         struct mds_obd *mds = &obd->u.mds;
2558         struct lprocfs_static_vars lvars;
2559         int mds_min_threads;
2560         int mds_max_threads;
2561         int rc = 0;
2562         ENTRY;
2563
2564         lprocfs_mdt_init_vars(&lvars);
2565         lprocfs_obd_setup(obd, lvars.obd_vars);
2566
2567         sema_init(&mds->mds_health_sem, 1);
2568
2569         if (mds_num_threads) {
2570                 /* If mds_num_threads is set, it is the min and the max. */
2571                 if (mds_num_threads > MDS_THREADS_MAX)
2572                         mds_num_threads = MDS_THREADS_MAX;
2573                 if (mds_num_threads < MDS_THREADS_MIN)
2574                         mds_num_threads = MDS_THREADS_MIN;
2575                 mds_max_threads = mds_min_threads = mds_num_threads;
2576         } else {
2577                 /* Base min threads on memory and cpus */
2578                 mds_min_threads = num_possible_cpus() * CFS_NUM_CACHEPAGES >>
2579                         (27 - CFS_PAGE_SHIFT);
2580                 if (mds_min_threads < MDS_THREADS_MIN)
2581                         mds_min_threads = MDS_THREADS_MIN;
2582                 /* Largest auto threads start value */
2583                 if (mds_min_threads > 32)
2584                         mds_min_threads = 32;
2585                 mds_max_threads = min(MDS_THREADS_MAX, mds_min_threads * 4);
2586         }
2587
2588         mds->mds_service =
2589                 ptlrpc_init_svc(MDS_NBUFS, MDS_BUFSIZE, MDS_MAXREQSIZE,
2590                                 MDS_MAXREPSIZE, MDS_REQUEST_PORTAL,
2591                                 MDC_REPLY_PORTAL, MDS_SERVICE_WATCHDOG_FACTOR,
2592                                 mds_handle, LUSTRE_MDS_NAME,
2593                                 obd->obd_proc_entry, target_print_req,
2594                                 mds_min_threads, mds_max_threads, "ll_mdt",
2595                                 NULL);
2596
2597         if (!mds->mds_service) {
2598                 CERROR("failed to start service\n");
2599                 GOTO(err_lprocfs, rc = -ENOMEM);
2600         }
2601
2602         rc = ptlrpc_start_threads(obd, mds->mds_service);
2603         if (rc)
2604                 GOTO(err_thread, rc);
2605
2606         mds->mds_setattr_service =
2607                 ptlrpc_init_svc(MDS_NBUFS, MDS_BUFSIZE, MDS_MAXREQSIZE,
2608                                 MDS_MAXREPSIZE, MDS_SETATTR_PORTAL,
2609                                 MDC_REPLY_PORTAL, MDS_SERVICE_WATCHDOG_FACTOR,
2610                                 mds_handle, "mds_setattr",
2611                                 obd->obd_proc_entry, target_print_req,
2612                                 mds_min_threads, mds_max_threads,
2613                                 "ll_mdt_attr", NULL);
2614         if (!mds->mds_setattr_service) {
2615                 CERROR("failed to start getattr service\n");
2616                 GOTO(err_thread, rc = -ENOMEM);
2617         }
2618
2619         rc = ptlrpc_start_threads(obd, mds->mds_setattr_service);
2620         if (rc)
2621                 GOTO(err_thread2, rc);
2622
2623         mds->mds_readpage_service =
2624                 ptlrpc_init_svc(MDS_NBUFS, MDS_BUFSIZE, MDS_MAXREQSIZE,
2625                                 MDS_MAXREPSIZE, MDS_READPAGE_PORTAL,
2626                                 MDC_REPLY_PORTAL, MDS_SERVICE_WATCHDOG_FACTOR,
2627                                 mds_handle, "mds_readpage",
2628                                 obd->obd_proc_entry, target_print_req,
2629                                 MDS_THREADS_MIN_READPAGE, mds_max_threads,
2630                                 "ll_mdt_rdpg", NULL);
2631         if (!mds->mds_readpage_service) {
2632                 CERROR("failed to start readpage service\n");
2633                 GOTO(err_thread2, rc = -ENOMEM);
2634         }
2635
2636         rc = ptlrpc_start_threads(obd, mds->mds_readpage_service);
2637
2638         if (rc)
2639                 GOTO(err_thread3, rc);
2640
2641         ping_evictor_start();
2642
2643         RETURN(0);
2644
2645 err_thread3:
2646         ptlrpc_unregister_service(mds->mds_readpage_service);
2647         mds->mds_readpage_service = NULL;
2648 err_thread2:
2649         ptlrpc_unregister_service(mds->mds_setattr_service);
2650         mds->mds_setattr_service = NULL;
2651 err_thread:
2652         ptlrpc_unregister_service(mds->mds_service);
2653         mds->mds_service = NULL;
2654 err_lprocfs:
2655         lprocfs_obd_cleanup(obd);
2656         return rc;
2657 }
2658
2659 static int mdt_cleanup(struct obd_device *obd)
2660 {
2661         struct mds_obd *mds = &obd->u.mds;
2662         ENTRY;
2663
2664         ping_evictor_stop();
2665
2666         down(&mds->mds_health_sem);
2667         ptlrpc_unregister_service(mds->mds_readpage_service);
2668         ptlrpc_unregister_service(mds->mds_setattr_service);
2669         ptlrpc_unregister_service(mds->mds_service);
2670         mds->mds_readpage_service = NULL;
2671         mds->mds_setattr_service = NULL;
2672         mds->mds_service = NULL;
2673         up(&mds->mds_health_sem);
2674
2675         lprocfs_obd_cleanup(obd);
2676
2677         RETURN(0);
2678 }
2679
2680 static int mdt_health_check(struct obd_device *obd)
2681 {
2682         struct mds_obd *mds = &obd->u.mds;
2683         int rc = 0;
2684
2685         down(&mds->mds_health_sem);
2686         rc |= ptlrpc_service_health_check(mds->mds_readpage_service);
2687         rc |= ptlrpc_service_health_check(mds->mds_setattr_service);
2688         rc |= ptlrpc_service_health_check(mds->mds_service);
2689         up(&mds->mds_health_sem);
2690
2691         /*
2692          * health_check to return 0 on healthy
2693          * and 1 on unhealthy.
2694          */
2695         if(rc != 0)
2696                 rc = 1;
2697
2698         return rc;
2699 }
2700
2701 static struct dentry *mds_lvfs_fid2dentry(__u64 id, __u32 gen, __u64 gr,
2702                                           void *data)
2703 {
2704         struct obd_device *obd = data;
2705         struct ll_fid fid;
2706         fid.id = id;
2707         fid.generation = gen;
2708         return mds_fid2dentry(&obd->u.mds, &fid, NULL);
2709 }
2710
2711 static int mds_health_check(struct obd_device *obd)
2712 {
2713         struct obd_device_target *odt = &obd->u.obt;
2714 #ifdef USE_HEALTH_CHECK_WRITE
2715         struct mds_obd *mds = &obd->u.mds;
2716 #endif
2717         int rc = 0;
2718
2719         if (odt->obt_sb->s_flags & MS_RDONLY)
2720                 rc = 1;
2721
2722 #ifdef USE_HEALTH_CHECK_WRITE
2723         LASSERT(mds->mds_obt.obt_health_check_filp != NULL);
2724         rc |= !!lvfs_check_io_health(obd, mds->mds_obt.obt_health_check_filp);
2725 #endif
2726
2727         return rc;
2728 }
2729
2730 static int mds_process_config(struct obd_device *obd, obd_count len, void *buf)
2731 {
2732         struct lustre_cfg *lcfg = buf;
2733         int rc = 0;
2734
2735         switch(lcfg->lcfg_command) {
2736         case LCFG_PARAM: {
2737                 struct lprocfs_static_vars lvars;
2738                 lprocfs_mds_init_vars(&lvars);
2739
2740                 rc = class_process_proc_param(PARAM_MDT, lvars.obd_vars, lcfg, obd);
2741                 break;
2742         }
2743         default:
2744                 break;
2745         }
2746
2747         return(rc);
2748 }
2749
2750 struct lvfs_callback_ops mds_lvfs_ops = {
2751         l_fid2dentry:     mds_lvfs_fid2dentry,
2752 };
2753
2754 /* use obd ops to offer management infrastructure */
2755 static struct obd_ops mds_obd_ops = {
2756         .o_owner           = THIS_MODULE,
2757         .o_connect         = mds_connect,
2758         .o_reconnect       = mds_reconnect,
2759         .o_init_export     = mds_init_export,
2760         .o_destroy_export  = mds_destroy_export,
2761         .o_disconnect      = mds_disconnect,
2762         .o_setup           = mds_setup,
2763         .o_precleanup      = mds_precleanup,
2764         .o_cleanup         = mds_cleanup,
2765         .o_postrecov       = mds_postrecov,
2766         .o_statfs          = mds_obd_statfs,
2767         .o_iocontrol       = mds_iocontrol,
2768         .o_create          = mds_obd_create,
2769         .o_destroy         = mds_obd_destroy,
2770         .o_llog_init       = mds_llog_init,
2771         .o_llog_finish     = mds_llog_finish,
2772         .o_notify          = mds_notify,
2773         .o_health_check    = mds_health_check,
2774         .o_process_config  = mds_process_config,
2775 };
2776
2777 static struct obd_ops mdt_obd_ops = {
2778         .o_owner           = THIS_MODULE,
2779         .o_setup           = mdt_setup,
2780         .o_cleanup         = mdt_cleanup,
2781         .o_health_check    = mdt_health_check,
2782 };
2783
2784 quota_interface_t *mds_quota_interface_ref;
2785 extern quota_interface_t mds_quota_interface;
2786
2787 static int __init mds_init(void)
2788 {
2789         int rc;
2790         struct lprocfs_static_vars lvars;
2791
2792         request_module("lquota");
2793         mds_quota_interface_ref = PORTAL_SYMBOL_GET(mds_quota_interface);
2794         rc = lquota_init(mds_quota_interface_ref);
2795         if (rc) {
2796                 if (mds_quota_interface_ref)
2797                         PORTAL_SYMBOL_PUT(mds_quota_interface);
2798                 return rc;
2799         }
2800         init_obd_quota_ops(mds_quota_interface_ref, &mds_obd_ops);
2801
2802         lprocfs_mds_init_vars(&lvars);
2803         class_register_type(&mds_obd_ops, lvars.module_vars, LUSTRE_MDS_NAME);
2804         lprocfs_mdt_init_vars(&lvars);
2805         class_register_type(&mdt_obd_ops, lvars.module_vars, LUSTRE_MDT_NAME);
2806
2807         return 0;
2808 }
2809
2810 static void /*__exit*/ mds_exit(void)
2811 {
2812         lquota_exit(mds_quota_interface_ref);
2813         if (mds_quota_interface_ref)
2814                 PORTAL_SYMBOL_PUT(mds_quota_interface);
2815
2816         class_unregister_type(LUSTRE_MDS_NAME);
2817         class_unregister_type(LUSTRE_MDT_NAME);
2818 }
2819
2820 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
2821 MODULE_DESCRIPTION("Lustre Metadata Server (MDS)");
2822 MODULE_LICENSE("GPL");
2823
2824 module_init(mds_init);
2825 module_exit(mds_exit);