lustre/mds/handler.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  lustre/mds/handler.c
   5  *  Lustre Metadata Server (mds) request handler
   6  *
   7  *  Copyright (c) 2001-2005 Cluster File Systems, Inc.
   8  *   Author: Peter Braam <braam@clusterfs.com>
   9  *   Author: Andreas Dilger <adilger@clusterfs.com>
  10  *   Author: Phil Schwan <phil@clusterfs.com>
  11  *   Author: Mike Shaver <shaver@clusterfs.com>
  12  *
  13  *   This file is part of Lustre, http://www.lustre.org.
  14  *
  15  *   Lustre is free software; you can redistribute it and/or
  16  *   modify it under the terms of version 2 of the GNU General Public
  17  *   License as published by the Free Software Foundation.
  18  *
  19  *   Lustre is distributed in the hope that it will be useful,
  20  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  21  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  22  *   GNU General Public License for more details.
  23  *
  24  *   You should have received a copy of the GNU General Public License
  25  *   along with Lustre; if not, write to the Free Software
  26  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  27  */
  28
  29 #ifndef EXPORT_SYMTAB
  30 # define EXPORT_SYMTAB
  31 #endif
  32 #define DEBUG_SUBSYSTEM S_MDS
  33
  34 #include <linux/module.h>
  35 #include <linux/lustre_mds.h>
  36 #include <linux/lustre_dlm.h>
  37 #include <linux/init.h>
  38 #include <linux/obd_class.h>
  39 #include <linux/random.h>
  40 #include <linux/fs.h>
  41 #include <linux/jbd.h>
  42 #include <linux/ext3_fs.h>
  43 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
  44 # include <linux/smp_lock.h>
  45 # include <linux/buffer_head.h>
  46 # include <linux/workqueue.h>
  47 # include <linux/mount.h>
  48 #else
  49 # include <linux/locks.h>
  50 #endif
  51 #include <linux/obd_lov.h>
  52 #include <linux/lustre_mds.h>
  53 #include <linux/lustre_fsfilt.h>
  54 #include <linux/lprocfs_status.h>
  55 #include <linux/lustre_commit_confd.h>
  56 #include <linux/lustre_quota.h>
  57
  58 #include "mds_internal.h"
  59
  60 static int mds_intent_policy(struct ldlm_namespace *ns,
  61                              struct ldlm_lock **lockp, void *req_cookie,
  62                              ldlm_mode_t mode, int flags, void *data);
  63 static int mds_postsetup(struct obd_device *obd);
  64 static int mds_cleanup(struct obd_device *obd);
  65
  66 /* Assumes caller has already pushed into the kernel filesystem context */
  67 static int mds_sendpage(struct ptlrpc_request *req, struct file *file,
  68                         loff_t offset, int count)
  69 {
  70         struct ptlrpc_bulk_desc *desc;
  71         struct l_wait_info lwi;
  72         struct page **pages;
  73         int rc = 0, npages, i, tmpcount, tmpsize = 0;
  74         ENTRY;
  75
  76         LASSERT((offset & (PAGE_SIZE - 1)) == 0); /* I'm dubious about this */
  77
  78         npages = (count + PAGE_SIZE - 1) >> PAGE_SHIFT;
  79         OBD_ALLOC(pages, sizeof(*pages) * npages);
  80         if (!pages)
  81                 GOTO(out, rc = -ENOMEM);
  82
  83         desc = ptlrpc_prep_bulk_exp(req, npages, BULK_PUT_SOURCE,
  84                                     MDS_BULK_PORTAL);
  85         if (desc == NULL)
  86                 GOTO(out_free, rc = -ENOMEM);
  87
  88         for (i = 0, tmpcount = count; i < npages; i++, tmpcount -= tmpsize) {
  89                 tmpsize = tmpcount > PAGE_SIZE ? PAGE_SIZE : tmpcount;
  90
  91                 pages[i] = alloc_pages(GFP_KERNEL, 0);
  92                 if (pages[i] == NULL)
  93                         GOTO(cleanup_buf, rc = -ENOMEM);
  94
  95                 ptlrpc_prep_bulk_page(desc, pages[i], 0, tmpsize);
  96         }
  97
  98         for (i = 0, tmpcount = count; i < npages; i++, tmpcount -= tmpsize) {
  99                 tmpsize = tmpcount > PAGE_SIZE ? PAGE_SIZE : tmpcount;
 100                 CDEBUG(D_EXT2, "reading %u@%llu from dir %lu (size %llu)\n",
 101                        tmpsize, offset, file->f_dentry->d_inode->i_ino,
 102                        file->f_dentry->d_inode->i_size);
 103
 104                 rc = fsfilt_readpage(req->rq_export->exp_obd, file,
 105                                      kmap(pages[i]), tmpsize, &offset);
 106                 kunmap(pages[i]);
 107
 108                 if (rc != tmpsize)
 109                         GOTO(cleanup_buf, rc = -EIO);
 110         }
 111
 112         LASSERT(desc->bd_nob == count);
 113
 114         rc = ptlrpc_start_bulk_transfer(desc);
 115         if (rc)
 116                 GOTO(cleanup_buf, rc);
 117
 118         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE)) {
 119                 CERROR("obd_fail_loc=%x, fail operation rc=%d\n",
 120                        OBD_FAIL_MDS_SENDPAGE, rc);
 121                 GOTO(abort_bulk, rc);
 122         }
 123
 124         lwi = LWI_TIMEOUT(obd_timeout * HZ / 4, NULL, NULL);
 125         rc = l_wait_event(desc->bd_waitq, !ptlrpc_bulk_active(desc), &lwi);
 126         LASSERT (rc == 0 || rc == -ETIMEDOUT);
 127
 128         if (rc == 0) {
 129                 if (desc->bd_success &&
 130                     desc->bd_nob_transferred == count)
 131                         GOTO(cleanup_buf, rc);
 132
 133                 rc = -ETIMEDOUT; /* XXX should this be a different errno? */
 134         }
 135
 136         DEBUG_REQ(D_ERROR, req, "bulk failed: %s %d(%d), evicting %s@%s\n",
 137                   (rc == -ETIMEDOUT) ? "timeout" : "network error",
 138                   desc->bd_nob_transferred, count,
 139                   req->rq_export->exp_client_uuid.uuid,
 140                   req->rq_export->exp_connection->c_remote_uuid.uuid);
 141
 142         class_fail_export(req->rq_export);
 143
 144         EXIT;
 145  abort_bulk:
 146         ptlrpc_abort_bulk (desc);
 147  cleanup_buf:
 148         for (i = 0; i < npages; i++)
 149                 if (pages[i])
 150                         __free_pages(pages[i], 0);
 151
 152         ptlrpc_free_bulk(desc);
 153  out_free:
 154         OBD_FREE(pages, sizeof(*pages) * npages);
 155  out:
 156         return rc;
 157 }
 158
 159 /* only valid locked dentries or errors should be returned */
 160 struct dentry *mds_fid2locked_dentry(struct obd_device *obd, struct ll_fid *fid,
 161                                      struct vfsmount **mnt, int lock_mode,
 162                                      struct lustre_handle *lockh,
 163                                      char *name, int namelen)
 164 {
 165         struct mds_obd *mds = &obd->u.mds;
 166         struct dentry *de = mds_fid2dentry(mds, fid, mnt), *retval = de;
 167         struct ldlm_res_id res_id = { .name = {0} };
 168         int flags = 0, rc;
 169         ENTRY;
 170
 171         if (IS_ERR(de))
 172                 RETURN(de);
 173
 174         res_id.name[0] = de->d_inode->i_ino;
 175         res_id.name[1] = de->d_inode->i_generation;
 176         rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, res_id,
 177                               LDLM_PLAIN, NULL, lock_mode, &flags,
 178                               mds_blocking_ast, ldlm_completion_ast, NULL, NULL,
 179                               NULL, 0, NULL, lockh);
 180         if (rc != ELDLM_OK) {
 181                 l_dput(de);
 182                 retval = ERR_PTR(-EIO); /* XXX translate ldlm code */
 183         }
 184
 185         RETURN(retval);
 186 }
 187
 188 /* Look up an entry by inode number. */
 189 /* this function ONLY returns valid dget'd dentries with an initialized inode
 190    or errors */
 191 struct dentry *mds_fid2dentry(struct mds_obd *mds, struct ll_fid *fid,
 192                               struct vfsmount **mnt)
 193 {
 194         char fid_name[32];
 195         unsigned long ino = fid->id;
 196         __u32 generation = fid->generation;
 197         struct inode *inode;
 198         struct dentry *result;
 199
 200         if (ino == 0)
 201                 RETURN(ERR_PTR(-ESTALE));
 202
 203         snprintf(fid_name, sizeof(fid_name), "0x%lx", ino);
 204
 205         CDEBUG(D_DENTRY, "--> mds_fid2dentry: ino/gen %lu/%u, sb %p\n",
 206                ino, generation, mds->mds_sb);
 207
 208         /* under ext3 this is neither supposed to return bad inodes
 209            nor NULL inodes. */
 210         result = ll_lookup_one_len(fid_name, mds->mds_fid_de, strlen(fid_name));
 211         if (IS_ERR(result))
 212                 RETURN(result);
 213
 214         inode = result->d_inode;
 215         if (!inode)
 216                 RETURN(ERR_PTR(-ENOENT));
 217
 218         if (generation && inode->i_generation != generation) {
 219                 /* we didn't find the right inode.. */
 220                 CERROR("bad inode %lu, link: %lu ct: %d or generation %u/%u\n",
 221                        inode->i_ino, (unsigned long)inode->i_nlink,
 222                        atomic_read(&inode->i_count), inode->i_generation,
 223                        generation);
 224                 dput(result);
 225                 RETURN(ERR_PTR(-ENOENT));
 226         }
 227
 228         if (mnt) {
 229                 *mnt = mds->mds_vfsmnt;
 230                 mntget(*mnt);
 231         }
 232
 233         RETURN(result);
 234 }
 235
 236
 237 /* Establish a connection to the MDS.
 238  *
 239  * This will set up an export structure for the client to hold state data
 240  * about that client, like open files, the last operation number it did
 241  * on the server, etc.
 242  */
 243 static int mds_connect(struct lustre_handle *conn, struct obd_device *obd,
 244                        struct obd_uuid *cluuid, struct obd_connect_data *data)
 245 {
 246         struct obd_export *exp;
 247         struct mds_export_data *med;
 248         struct mds_client_data *mcd;
 249         int rc, abort_recovery;
 250         ENTRY;
 251
 252         if (!conn || !obd || !cluuid)
 253                 RETURN(-EINVAL);
 254
 255         /* Check for aborted recovery. */
 256         spin_lock_bh(&obd->obd_processing_task_lock);
 257         abort_recovery = obd->obd_abort_recovery;
 258         spin_unlock_bh(&obd->obd_processing_task_lock);
 259         if (abort_recovery)
 260                 target_abort_recovery(obd);
 261
 262         /* XXX There is a small race between checking the list and adding a
 263          * new connection for the same UUID, but the real threat (list
 264          * corruption when multiple different clients connect) is solved.
 265          *
 266          * There is a second race between adding the export to the list,
 267          * and filling in the client data below.  Hence skipping the case
 268          * of NULL mcd above.  We should already be controlling multiple
 269          * connects at the client, and we can't hold the spinlock over
 270          * memory allocations without risk of deadlocking.
 271          */
 272         rc = class_connect(conn, obd, cluuid);
 273         if (rc)
 274                 RETURN(rc);
 275         exp = class_conn2export(conn);
 276         LASSERT(exp);
 277         med = &exp->exp_mds_data;
 278
 279         if (data != NULL) {
 280                 data->ocd_connect_flags &= MDS_CONNECT_SUPPORTED;
 281                 exp->exp_connect_flags = data->ocd_connect_flags;
 282         }
 283
 284         OBD_ALLOC(mcd, sizeof(*mcd));
 285         if (!mcd) {
 286                 CERROR("mds: out of memory for client data\n");
 287                 GOTO(out, rc = -ENOMEM);
 288         }
 289
 290         memcpy(mcd->mcd_uuid, cluuid, sizeof(mcd->mcd_uuid));
 291         med->med_mcd = mcd;
 292
 293         rc = mds_client_add(obd, &obd->u.mds, med, -1);
 294         GOTO(out, rc);
 295
 296 out:
 297         if (rc) {
 298                 if (mcd) {
 299                         OBD_FREE(mcd, sizeof(*mcd));
 300                         med->med_mcd = NULL;
 301                 }
 302                 class_disconnect(exp);
 303         } else {
 304                 class_export_put(exp);
 305         }
 306
 307         RETURN(rc);
 308 }
 309
 310 static int mds_init_export(struct obd_export *exp)
 311 {
 312         struct mds_export_data *med = &exp->exp_mds_data;
 313
 314         INIT_LIST_HEAD(&med->med_open_head);
 315         spin_lock_init(&med->med_open_lock);
 316         RETURN(0);
 317 }
 318
 319 static int mds_destroy_export(struct obd_export *export)
 320 {
 321         struct mds_export_data *med;
 322         struct obd_device *obd = export->exp_obd;
 323         struct lvfs_run_ctxt saved;
 324         int rc = 0;
 325         ENTRY;
 326
 327         med = &export->exp_mds_data;
 328         target_destroy_export(export);
 329
 330         if (obd_uuid_equals(&export->exp_client_uuid, &obd->obd_uuid))
 331                 GOTO(out, 0);
 332
 333         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 334         /* Close any open files (which may also cause orphan unlinking). */
 335         spin_lock(&med->med_open_lock);
 336         while (!list_empty(&med->med_open_head)) {
 337                 struct list_head *tmp = med->med_open_head.next;
 338                 struct mds_file_data *mfd =
 339                         list_entry(tmp, struct mds_file_data, mfd_list);
 340                 struct dentry *dentry = mfd->mfd_dentry;
 341
 342                 /* Remove mfd handle so it can't be found again.
 343                  * We are consuming the mfd_list reference here. */
 344                 mds_mfd_unlink(mfd, 0);
 345                 spin_unlock(&med->med_open_lock);
 346
 347                 /* If you change this message, be sure to update
 348                  * replay_single:test_46 */
 349                 CDEBUG(D_INODE|D_IOCTL, "%s: force closing file handle for "
 350                        "%.*s (ino %lu)\n", obd->obd_name, dentry->d_name.len,
 351                        dentry->d_name.name, dentry->d_inode->i_ino);
 352                 /* child orphan sem protects orphan_dec_test and
 353                  * is_orphan race, mds_mfd_close drops it */
 354                 MDS_DOWN_WRITE_ORPHAN_SEM(dentry->d_inode);
 355                 rc = mds_mfd_close(NULL, obd, mfd,
 356                                    !(export->exp_flags & OBD_OPT_FAILOVER));
 357
 358                 if (rc)
 359                         CDEBUG(D_INODE|D_IOCTL, "Error closing file: %d\n", rc);
 360                 spin_lock(&med->med_open_lock);
 361         }
 362         spin_unlock(&med->med_open_lock);
 363         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 364 out:
 365         mds_client_free(export);
 366
 367         RETURN(rc);
 368 }
 369
 370 static int mds_disconnect(struct obd_export *exp)
 371 {
 372         unsigned long irqflags;
 373         int rc;
 374         ENTRY;
 375
 376         LASSERT(exp);
 377         class_export_get(exp);
 378
 379         /* Disconnect early so that clients can't keep using export */
 380         rc = class_disconnect(exp);
 381         ldlm_cancel_locks_for_export(exp);
 382
 383         /* complete all outstanding replies */
 384         spin_lock_irqsave(&exp->exp_lock, irqflags);
 385         while (!list_empty(&exp->exp_outstanding_replies)) {
 386                 struct ptlrpc_reply_state *rs =
 387                         list_entry(exp->exp_outstanding_replies.next,
 388                                    struct ptlrpc_reply_state, rs_exp_list);
 389                 struct ptlrpc_service *svc = rs->rs_service;
 390
 391                 spin_lock(&svc->srv_lock);
 392                 list_del_init(&rs->rs_exp_list);
 393                 ptlrpc_schedule_difficult_reply(rs);
 394                 spin_unlock(&svc->srv_lock);
 395         }
 396         spin_unlock_irqrestore(&exp->exp_lock, irqflags);
 397
 398         class_export_put(exp);
 399         RETURN(rc);
 400 }
 401
 402 static int mds_getstatus(struct ptlrpc_request *req)
 403 {
 404         struct mds_obd *mds = mds_req2mds(req);
 405         struct mds_body *body;
 406         int rc, size = sizeof(*body);
 407         ENTRY;
 408
 409         rc = lustre_pack_reply(req, 1, &size, NULL);
 410         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_GETSTATUS_PACK)) {
 411                 CERROR("mds: out of memory for message: size=%d\n", size);
 412                 req->rq_status = -ENOMEM;       /* superfluous? */
 413                 RETURN(-ENOMEM);
 414         }
 415
 416         body = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*body));
 417         memcpy(&body->fid1, &mds->mds_rootfid, sizeof(body->fid1));
 418
 419         /* the last_committed and last_xid fields are filled in for all
 420          * replies already - no need to do so here also.
 421          */
 422         RETURN(0);
 423 }
 424
 425 int mds_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
 426                      void *data, int flag)
 427 {
 428         int do_ast;
 429         ENTRY;
 430
 431         if (flag == LDLM_CB_CANCELING) {
 432                 /* Don't need to do anything here. */
 433                 RETURN(0);
 434         }
 435
 436         /* XXX layering violation!  -phil */
 437         l_lock(&lock->l_resource->lr_namespace->ns_lock);
 438         /* Get this: if mds_blocking_ast is racing with mds_intent_policy,
 439          * such that mds_blocking_ast is called just before l_i_p takes the
 440          * ns_lock, then by the time we get the lock, we might not be the
 441          * correct blocking function anymore.  So check, and return early, if
 442          * so. */
 443         if (lock->l_blocking_ast != mds_blocking_ast) {
 444                 l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 445                 RETURN(0);
 446         }
 447
 448         lock->l_flags |= LDLM_FL_CBPENDING;
 449         do_ast = (!lock->l_readers && !lock->l_writers);
 450         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 451
 452         if (do_ast) {
 453                 struct lustre_handle lockh;
 454                 int rc;
 455
 456                 LDLM_DEBUG(lock, "already unused, calling ldlm_cli_cancel");
 457                 ldlm_lock2handle(lock, &lockh);
 458                 rc = ldlm_cli_cancel(&lockh);
 459                 if (rc < 0)
 460                         CERROR("ldlm_cli_cancel: %d\n", rc);
 461         } else {
 462                 LDLM_DEBUG(lock, "Lock still has references, will be "
 463                            "cancelled later");
 464         }
 465         RETURN(0);
 466 }
 467
 468 int mds_get_md(struct obd_device *obd, struct inode *inode, void *md,
 469                int *size, int lock)
 470 {
 471         int rc = 0;
 472         int lmm_size;
 473
 474         if (lock)
 475                 down(&inode->i_sem);
 476         rc = fsfilt_get_md(obd, inode, md, *size);
 477
 478         if (rc < 0) {
 479                 CERROR("Error %d reading eadata for ino %lu\n",
 480                        rc, inode->i_ino);
 481         } else if (rc > 0) {
 482                 lmm_size = rc;
 483                 rc = mds_convert_lov_ea(obd, inode, md, lmm_size);
 484
 485                 if (rc == 0) {
 486                         *size = lmm_size;
 487                         rc = lmm_size;
 488                 } else if (rc > 0) {
 489                         *size = rc;
 490                 }
 491         }
 492         if (lock)
 493                 up(&inode->i_sem);
 494
 495         RETURN (rc);
 496 }
 497
 498
 499 /* Call with lock=1 if you want mds_pack_md to take the i_sem.
 500  * Call with lock=0 if the caller has already taken the i_sem. */
 501 int mds_pack_md(struct obd_device *obd, struct lustre_msg *msg, int offset,
 502                 struct mds_body *body, struct inode *inode, int lock)
 503 {
 504         struct mds_obd *mds = &obd->u.mds;
 505         void *lmm;
 506         int lmm_size;
 507         int rc;
 508         ENTRY;
 509
 510         lmm = lustre_msg_buf(msg, offset, 0);
 511         if (lmm == NULL) {
 512                 /* Some problem with getting eadata when I sized the reply
 513                  * buffer... */
 514                 CDEBUG(D_INFO, "no space reserved for inode %lu MD\n",
 515                        inode->i_ino);
 516                 RETURN(0);
 517         }
 518         lmm_size = msg->buflens[offset];
 519
 520         /* I don't really like this, but it is a sanity check on the client
 521          * MD request.  However, if the client doesn't know how much space
 522          * to reserve for the MD, it shouldn't be bad to have too much space.
 523          */
 524         if (lmm_size > mds->mds_max_mdsize) {
 525                 CWARN("Reading MD for inode %lu of %d bytes > max %d\n",
 526                        inode->i_ino, lmm_size, mds->mds_max_mdsize);
 527                 // RETURN(-EINVAL);
 528         }
 529
 530         rc = mds_get_md(obd, inode, lmm, &lmm_size, lock);
 531         if (rc > 0) {
 532                 if (S_ISDIR(inode->i_mode))
 533                         body->valid |= OBD_MD_FLDIREA;
 534                 else
 535                         body->valid |= OBD_MD_FLEASIZE;
 536                 body->eadatasize = lmm_size;
 537                 rc = 0;
 538         }
 539
 540         RETURN(rc);
 541 }
 542
 543 static int mds_getattr_internal(struct obd_device *obd, struct dentry *dentry,
 544                                 struct ptlrpc_request *req,
 545                                 struct mds_body *reqbody, int reply_off)
 546 {
 547         struct mds_body *body;
 548         struct inode *inode = dentry->d_inode;
 549         int rc = 0;
 550         ENTRY;
 551
 552         if (inode == NULL)
 553                 RETURN(-ENOENT);
 554
 555         body = lustre_msg_buf(req->rq_repmsg, reply_off, sizeof(*body));
 556         LASSERT(body != NULL);                 /* caller prepped reply */
 557
 558         mds_pack_inode2fid(&body->fid1, inode);
 559         mds_pack_inode2body(body, inode);
 560
 561         if ((S_ISREG(inode->i_mode) && (reqbody->valid & OBD_MD_FLEASIZE)) ||
 562             (S_ISDIR(inode->i_mode) && (reqbody->valid & OBD_MD_FLDIREA))) {
 563                 rc = mds_pack_md(obd, req->rq_repmsg, reply_off + 1, body,
 564                                  inode, 1);
 565
 566                 /* If we have LOV EA data, the OST holds size, atime, mtime */
 567                 if (!(body->valid & OBD_MD_FLEASIZE) &&
 568                     !(body->valid & OBD_MD_FLDIREA))
 569                         body->valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
 570                                         OBD_MD_FLATIME | OBD_MD_FLMTIME);
 571         } else if (S_ISLNK(inode->i_mode) &&
 572                    (reqbody->valid & OBD_MD_LINKNAME) != 0) {
 573                 char *symname = lustre_msg_buf(req->rq_repmsg, reply_off + 1,0);
 574                 int len;
 575
 576                 LASSERT (symname != NULL);       /* caller prepped reply */
 577                 len = req->rq_repmsg->buflens[reply_off + 1];
 578
 579                 rc = inode->i_op->readlink(dentry, symname, len);
 580                 if (rc < 0) {
 581                         CERROR("readlink failed: %d\n", rc);
 582                 } else if (rc != len - 1) {
 583                         CERROR ("Unexpected readlink rc %d: expecting %d\n",
 584                                 rc, len - 1);
 585                         rc = -EINVAL;
 586                 } else {
 587                         CDEBUG(D_INODE, "read symlink dest %s\n", symname);
 588                         body->valid |= OBD_MD_LINKNAME;
 589                         body->eadatasize = rc + 1;
 590                         symname[rc] = 0;        /* NULL terminate */
 591                         rc = 0;
 592                 }
 593         }
 594
 595         RETURN(rc);
 596 }
 597
 598 static int mds_getattr_pack_msg(struct ptlrpc_request *req, struct inode *inode,
 599                                 int offset)
 600 {
 601         struct mds_obd *mds = mds_req2mds(req);
 602         struct mds_body *body;
 603         int rc = 0, size[2] = {sizeof(*body)}, bufcount = 1;
 604         ENTRY;
 605
 606         body = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*body));
 607         LASSERT(body != NULL);                 /* checked by caller */
 608         LASSERT_REQSWABBED(req, offset);       /* swabbed by caller */
 609
 610         if ((S_ISREG(inode->i_mode) && (body->valid & OBD_MD_FLEASIZE)) ||
 611             (S_ISDIR(inode->i_mode) && (body->valid & OBD_MD_FLDIREA))) {
 612                 int ret;
 613                 down(&inode->i_sem);
 614                 ret = fsfilt_get_md(req->rq_export->exp_obd, inode, NULL, 0);
 615                 up(&inode->i_sem);
 616                 CDEBUG(D_INODE, "got %d bytes MD data for inode %lu\n",
 617                        rc, inode->i_ino);
 618                 if (ret < 0) {
 619                         if (ret != -ENODATA) {
 620                                 CERROR("error getting inode %lu MD: rc = %d\n",
 621                                        inode->i_ino, ret);
 622                                 /* should we return ret in req->rq_status? */
 623                         }
 624                         size[bufcount] = 0;
 625                 } else if (ret > mds->mds_max_mdsize) {
 626                         size[bufcount] = 0;
 627                         CERROR("MD size %d larger than maximum possible %u\n",
 628                                ret, mds->mds_max_mdsize);
 629                 } else {
 630                         size[bufcount] = ret;
 631                 }
 632                 bufcount++;
 633         } else if (S_ISLNK(inode->i_mode) && (body->valid & OBD_MD_LINKNAME)) {
 634                 if (inode->i_size + 1 != body->eadatasize)
 635                         CERROR("symlink size: %Lu, reply space: %d\n",
 636                                inode->i_size + 1, body->eadatasize);
 637                 size[bufcount] = min_t(int, inode->i_size+1, body->eadatasize);
 638                 bufcount++;
 639                 CDEBUG(D_INODE, "symlink size: %Lu, reply space: %d\n",
 640                        inode->i_size + 1, body->eadatasize);
 641         }
 642
 643         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) {
 644                 CERROR("failed MDS_GETATTR_PACK test\n");
 645                 req->rq_status = -ENOMEM;
 646                 GOTO(out, rc = -ENOMEM);
 647         }
 648
 649         rc = lustre_pack_reply(req, bufcount, size, NULL);
 650         if (rc) {
 651                 CERROR("lustre_pack_reply failed: rc %d\n", rc);
 652                 GOTO(out, req->rq_status = rc);
 653         }
 654
 655         EXIT;
 656  out:
 657         return(rc);
 658 }
 659
 660 static int mds_getattr_name(int offset, struct ptlrpc_request *req,
 661                             struct lustre_handle *child_lockh)
 662 {
 663         struct obd_device *obd = req->rq_export->exp_obd;
 664         struct ldlm_reply *rep = NULL;
 665         struct lvfs_run_ctxt saved;
 666         struct mds_body *body;
 667         struct dentry *dparent = NULL, *dchild = NULL;
 668         struct lvfs_ucred uc;
 669         struct lustre_handle parent_lockh;
 670         int namesize;
 671         int rc = 0, cleanup_phase = 0, resent_req = 0;
 672         char *name;
 673         ENTRY;
 674
 675         LASSERT(!strcmp(obd->obd_type->typ_name, "mds"));
 676
 677         /* Swab now, before anyone looks inside the request */
 678
 679         body = lustre_swab_reqbuf(req, offset, sizeof(*body),
 680                                   lustre_swab_mds_body);
 681         if (body == NULL) {
 682                 CERROR("Can't swab mds_body\n");
 683                 GOTO(cleanup, rc = -EFAULT);
 684         }
 685
 686         LASSERT_REQSWAB(req, offset + 1);
 687         name = lustre_msg_string(req->rq_reqmsg, offset + 1, 0);
 688         if (name == NULL) {
 689                 CERROR("Can't unpack name\n");
 690                 GOTO(cleanup, rc = -EFAULT);
 691         }
 692         namesize = req->rq_reqmsg->buflens[offset + 1];
 693
 694         LASSERT (offset == 0 || offset == 2);
 695         /* if requests were at offset 2, the getattr reply goes back at 1 */
 696         if (offset) {
 697                 rep = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*rep));
 698                 offset = 1;
 699         }
 700
 701 #if CRAY_PORTALS
 702         uc.luc_fsuid = req->rq_uid;
 703 #else
 704         uc.luc_fsuid = body->fsuid;
 705 #endif
 706         uc.luc_fsgid = body->fsgid;
 707         uc.luc_cap = body->capability;
 708         uc.luc_suppgid1 = body->suppgid;
 709         uc.luc_suppgid2 = -1;
 710         push_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
 711         cleanup_phase = 1; /* kernel context */
 712         intent_set_disposition(rep, DISP_LOOKUP_EXECD);
 713
 714         /* FIXME: handle raw lookup */
 715 #if 0
 716         if (body->valid == OBD_MD_FLID) {
 717                 struct mds_body *mds_reply;
 718                 int size = sizeof(*mds_reply);
 719                 ino_t inum;
 720                 // The user requested ONLY the inode number, so do a raw lookup
 721                 rc = lustre_pack_reply(req, 1, &size, NULL);
 722                 if (rc) {
 723                         CERROR("out of memory\n");
 724                         GOTO(cleanup, rc);
 725                 }
 726
 727                 rc = dir->i_op->lookup_raw(dir, name, namesize - 1, &inum);
 728
 729                 mds_reply = lustre_msg_buf(req->rq_repmsg, offset,
 730                                            sizeof(*mds_reply));
 731                 mds_reply->fid1.id = inum;
 732                 mds_reply->valid = OBD_MD_FLID;
 733                 GOTO(cleanup, rc);
 734         }
 735 #endif
 736
 737         if (child_lockh->cookie != 0) {
 738                 LASSERT(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT);
 739                 resent_req = 1;
 740         }
 741
 742         if (resent_req == 0) {
 743                 rc = mds_get_parent_child_locked(obd, &obd->u.mds, &body->fid1,
 744                                                  &parent_lockh, &dparent,
 745                                                  LCK_PR, name, namesize,
 746                                                  child_lockh, &dchild, LCK_PR);
 747                 if (rc)
 748                         GOTO(cleanup, rc);
 749         } else {
 750                 struct ldlm_lock *granted_lock;
 751                 struct ll_fid child_fid;
 752                 struct ldlm_resource *res;
 753                 DEBUG_REQ(D_DLMTRACE, req, "resent, not enqueuing new locks");
 754                 granted_lock = ldlm_handle2lock(child_lockh);
 755                 LASSERTF(granted_lock != NULL, LPU64"/%u lockh "LPX64"\n",
 756                          body->fid1.id, body->fid1.generation,
 757                          child_lockh->cookie);
 758
 759
 760                 res = granted_lock->l_resource;
 761                 child_fid.id = res->lr_name.name[0];
 762                 child_fid.generation = res->lr_name.name[1];
 763                 dchild = mds_fid2dentry(&obd->u.mds, &child_fid, NULL);
 764                 LASSERT(!IS_ERR(dchild));
 765                 LDLM_LOCK_PUT(granted_lock);
 766         }
 767
 768         cleanup_phase = 2; /* dchild, dparent, locks */
 769
 770         if (dchild->d_inode == NULL) {
 771                 intent_set_disposition(rep, DISP_LOOKUP_NEG);
 772                 /* in the intent case, the policy clears this error:
 773                    the disposition is enough */
 774                 GOTO(cleanup, rc = -ENOENT);
 775         } else {
 776                 intent_set_disposition(rep, DISP_LOOKUP_POS);
 777         }
 778
 779         if (req->rq_repmsg == NULL) {
 780                 rc = mds_getattr_pack_msg(req, dchild->d_inode, offset);
 781                 if (rc != 0) {
 782                         CERROR ("mds_getattr_pack_msg: %d\n", rc);
 783                         GOTO (cleanup, rc);
 784                 }
 785         }
 786
 787         rc = mds_getattr_internal(obd, dchild, req, body, offset);
 788         GOTO(cleanup, rc); /* returns the lock to the client */
 789
 790  cleanup:
 791         switch (cleanup_phase) {
 792         case 2:
 793                 if (resent_req == 0) {
 794                         if (rc && dchild->d_inode)
 795                                 ldlm_lock_decref(child_lockh, LCK_PR);
 796                         ldlm_lock_decref(&parent_lockh, LCK_PR);
 797                         l_dput(dparent);
 798                 }
 799                 l_dput(dchild);
 800         case 1:
 801                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
 802         default: ;
 803         }
 804         return rc;
 805 }
 806
 807 static int mds_getattr(int offset, struct ptlrpc_request *req)
 808 {
 809         struct mds_obd *mds = mds_req2mds(req);
 810         struct obd_device *obd = req->rq_export->exp_obd;
 811         struct lvfs_run_ctxt saved;
 812         struct dentry *de;
 813         struct mds_body *body;
 814         struct lvfs_ucred uc;
 815         int rc = 0;
 816         ENTRY;
 817
 818         body = lustre_swab_reqbuf(req, offset, sizeof(*body),
 819                                   lustre_swab_mds_body);
 820         if (body == NULL) {
 821                 CERROR("Can't unpack body\n");
 822                 RETURN(-EFAULT);
 823         }
 824
 825 #if CRAY_PORTALS
 826         uc.luc_fsuid = req->rq_uid;
 827 #else
 828         uc.luc_fsuid = body->fsuid;
 829 #endif
 830         uc.luc_fsgid = body->fsgid;
 831         uc.luc_cap = body->capability;
 832         push_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
 833         de = mds_fid2dentry(mds, &body->fid1, NULL);
 834         if (IS_ERR(de)) {
 835                 rc = req->rq_status = PTR_ERR(de);
 836                 GOTO(out_pop, rc);
 837         }
 838
 839         rc = mds_getattr_pack_msg(req, de->d_inode, offset);
 840         if (rc != 0) {
 841                 CERROR("mds_getattr_pack_msg: %d\n", rc);
 842                 GOTO(out_pop, rc);
 843         }
 844
 845         req->rq_status = mds_getattr_internal(obd, de, req, body, 0);
 846
 847         l_dput(de);
 848         GOTO(out_pop, rc);
 849 out_pop:
 850         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
 851         return rc;
 852 }
 853
 854
 855 static int mds_obd_statfs(struct obd_device *obd, struct obd_statfs *osfs,
 856                           unsigned long max_age)
 857 {
 858         int rc;
 859
 860         spin_lock(&obd->obd_osfs_lock);
 861         rc = fsfilt_statfs(obd, obd->u.mds.mds_sb, max_age);
 862         if (rc == 0)
 863                 memcpy(osfs, &obd->obd_osfs, sizeof(*osfs));
 864         spin_unlock(&obd->obd_osfs_lock);
 865
 866         return rc;
 867 }
 868
 869 static int mds_statfs(struct ptlrpc_request *req)
 870 {
 871         struct obd_device *obd = req->rq_export->exp_obd;
 872         int rc, size = sizeof(struct obd_statfs);
 873         ENTRY;
 874
 875         /* This will trigger a watchdog timeout */
 876         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_STATFS_LCW_SLEEP,
 877                          (MDS_SERVICE_WATCHDOG_TIMEOUT / 1000) + 1);
 878
 879         rc = lustre_pack_reply(req, 1, &size, NULL);
 880         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_STATFS_PACK)) {
 881                 CERROR("mds: statfs lustre_pack_reply failed: rc = %d\n", rc);
 882                 GOTO(out, rc);
 883         }
 884
 885         /* We call this so that we can cache a bit - 1 jiffie worth */
 886         rc = mds_obd_statfs(obd, lustre_msg_buf(req->rq_repmsg, 0, size),
 887                             jiffies - HZ);
 888         if (rc) {
 889                 CERROR("mds_obd_statfs failed: rc %d\n", rc);
 890                 GOTO(out, rc);
 891         }
 892
 893         EXIT;
 894 out:
 895         req->rq_status = rc;
 896         return 0;
 897 }
 898
 899 static int mds_sync(struct ptlrpc_request *req)
 900 {
 901         struct obd_device *obd = req->rq_export->exp_obd;
 902         struct mds_obd *mds = &obd->u.mds;
 903         struct mds_body *body;
 904         int rc, size = sizeof(*body);
 905         ENTRY;
 906
 907         body = lustre_swab_reqbuf(req, 0, sizeof(*body), lustre_swab_mds_body);
 908         if (body == NULL)
 909                 GOTO(out, rc = -EFAULT);
 910
 911         rc = lustre_pack_reply(req, 1, &size, NULL);
 912         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_SYNC_PACK)) {
 913                 CERROR("fsync lustre_pack_reply failed: rc = %d\n", rc);
 914                 GOTO(out, rc);
 915         }
 916
 917         if (body->fid1.id == 0) {
 918                 /* a fid of zero is taken to mean "sync whole filesystem" */
 919                 rc = fsfilt_sync(obd, mds->mds_sb);
 920                 GOTO(out, rc);
 921         } else {
 922                 struct dentry *de;
 923
 924                 de = mds_fid2dentry(mds, &body->fid1, NULL);
 925                 if (IS_ERR(de))
 926                         GOTO(out, rc = PTR_ERR(de));
 927
 928                 /* The file parameter isn't used for anything */
 929                 if (de->d_inode->i_fop && de->d_inode->i_fop->fsync)
 930                         rc = de->d_inode->i_fop->fsync(NULL, de, 1);
 931                 if (rc == 0) {
 932                         body = lustre_msg_buf(req->rq_repmsg, 0, sizeof(*body));
 933                         mds_pack_inode2fid(&body->fid1, de->d_inode);
 934                         mds_pack_inode2body(body, de->d_inode);
 935                 }
 936
 937                 l_dput(de);
 938                 GOTO(out, rc);
 939         }
 940 out:
 941         req->rq_status = rc;
 942         return 0;
 943 }
 944
 945 /* mds_readpage does not take a DLM lock on the inode, because the client must
 946  * already have a PR lock.
 947  *
 948  * If we were to take another one here, a deadlock will result, if another
 949  * thread is already waiting for a PW lock. */
 950 static int mds_readpage(struct ptlrpc_request *req)
 951 {
 952         struct obd_device *obd = req->rq_export->exp_obd;
 953         struct vfsmount *mnt;
 954         struct dentry *de;
 955         struct file *file;
 956         struct mds_body *body, *repbody;
 957         struct lvfs_run_ctxt saved;
 958         int rc, size = sizeof(*repbody);
 959         struct lvfs_ucred uc;
 960         ENTRY;
 961
 962         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_READPAGE_PACK))
 963                 RETURN(-ENOMEM);
 964
 965         rc = lustre_pack_reply(req, 1, &size, NULL);
 966         if (rc) {
 967                 CERROR("mds: out of memory while packing readpage reply\n");
 968                 RETURN(-ENOMEM);
 969         }
 970
 971         body = lustre_swab_reqbuf(req, 0, sizeof(*body), lustre_swab_mds_body);
 972         if (body == NULL)
 973                 GOTO (out, rc = -EFAULT);
 974
 975 #if CRAY_PORTALS
 976         uc.luc_fsuid = req->rq_uid;
 977 #else
 978         uc.luc_fsuid = body->fsuid;
 979 #endif
 980         uc.luc_fsgid = body->fsgid;
 981         uc.luc_cap = body->capability;
 982         push_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
 983         de = mds_fid2dentry(&obd->u.mds, &body->fid1, &mnt);
 984         if (IS_ERR(de))
 985                 GOTO(out_pop, rc = PTR_ERR(de));
 986
 987         CDEBUG(D_INODE, "ino %lu\n", de->d_inode->i_ino);
 988
 989         file = dentry_open(de, mnt, O_RDONLY | O_LARGEFILE);
 990         /* note: in case of an error, dentry_open puts dentry */
 991         if (IS_ERR(file))
 992                 GOTO(out_pop, rc = PTR_ERR(file));
 993
 994         /* body->size is actually the offset -eeb */
 995         if ((body->size & (de->d_inode->i_blksize - 1)) != 0) {
 996                 CERROR("offset "LPU64" not on a block boundary of %lu\n",
 997                        body->size, de->d_inode->i_blksize);
 998                 GOTO(out_file, rc = -EFAULT);
 999         }
1000
1001         /* body->nlink is actually the #bytes to read -eeb */
1002         if (body->nlink & (de->d_inode->i_blksize - 1)) {
1003                 CERROR("size %u is not multiple of blocksize %lu\n",
1004                        body->nlink, de->d_inode->i_blksize);
1005                 GOTO(out_file, rc = -EFAULT);
1006         }
1007
1008         repbody = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*repbody));
1009         repbody->size = file->f_dentry->d_inode->i_size;
1010         repbody->valid = OBD_MD_FLSIZE;
1011
1012         /* to make this asynchronous make sure that the handling function
1013            doesn't send a reply when this function completes. Instead a
1014            callback function would send the reply */
1015         /* body->size is actually the offset -eeb */
1016         rc = mds_sendpage(req, file, body->size, body->nlink);
1017
1018 out_file:
1019         filp_close(file, 0);
1020 out_pop:
1021         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc);
1022 out:
1023         req->rq_status = rc;
1024         RETURN(0);
1025 }
1026
1027 int mds_reint(struct ptlrpc_request *req, int offset,
1028               struct lustre_handle *lockh)
1029 {
1030         struct mds_update_record *rec; /* 116 bytes on the stack?  no sir! */
1031         int rc;
1032
1033         OBD_ALLOC(rec, sizeof(*rec));
1034         if (rec == NULL)
1035                 RETURN(-ENOMEM);
1036
1037         rc = mds_update_unpack(req, offset, rec);
1038         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNPACK)) {
1039                 CERROR("invalid record\n");
1040                 GOTO(out, req->rq_status = -EINVAL);
1041         }
1042         /* rc will be used to interrupt a for loop over multiple records */
1043         rc = mds_reint_rec(rec, offset, req, lockh);
1044  out:
1045         OBD_FREE(rec, sizeof(*rec));
1046         return rc;
1047 }
1048
1049 static int mds_filter_recovery_request(struct ptlrpc_request *req,
1050                                        struct obd_device *obd, int *process)
1051 {
1052         switch (req->rq_reqmsg->opc) {
1053         case MDS_CONNECT: /* This will never get here, but for completeness. */
1054         case OST_CONNECT: /* This will never get here, but for completeness. */
1055         case MDS_DISCONNECT:
1056         case OST_DISCONNECT:
1057                *process = 1;
1058                RETURN(0);
1059
1060         case MDS_CLOSE:
1061         case MDS_SYNC: /* used in unmounting */
1062         case OBD_PING:
1063         case MDS_REINT:
1064         case LDLM_ENQUEUE:
1065                 *process = target_queue_recovery_request(req, obd);
1066                 RETURN(0);
1067
1068         default:
1069                 DEBUG_REQ(D_ERROR, req, "not permitted during recovery");
1070                 *process = 0;
1071                 /* XXX what should we set rq_status to here? */
1072                 req->rq_status = -EAGAIN;
1073                 RETURN(ptlrpc_error(req));
1074         }
1075 }
1076
1077 static char *reint_names[] = {
1078         [REINT_SETATTR] "setattr",
1079         [REINT_CREATE]  "create",
1080         [REINT_LINK]    "link",
1081         [REINT_UNLINK]  "unlink",
1082         [REINT_RENAME]  "rename",
1083         [REINT_OPEN]    "open",
1084 };
1085
1086 static int mds_set_info(struct obd_export *exp, struct ptlrpc_request *req)
1087 {
1088         char *key;
1089         __u32 *val;
1090         int keylen, rc = 0;
1091         ENTRY;
1092
1093         key = lustre_msg_buf(req->rq_reqmsg, 0, 1);
1094         if (key == NULL) {
1095                 DEBUG_REQ(D_HA, req, "no set_info key");
1096                 RETURN(-EFAULT);
1097         }
1098         keylen = req->rq_reqmsg->buflens[0];
1099
1100         val = lustre_msg_buf(req->rq_reqmsg, 1, sizeof(*val));
1101         if (val == NULL) {
1102                 DEBUG_REQ(D_HA, req, "no set_info val");
1103                 RETURN(-EFAULT);
1104         }
1105
1106         rc = lustre_pack_reply(req, 0, NULL, NULL);
1107         if (rc)
1108                 RETURN(rc);
1109         req->rq_repmsg->status = 0;
1110
1111         if (keylen < strlen("read-only") ||
1112             memcmp(key, "read-only", keylen) != 0)
1113                 RETURN(-EINVAL);
1114
1115         if (*val)
1116                 exp->exp_connect_flags |= OBD_CONNECT_RDONLY;
1117         else
1118                 exp->exp_connect_flags &= ~OBD_CONNECT_RDONLY;
1119
1120         RETURN(0);
1121 }
1122
1123
1124 int mds_handle(struct ptlrpc_request *req)
1125 {
1126         int should_process, fail = OBD_FAIL_MDS_ALL_REPLY_NET;
1127         int rc = 0;
1128         struct mds_obd *mds = NULL; /* quell gcc overwarning */
1129         struct obd_device *obd = NULL;
1130         ENTRY;
1131
1132         OBD_FAIL_RETURN(OBD_FAIL_MDS_ALL_REQUEST_NET | OBD_FAIL_ONCE, 0);
1133
1134         LASSERT(current->journal_info == NULL);
1135         /* XXX identical to OST */
1136         if (req->rq_reqmsg->opc != MDS_CONNECT) {
1137                 struct mds_export_data *med;
1138                 int recovering, abort_recovery;
1139
1140                 if (req->rq_export == NULL) {
1141                         CERROR("lustre_mds: operation %d on unconnected MDS\n",
1142                                req->rq_reqmsg->opc);
1143                         req->rq_status = -ENOTCONN;
1144                         GOTO(out, rc = -ENOTCONN);
1145                 }
1146
1147                 med = &req->rq_export->exp_mds_data;
1148                 obd = req->rq_export->exp_obd;
1149                 mds = &obd->u.mds;
1150
1151                 /* sanity check: if the xid matches, the request must
1152                  * be marked as a resent or replayed */
1153                 if (req->rq_xid == med->med_mcd->mcd_last_xid)
1154                         LASSERTF(lustre_msg_get_flags(req->rq_reqmsg) &
1155                                  (MSG_RESENT | MSG_REPLAY),
1156                                  "rq_xid "LPU64" matches last_xid, "
1157                                  "expected RESENT flag\n",
1158                                  req->rq_xid);
1159                 /* else: note the opposite is not always true; a
1160                  * RESENT req after a failover will usually not match
1161                  * the last_xid, since it was likely never
1162                  * committed. A REPLAYed request will almost never
1163                  * match the last xid, however it could for a
1164                  * committed, but still retained, open. */
1165
1166                 /* Check for aborted recovery. */
1167                 spin_lock_bh(&obd->obd_processing_task_lock);
1168                 abort_recovery = obd->obd_abort_recovery;
1169                 recovering = obd->obd_recovering;
1170                 spin_unlock_bh(&obd->obd_processing_task_lock);
1171                 if (abort_recovery) {
1172                         target_abort_recovery(obd);
1173                 } else if (recovering) {
1174                         rc = mds_filter_recovery_request(req, obd,
1175                                                          &should_process);
1176                         if (rc || !should_process)
1177                                 RETURN(rc);
1178                 }
1179         }
1180
1181         switch (req->rq_reqmsg->opc) {
1182         case MDS_CONNECT:
1183                 DEBUG_REQ(D_INODE, req, "connect");
1184                 OBD_FAIL_RETURN(OBD_FAIL_MDS_CONNECT_NET, 0);
1185                 rc = target_handle_connect(req, mds_handle);
1186                 if (!rc) {
1187                         /* Now that we have an export, set mds. */
1188                         obd = req->rq_export->exp_obd;
1189                         mds = mds_req2mds(req);
1190                 }
1191                 break;
1192
1193         case MDS_DISCONNECT:
1194                 DEBUG_REQ(D_INODE, req, "disconnect");
1195                 OBD_FAIL_RETURN(OBD_FAIL_MDS_DISCONNECT_NET, 0);
1196                 rc = target_handle_disconnect(req);
1197                 req->rq_status = rc;            /* superfluous? */
1198                 break;
1199
1200         case MDS_GETSTATUS:
1201                 DEBUG_REQ(D_INODE, req, "getstatus");
1202                 OBD_FAIL_RETURN(OBD_FAIL_MDS_GETSTATUS_NET, 0);
1203                 rc = mds_getstatus(req);
1204                 break;
1205
1206         case MDS_GETATTR:
1207                 DEBUG_REQ(D_INODE, req, "getattr");
1208                 OBD_FAIL_RETURN(OBD_FAIL_MDS_GETATTR_NET, 0);
1209                 rc = mds_getattr(0, req);
1210                 break;
1211
1212         case MDS_GETATTR_NAME: {
1213                 struct lustre_handle lockh;
1214                 DEBUG_REQ(D_INODE, req, "getattr_name");
1215                 OBD_FAIL_RETURN(OBD_FAIL_MDS_GETATTR_NAME_NET, 0);
1216
1217                 /* If this request gets a reconstructed reply, we won't be
1218                  * acquiring any new locks in mds_getattr_name, so we don't
1219                  * want to cancel.
1220                  */
1221                 lockh.cookie = 0;
1222                 rc = mds_getattr_name(0, req, &lockh);
1223                 /* this non-intent call (from an ioctl) is special */
1224                 req->rq_status = rc;
1225                 if (rc == 0 && lockh.cookie)
1226                         ldlm_lock_decref(&lockh, LCK_PR);
1227                 break;
1228         }
1229         case MDS_STATFS:
1230                 DEBUG_REQ(D_INODE, req, "statfs");
1231                 OBD_FAIL_RETURN(OBD_FAIL_MDS_STATFS_NET, 0);
1232                 rc = mds_statfs(req);
1233                 break;
1234
1235         case MDS_READPAGE:
1236                 DEBUG_REQ(D_INODE, req, "readpage");
1237                 OBD_FAIL_RETURN(OBD_FAIL_MDS_READPAGE_NET, 0);
1238                 rc = mds_readpage(req);
1239
1240                 if (OBD_FAIL_CHECK_ONCE(OBD_FAIL_MDS_SENDPAGE)) {
1241                         RETURN(0);
1242                 }
1243
1244                 break;
1245
1246         case MDS_REINT: {
1247                 __u32 *opcp = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*opcp));
1248                 __u32  opc;
1249                 int size[3] = {sizeof(struct mds_body), mds->mds_max_mdsize,
1250                                mds->mds_max_cookiesize};
1251                 int bufcount;
1252
1253                 /* NB only peek inside req now; mds_reint() will swab it */
1254                 if (opcp == NULL) {
1255                         CERROR ("Can't inspect opcode\n");
1256                         rc = -EINVAL;
1257                         break;
1258                 }
1259                 opc = *opcp;
1260                 if (lustre_msg_swabbed (req->rq_reqmsg))
1261                         __swab32s(&opc);
1262
1263                 DEBUG_REQ(D_INODE, req, "reint %d (%s)", opc,
1264                           (opc < sizeof(reint_names) / sizeof(reint_names[0]) ||
1265                            reint_names[opc] == NULL) ? reint_names[opc] :
1266                                                        "unknown opcode");
1267
1268                 OBD_FAIL_RETURN(OBD_FAIL_MDS_REINT_NET, 0);
1269
1270                 if (opc == REINT_UNLINK || opc == REINT_RENAME)
1271                         bufcount = 3;
1272                 else if (opc == REINT_OPEN)
1273                         bufcount = 2;
1274                 else
1275                         bufcount = 1;
1276
1277                 rc = lustre_pack_reply(req, bufcount, size, NULL);
1278                 if (rc)
1279                         break;
1280
1281                 rc = mds_reint(req, 0, NULL);
1282                 fail = OBD_FAIL_MDS_REINT_NET_REP;
1283                 break;
1284         }
1285
1286         case MDS_CLOSE:
1287                 DEBUG_REQ(D_INODE, req, "close");
1288                 OBD_FAIL_RETURN(OBD_FAIL_MDS_CLOSE_NET, 0);
1289                 rc = mds_close(req);
1290                 break;
1291
1292         case MDS_DONE_WRITING:
1293                 DEBUG_REQ(D_INODE, req, "done_writing");
1294                 OBD_FAIL_RETURN(OBD_FAIL_MDS_DONE_WRITING_NET, 0);
1295                 rc = mds_done_writing(req);
1296                 break;
1297
1298         case MDS_PIN:
1299                 DEBUG_REQ(D_INODE, req, "pin");
1300                 OBD_FAIL_RETURN(OBD_FAIL_MDS_PIN_NET, 0);
1301                 rc = mds_pin(req);
1302                 break;
1303
1304         case MDS_SYNC:
1305                 DEBUG_REQ(D_INODE, req, "sync");
1306                 OBD_FAIL_RETURN(OBD_FAIL_MDS_SYNC_NET, 0);
1307                 rc = mds_sync(req);
1308                 break;
1309
1310         case MDS_SET_INFO:
1311                 DEBUG_REQ(D_INODE, req, "set_info");
1312                 rc = mds_set_info(req->rq_export, req);
1313                 break;
1314
1315         case MDS_QUOTACHECK:
1316                 DEBUG_REQ(D_INODE, req, "quotacheck");
1317                 OBD_FAIL_RETURN(OBD_FAIL_MDS_QUOTACHECK_NET, 0);
1318                 rc = mds_quotacheck(req);
1319                 break;
1320
1321         case MDS_QUOTACTL:
1322                 DEBUG_REQ(D_INODE, req, "quotactl");
1323                 OBD_FAIL_RETURN(OBD_FAIL_MDS_QUOTACTL_NET, 0);
1324                 rc = mds_quotactl(req);
1325                 break;
1326
1327         case OBD_PING:
1328                 DEBUG_REQ(D_INODE, req, "ping");
1329                 rc = target_handle_ping(req);
1330                 break;
1331
1332         case OBD_LOG_CANCEL:
1333                 CDEBUG(D_INODE, "log cancel\n");
1334                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOG_CANCEL_NET, 0);
1335                 rc = -ENOTSUPP; /* la la la */
1336                 break;
1337
1338         case LDLM_ENQUEUE:
1339                 DEBUG_REQ(D_INODE, req, "enqueue");
1340                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_ENQUEUE, 0);
1341                 rc = ldlm_handle_enqueue(req, ldlm_server_completion_ast,
1342                                          ldlm_server_blocking_ast, NULL);
1343                 fail = OBD_FAIL_LDLM_REPLY;
1344                 break;
1345         case LDLM_CONVERT:
1346                 DEBUG_REQ(D_INODE, req, "convert");
1347                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CONVERT, 0);
1348                 rc = ldlm_handle_convert(req);
1349                 break;
1350         case LDLM_BL_CALLBACK:
1351         case LDLM_CP_CALLBACK:
1352                 DEBUG_REQ(D_INODE, req, "callback");
1353                 CERROR("callbacks should not happen on MDS\n");
1354                 LBUG();
1355                 OBD_FAIL_RETURN(OBD_FAIL_LDLM_BL_CALLBACK, 0);
1356                 break;
1357         case LLOG_ORIGIN_HANDLE_CREATE:
1358                 DEBUG_REQ(D_INODE, req, "llog_init");
1359                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1360                 rc = llog_origin_handle_create(req);
1361                 break;
1362         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
1363                 DEBUG_REQ(D_INODE, req, "llog next block");
1364                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1365                 rc = llog_origin_handle_next_block(req);
1366                 break;
1367         case LLOG_ORIGIN_HANDLE_READ_HEADER:
1368                 DEBUG_REQ(D_INODE, req, "llog read header");
1369                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1370                 rc = llog_origin_handle_read_header(req);
1371                 break;
1372         case LLOG_ORIGIN_HANDLE_CLOSE:
1373                 DEBUG_REQ(D_INODE, req, "llog close");
1374                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1375                 rc = llog_origin_handle_close(req);
1376                 break;
1377         case LLOG_CATINFO:
1378                 DEBUG_REQ(D_INODE, req, "llog catinfo");
1379                 OBD_FAIL_RETURN(OBD_FAIL_OBD_LOGD_NET, 0);
1380                 rc = llog_catinfo(req);
1381                 break;
1382         default:
1383                 req->rq_status = -ENOTSUPP;
1384                 rc = ptlrpc_error(req);
1385                 RETURN(rc);
1386         }
1387
1388         LASSERT(current->journal_info == NULL);
1389
1390         /* If we're DISCONNECTing, the mds_export_data is already freed */
1391         if (!rc && req->rq_reqmsg->opc != MDS_DISCONNECT) {
1392                 struct mds_export_data *med = &req->rq_export->exp_mds_data;
1393                 req->rq_repmsg->last_xid =
1394                         le64_to_cpu(med->med_mcd->mcd_last_xid);
1395
1396                 target_committed_to_req(req);
1397         }
1398
1399         EXIT;
1400  out:
1401
1402         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_LAST_REPLAY) {
1403                 if (obd && obd->obd_recovering) {
1404                         DEBUG_REQ(D_HA, req, "LAST_REPLAY, queuing reply");
1405                         return target_queue_final_reply(req, rc);
1406                 }
1407                 /* Lost a race with recovery; let the error path DTRT. */
1408                 rc = req->rq_status = -ENOTCONN;
1409         }
1410
1411         target_send_reply(req, rc, fail);
1412         return 0;
1413 }
1414
1415 /* Update the server data on disk.  This stores the new mount_count and
1416  * also the last_rcvd value to disk.  If we don't have a clean shutdown,
1417  * then the server last_rcvd value may be less than that of the clients.
1418  * This will alert us that we may need to do client recovery.
1419  *
1420  * Also assumes for mds_last_transno that we are not modifying it (no locking).
1421  */
1422 int mds_update_server_data(struct obd_device *obd, int force_sync)
1423 {
1424         struct mds_obd *mds = &obd->u.mds;
1425         struct mds_server_data *msd = mds->mds_server_data;
1426         struct file *filp = mds->mds_rcvd_filp;
1427         struct lvfs_run_ctxt saved;
1428         loff_t off = 0;
1429         int rc;
1430         ENTRY;
1431
1432         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1433         msd->msd_last_transno = cpu_to_le64(mds->mds_last_transno);
1434
1435         CDEBUG(D_SUPER, "MDS mount_count is "LPU64", last_transno is "LPU64"\n",
1436                mds->mds_mount_count, mds->mds_last_transno);
1437         rc = fsfilt_write_record(obd, filp, msd, sizeof(*msd), &off,force_sync);
1438         if (rc)
1439                 CERROR("error writing MDS server data: rc = %d\n", rc);
1440         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1441
1442         RETURN(rc);
1443 }
1444
1445
1446 /* mount the file system (secretly) */
1447 static int mds_setup(struct obd_device *obd, obd_count len, void *buf)
1448 {
1449         struct lprocfs_static_vars lvars;
1450         struct lustre_cfg* lcfg = buf;
1451         char *options = NULL;
1452         struct mds_obd *mds = &obd->u.mds;
1453         struct vfsmount *mnt;
1454         char ns_name[48];
1455         unsigned long page;
1456         int rc = 0;
1457         ENTRY;
1458
1459         /* setup 1:/dev/loop/0 2:ext3 3:mdsA 4:errors=remount-ro,iopen_nopriv*/
1460
1461         if (lcfg->lcfg_bufcount < 3)
1462                 RETURN(rc = -EINVAL);
1463
1464         if (LUSTRE_CFG_BUFLEN(lcfg, 1) == 0 || LUSTRE_CFG_BUFLEN(lcfg, 2) == 0)
1465                 RETURN(rc = -EINVAL);
1466
1467         obd->obd_fsops = fsfilt_get_ops(lustre_cfg_string(lcfg, 2));
1468         if (IS_ERR(obd->obd_fsops))
1469                 RETURN(rc = PTR_ERR(obd->obd_fsops));
1470
1471         page = __get_free_page(GFP_KERNEL);
1472         if (!page)
1473                 RETURN(-ENOMEM);
1474
1475         options = (char *)page;
1476         memset(options, 0, PAGE_SIZE);
1477
1478         /* here we use "iopen_nopriv" hardcoded, because it affects MDS utility
1479          * and the rest of options are passed by mount options. Probably this
1480          * should be moved to somewhere else like startup scripts or lconf. */
1481         sprintf(options, "iopen_nopriv");
1482
1483         if (LUSTRE_CFG_BUFLEN(lcfg, 4) > 0 && lustre_cfg_buf(lcfg, 4))
1484                 sprintf(options + strlen(options), ",%s",
1485                         lustre_cfg_string(lcfg, 4));
1486
1487         //FIXME mount was already done in lustre_fill_super,
1488         //we just need to access it
1489         mnt = do_kern_mount(lustre_cfg_string(lcfg, 2), 0,
1490                             lustre_cfg_string(lcfg, 1), (void *)options);
1491         free_page(page);
1492         if (IS_ERR(mnt)) {
1493                 rc = PTR_ERR(mnt);
1494                 CERROR("do_kern_mount failed: rc = %d\n", rc);
1495                 GOTO(err_ops, rc);
1496         }
1497
1498         CDEBUG(D_SUPER, "%s: mnt = %p\n", lustre_cfg_string(lcfg, 1), mnt);
1499
1500         LASSERT(!lvfs_check_rdonly(lvfs_sbdev(mnt->mnt_sb)));
1501
1502         sema_init(&mds->mds_orphan_recovery_sem, 1);
1503         sema_init(&mds->mds_epoch_sem, 1);
1504         spin_lock_init(&mds->mds_transno_lock);
1505         mds->mds_max_mdsize = sizeof(struct lov_mds_md);
1506         mds->mds_max_cookiesize = sizeof(struct llog_cookie);
1507
1508         sprintf(ns_name, "mds-%s", obd->obd_uuid.uuid);
1509         obd->obd_namespace = ldlm_namespace_new(ns_name, LDLM_NAMESPACE_SERVER);
1510         if (obd->obd_namespace == NULL) {
1511                 mds_cleanup(obd);
1512                 GOTO(err_put, rc = -ENOMEM);
1513         }
1514         ldlm_register_intent(obd->obd_namespace, mds_intent_policy);
1515
1516         rc = mds_fs_setup(obd, mnt);
1517         if (rc) {
1518                 CERROR("%s: MDS filesystem method init failed: rc = %d\n",
1519                        obd->obd_name, rc);
1520                 GOTO(err_ns, rc);
1521         }
1522
1523         rc = llog_start_commit_thread();
1524         if (rc < 0)
1525                 GOTO(err_fs, rc);
1526
1527         if (lcfg->lcfg_bufcount >= 4 && LUSTRE_CFG_BUFLEN(lcfg, 3) > 0) {
1528                 class_uuid_t uuid;
1529
1530                 generate_random_uuid(uuid);
1531                 class_uuid_unparse(uuid, &mds->mds_lov_uuid);
1532
1533                 OBD_ALLOC(mds->mds_profile, LUSTRE_CFG_BUFLEN(lcfg, 3));
1534                 if (mds->mds_profile == NULL)
1535                         GOTO(err_fs, rc = -ENOMEM);
1536
1537                 strncpy(mds->mds_profile, lustre_cfg_string(lcfg, 3),
1538                         LUSTRE_CFG_BUFLEN(lcfg, 3));
1539
1540         }
1541
1542         ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL,
1543                            "mds_ldlm_client", &obd->obd_ldlm_client);
1544         obd->obd_replayable = 1;
1545
1546         mds_quota_setup(mds);
1547
1548         rc = mds_postsetup(obd);
1549         if (rc)
1550                 GOTO(err_fs, rc);
1551
1552         lprocfs_init_vars(mds, &lvars);
1553         lprocfs_obd_setup(obd, lvars.obd_vars);
1554
1555         if (obd->obd_recovering) {
1556                 LCONSOLE_WARN("MDT %s now serving %s, but will be in recovery "
1557                               "until %d %s reconnect, or if no clients "
1558                               "reconnect for %d:%.02d; during that time new "
1559                               "clients will not be allowed to connect. "
1560                               "Recovery progress can be monitored by watching "
1561                               "/proc/fs/lustre/mds/%s/recovery_status.\n",
1562                               obd->obd_name,
1563                               lustre_cfg_string(lcfg, 1),
1564                               obd->obd_recoverable_clients,
1565                               (obd->obd_recoverable_clients == 1)
1566                               ? "client" : "clients",
1567                               (int)(OBD_RECOVERY_TIMEOUT / HZ) / 60,
1568                               (int)(OBD_RECOVERY_TIMEOUT / HZ) % 60,
1569                               obd->obd_name);
1570         } else {
1571                 LCONSOLE_INFO("MDT %s now serving %s with recovery %s.\n",
1572                               obd->obd_name,
1573                               lustre_cfg_string(lcfg, 1),
1574                               obd->obd_replayable ? "enabled" : "disabled");
1575         }
1576
1577         ldlm_timeout = 6;
1578         ping_evictor_start();
1579
1580         RETURN(0);
1581
1582 err_fs:
1583         /* No extra cleanup needed for llog_init_commit_thread() */
1584         mds_fs_cleanup(obd);
1585 err_ns:
1586         ldlm_namespace_free(obd->obd_namespace, 0);
1587         obd->obd_namespace = NULL;
1588 err_put:
1589         unlock_kernel();
1590         mntput(mds->mds_vfsmnt);
1591         mds->mds_sb = 0;
1592         lock_kernel();
1593 err_ops:
1594         fsfilt_put_ops(obd->obd_fsops);
1595         return rc;
1596 }
1597
1598 static int mds_postsetup(struct obd_device *obd)
1599 {
1600         struct mds_obd *mds = &obd->u.mds;
1601         int rc = 0;
1602         ENTRY;
1603
1604         rc = llog_setup(obd, LLOG_CONFIG_ORIG_CTXT, obd, 0, NULL,
1605                         &llog_lvfs_ops);
1606         if (rc)
1607                 RETURN(rc);
1608
1609         if (mds->mds_profile) {
1610                 struct lvfs_run_ctxt saved;
1611                 struct lustre_profile *lprof;
1612                 struct config_llog_instance cfg;
1613
1614                 cfg.cfg_instance = NULL;
1615                 cfg.cfg_uuid = mds->mds_lov_uuid;
1616                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1617                 rc = class_config_parse_llog(llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT),
1618                                              mds->mds_profile, &cfg);
1619                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1620                 switch (rc) {
1621                 case 0:
1622                         break;
1623                 case -EINVAL:
1624                         LCONSOLE_ERROR("%s: the profile %s could not be read. "
1625                                        "If you recently installed a new "
1626                                        "version of Lustre, you may need to "
1627                                        "re-run 'lconf --write_conf "
1628                                        "<yourconfig>.xml' command line before "
1629                                        "restarting the MDS.\n",
1630                                        obd->obd_name, mds->mds_profile);
1631                         /* fall through */
1632                 default:
1633                         GOTO(err_llog, rc);
1634                         break;
1635                 }
1636
1637                 lprof = class_get_profile(mds->mds_profile);
1638                 if (lprof == NULL) {
1639                         CERROR("No profile found: %s\n", mds->mds_profile);
1640                         GOTO(err_cleanup, rc = -ENOENT);
1641                 }
1642                 rc = mds_lov_connect(obd, lprof->lp_osc);
1643                 if (rc)
1644                         GOTO(err_cleanup, rc);
1645         }
1646
1647         RETURN(rc);
1648
1649 err_cleanup:
1650         mds_lov_clean(obd);
1651 err_llog:
1652         llog_cleanup(llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT));
1653         RETURN(rc);
1654 }
1655
1656 int mds_postrecov(struct obd_device *obd)
1657 {
1658         struct mds_obd *mds = &obd->u.mds;
1659         int rc, item = 0;
1660         ENTRY;
1661
1662         if (obd->obd_fail)
1663                 RETURN(0);
1664
1665         LASSERT(!obd->obd_recovering);
1666         LASSERT(llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT) != NULL);
1667
1668         /* set nextid first, so we are sure it happens */
1669         rc = mds_lov_set_nextid(obd);
1670         if (rc) {
1671                 CERROR ("%s: mds_lov_set_nextid failed\n",
1672                         obd->obd_name);
1673                 GOTO(out, rc);
1674         }
1675
1676         /* clean PENDING dir */
1677         rc = mds_cleanup_orphans(obd);
1678         if (rc < 0) {
1679                 GOTO(out, rc);
1680         } else {
1681                 item = rc;
1682         }
1683
1684         rc = obd_set_info(mds->mds_osc_exp, strlen("mds_conn"), "mds_conn",
1685                           0, NULL);
1686         if (rc)
1687                 GOTO(out, rc);
1688
1689         rc = llog_connect(llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT),
1690                           obd->u.mds.mds_lov_desc.ld_tgt_count,
1691                           NULL, NULL, NULL);
1692         if (rc) {
1693                 CERROR("%s: failed at llog_origin_connect: %d\n",
1694                        obd->obd_name, rc);
1695                 GOTO(out, rc);
1696         }
1697
1698         /* remove the orphaned precreated objects */
1699         rc = mds_lov_clearorphans(mds, NULL /* all OSTs */);
1700         if (rc) {
1701                 GOTO(err_llog, rc);
1702         }
1703
1704 out:
1705         RETURN(rc < 0 ? rc : item);
1706
1707 err_llog:
1708         /* cleanup all llogging subsystems */
1709         rc = obd_llog_finish(obd, mds->mds_lov_desc.ld_tgt_count);
1710         if (rc)
1711                 CERROR("%s: failed to cleanup llogging subsystems\n",
1712                         obd->obd_name);
1713         goto out;
1714 }
1715
1716 int mds_lov_clean(struct obd_device *obd)
1717 {
1718         struct mds_obd *mds = &obd->u.mds;
1719
1720         if (mds->mds_profile) {
1721                 char * cln_prof;
1722                 struct config_llog_instance cfg;
1723                 struct lvfs_run_ctxt saved;
1724                 int len = strlen(mds->mds_profile) + sizeof("-clean") + 1;
1725
1726                 OBD_ALLOC(cln_prof, len);
1727                 sprintf(cln_prof, "%s-clean", mds->mds_profile);
1728
1729                 cfg.cfg_instance = NULL;
1730                 cfg.cfg_uuid = mds->mds_lov_uuid;
1731
1732                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1733                 class_config_parse_llog(llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT),
1734                                         cln_prof, &cfg);
1735                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1736
1737                 OBD_FREE(cln_prof, len);
1738                 OBD_FREE(mds->mds_profile, strlen(mds->mds_profile) + 1);
1739                 mds->mds_profile = NULL;
1740         }
1741         RETURN(0);
1742 }
1743
1744 static int mds_precleanup(struct obd_device *obd, int stage)
1745 {
1746         int rc = 0;
1747         ENTRY;
1748
1749         switch (stage) {
1750         case 1:
1751                 mds_lov_set_cleanup_flags(obd);
1752                 target_cleanup_recovery(obd);
1753                 break;
1754         case 2:
1755                 mds_lov_disconnect(obd);
1756                 mds_lov_clean(obd);
1757                 llog_cleanup(llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT));
1758                 rc = obd_llog_finish(obd, 0);
1759         }
1760         RETURN(rc);
1761 }
1762
1763 static int mds_cleanup(struct obd_device *obd)
1764 {
1765         struct mds_obd *mds = &obd->u.mds;
1766         lvfs_sbdev_type save_dev;
1767         int must_relock = 0;
1768         ENTRY;
1769
1770         ping_evictor_stop();
1771
1772         if (mds->mds_sb == NULL)
1773                 RETURN(0);
1774         save_dev = lvfs_sbdev(mds->mds_sb);
1775
1776         if (mds->mds_osc_exp)
1777                 /* lov export was disconnected by mds_lov_clean;
1778                    we just need to drop our ref */
1779                 class_export_put(mds->mds_osc_exp);
1780
1781         lprocfs_obd_cleanup(obd);
1782
1783         mds_quota_cleanup(mds);
1784
1785         mds_update_server_data(obd, 1);
1786         if (mds->mds_lov_objids != NULL) {
1787                 OBD_FREE(mds->mds_lov_objids,
1788                          mds->mds_lov_desc.ld_tgt_count * sizeof(obd_id));
1789         }
1790         mds_fs_cleanup(obd);
1791
1792         /* 2 seems normal on mds, (may_umount() also expects 2
1793           fwiw), but we only see 1 at this point in obdfilter. */
1794         if (atomic_read(&obd->u.mds.mds_vfsmnt->mnt_count) > 2)
1795                 CERROR("%s: mount busy, mnt_count %d != 2\n", obd->obd_name,
1796                        atomic_read(&obd->u.mds.mds_vfsmnt->mnt_count));
1797
1798         /* We can only unlock kernel if we are in the context of sys_ioctl,
1799            otherwise we never called lock_kernel */
1800         if (kernel_locked()) {
1801                 unlock_kernel();
1802                 must_relock++;
1803         }
1804
1805         mntput(mds->mds_vfsmnt);
1806         mds->mds_sb = NULL;
1807
1808         ldlm_namespace_free(obd->obd_namespace, obd->obd_force);
1809
1810         spin_lock_bh(&obd->obd_processing_task_lock);
1811         if (obd->obd_recovering) {
1812                 target_cancel_recovery_timer(obd);
1813                 obd->obd_recovering = 0;
1814         }
1815         spin_unlock_bh(&obd->obd_processing_task_lock);
1816
1817         lvfs_clear_rdonly(save_dev);
1818
1819         if (must_relock)
1820                 lock_kernel();
1821
1822         fsfilt_put_ops(obd->obd_fsops);
1823
1824         LCONSOLE_INFO("MDT %s has stopped.\n", obd->obd_name);
1825
1826         RETURN(0);
1827 }
1828
1829 static void fixup_handle_for_resent_req(struct ptlrpc_request *req,
1830                                         struct ldlm_lock *new_lock,
1831                                         struct ldlm_lock **old_lock,
1832                                         struct lustre_handle *lockh)
1833 {
1834         struct obd_export *exp = req->rq_export;
1835         struct obd_device *obd = exp->exp_obd;
1836         struct ldlm_request *dlmreq =
1837                 lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*dlmreq));
1838         struct lustre_handle remote_hdl = dlmreq->lock_handle1;
1839         struct list_head *iter;
1840
1841         if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT))
1842                 return;
1843
1844         l_lock(&obd->obd_namespace->ns_lock);
1845         list_for_each(iter, &exp->exp_ldlm_data.led_held_locks) {
1846                 struct ldlm_lock *lock;
1847                 lock = list_entry(iter, struct ldlm_lock, l_export_chain);
1848                 if (lock == new_lock)
1849                         continue;
1850                 if (lock->l_remote_handle.cookie == remote_hdl.cookie) {
1851                         lockh->cookie = lock->l_handle.h_cookie;
1852                         LDLM_DEBUG(lock, "restoring lock cookie");
1853                         DEBUG_REQ(D_HA, req, "restoring lock cookie "LPX64,
1854                                   lockh->cookie);
1855                         if (old_lock)
1856                                 *old_lock = LDLM_LOCK_GET(lock);
1857                         l_unlock(&obd->obd_namespace->ns_lock);
1858                         return;
1859                 }
1860         }
1861         l_unlock(&obd->obd_namespace->ns_lock);
1862
1863         /* If the xid matches, then we know this is a resent request,
1864          * and allow it. (It's probably an OPEN, for which we don't
1865          * send a lock */
1866         if (req->rq_xid ==
1867             le64_to_cpu(exp->exp_mds_data.med_mcd->mcd_last_xid))
1868                 return;
1869
1870         /* This remote handle isn't enqueued, so we never received or
1871          * processed this request.  Clear MSG_RESENT, because it can
1872          * be handled like any normal request now. */
1873
1874         lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
1875
1876         DEBUG_REQ(D_HA, req, "no existing lock with rhandle "LPX64,
1877                   remote_hdl.cookie);
1878 }
1879
1880 int intent_disposition(struct ldlm_reply *rep, int flag)
1881 {
1882         if (!rep)
1883                 return 0;
1884         return (rep->lock_policy_res1 & flag);
1885 }
1886
1887 void intent_set_disposition(struct ldlm_reply *rep, int flag)
1888 {
1889         if (!rep)
1890                 return;
1891         rep->lock_policy_res1 |= flag;
1892 }
1893
1894 static int mds_intent_policy(struct ldlm_namespace *ns,
1895                              struct ldlm_lock **lockp, void *req_cookie,
1896                              ldlm_mode_t mode, int flags, void *data)
1897 {
1898         struct ptlrpc_request *req = req_cookie;
1899         struct ldlm_lock *lock = *lockp;
1900         struct ldlm_intent *it;
1901         struct mds_obd *mds = &req->rq_export->exp_obd->u.mds;
1902         struct ldlm_reply *rep;
1903         struct lustre_handle lockh = { 0 };
1904         struct ldlm_lock *new_lock = NULL;
1905         int rc, offset = 2, repsize[4] = {sizeof(struct ldlm_reply),
1906                                           sizeof(struct mds_body),
1907                                           mds->mds_max_mdsize,
1908                                           mds->mds_max_cookiesize};
1909         ENTRY;
1910
1911         LASSERT(req != NULL);
1912
1913         if (req->rq_reqmsg->bufcount <= 1) {
1914                 /* No intent was provided */
1915                 int size = sizeof(struct ldlm_reply);
1916                 rc = lustre_pack_reply(req, 1, &size, NULL);
1917                 LASSERT(rc == 0);
1918                 RETURN(0);
1919         }
1920
1921         it = lustre_swab_reqbuf(req, 1, sizeof(*it), lustre_swab_ldlm_intent);
1922         if (it == NULL) {
1923                 CERROR("Intent missing\n");
1924                 RETURN(req->rq_status = -EFAULT);
1925         }
1926
1927         LDLM_DEBUG(lock, "intent policy, opc: %s", ldlm_it2str(it->opc));
1928
1929         rc = lustre_pack_reply(req, it->opc == IT_UNLINK ? 4 : 3, repsize,
1930                                NULL);
1931         if (rc)
1932                 RETURN(req->rq_status = rc);
1933
1934         rep = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*rep));
1935         intent_set_disposition(rep, DISP_IT_EXECD);
1936
1937
1938         /* execute policy */
1939         switch ((long)it->opc) {
1940         case IT_OPEN:
1941         case IT_CREAT|IT_OPEN:
1942                 fixup_handle_for_resent_req(req, lock, NULL, &lockh);
1943                 /* XXX swab here to assert that an mds_open reint
1944                  * packet is following */
1945                 rep->lock_policy_res2 = mds_reint(req, offset, &lockh);
1946 #if 0
1947                 /* We abort the lock if the lookup was negative and
1948                  * we did not make it to the OPEN portion */
1949                 if (!intent_disposition(rep, DISP_LOOKUP_EXECD))
1950                         RETURN(ELDLM_LOCK_ABORTED);
1951                 if (intent_disposition(rep, DISP_LOOKUP_NEG) &&
1952                     !intent_disposition(rep, DISP_OPEN_OPEN))
1953 #endif
1954                         RETURN(ELDLM_LOCK_ABORTED);
1955                 break;
1956         case IT_GETATTR:
1957         case IT_LOOKUP:
1958         case IT_READDIR:
1959                 fixup_handle_for_resent_req(req, lock, &new_lock, &lockh);
1960                 rep->lock_policy_res2 = mds_getattr_name(offset, req, &lockh);
1961                 /* FIXME: LDLM can set req->rq_status. MDS sets
1962                    policy_res{1,2} with disposition and status.
1963                    - replay: returns 0 & req->status is old status
1964                    - otherwise: returns req->status */
1965                 if (intent_disposition(rep, DISP_LOOKUP_NEG))
1966                         rep->lock_policy_res2 = 0;
1967                 if (!intent_disposition(rep, DISP_LOOKUP_POS) ||
1968                     rep->lock_policy_res2)
1969                         RETURN(ELDLM_LOCK_ABORTED);
1970                 if (req->rq_status != 0) {
1971                         LBUG();
1972                         rep->lock_policy_res2 = req->rq_status;
1973                         RETURN(ELDLM_LOCK_ABORTED);
1974                 }
1975                 break;
1976         default:
1977                 CERROR("Unhandled intent "LPD64"\n", it->opc);
1978                 LBUG();
1979         }
1980
1981         /* By this point, whatever function we called above must have either
1982          * filled in 'lockh', been an intent replay, or returned an error.  We
1983          * want to allow replayed RPCs to not get a lock, since we would just
1984          * drop it below anyways because lock replay is done separately by the
1985          * client afterwards.  For regular RPCs we want to give the new lock to
1986          * the client instead of whatever lock it was about to get. */
1987         if (new_lock == NULL)
1988                 new_lock = ldlm_handle2lock(&lockh);
1989         if (new_lock == NULL && (flags & LDLM_FL_INTENT_ONLY))
1990                 RETURN(0);
1991
1992         LASSERTF(new_lock != NULL, "op "LPX64" lockh "LPX64"\n",
1993                  it->opc, lockh.cookie);
1994
1995         /* If we've already given this lock to a client once, then we should
1996          * have no readers or writers.  Otherwise, we should have one reader
1997          * _or_ writer ref (which will be zeroed below) before returning the
1998          * lock to a client. */
1999         if (new_lock->l_export == req->rq_export) {
2000                 LASSERT(new_lock->l_readers + new_lock->l_writers == 0);
2001         } else {
2002                 LASSERT(new_lock->l_export == NULL);
2003                 LASSERT(new_lock->l_readers + new_lock->l_writers == 1);
2004         }
2005
2006         *lockp = new_lock;
2007
2008         if (new_lock->l_export == req->rq_export) {
2009                 /* Already gave this to the client, which means that we
2010                  * reconstructed a reply. */
2011                 LASSERT(lustre_msg_get_flags(req->rq_reqmsg) &
2012                         MSG_RESENT);
2013                 RETURN(ELDLM_LOCK_REPLACED);
2014         }
2015
2016         /* Fixup the lock to be given to the client */
2017         l_lock(&new_lock->l_resource->lr_namespace->ns_lock);
2018         new_lock->l_readers = 0;
2019         new_lock->l_writers = 0;
2020
2021         new_lock->l_export = class_export_get(req->rq_export);
2022         list_add(&new_lock->l_export_chain,
2023                  &new_lock->l_export->exp_ldlm_data.led_held_locks);
2024
2025         new_lock->l_blocking_ast = lock->l_blocking_ast;
2026         new_lock->l_completion_ast = lock->l_completion_ast;
2027
2028         memcpy(&new_lock->l_remote_handle, &lock->l_remote_handle,
2029                sizeof(lock->l_remote_handle));
2030
2031         new_lock->l_flags &= ~LDLM_FL_LOCAL;
2032
2033         LDLM_LOCK_PUT(new_lock);
2034         l_unlock(&new_lock->l_resource->lr_namespace->ns_lock);
2035
2036         RETURN(ELDLM_LOCK_REPLACED);
2037 }
2038
2039 static int mdt_setup(struct obd_device *obd, obd_count len, void *buf)
2040 {
2041         struct mds_obd *mds = &obd->u.mds;
2042         struct lprocfs_static_vars lvars;
2043         int rc = 0;
2044         ENTRY;
2045
2046         lprocfs_init_vars(mdt, &lvars);
2047         lprocfs_obd_setup(obd, lvars.obd_vars);
2048
2049         mds->mds_service =
2050                 ptlrpc_init_svc(MDS_NBUFS, MDS_BUFSIZE, MDS_MAXREQSIZE,
2051                                 MDS_REQUEST_PORTAL, MDC_REPLY_PORTAL,
2052                                 MDS_SERVICE_WATCHDOG_TIMEOUT,
2053                                 mds_handle, "mds", obd->obd_proc_entry, NULL);
2054
2055         if (!mds->mds_service) {
2056                 CERROR("failed to start service\n");
2057                 GOTO(err_lprocfs, rc = -ENOMEM);
2058         }
2059
2060         rc = ptlrpc_start_n_threads(obd, mds->mds_service, MDT_NUM_THREADS,
2061                                     "ll_mdt");
2062         if (rc)
2063                 GOTO(err_thread, rc);
2064
2065         mds->mds_setattr_service =
2066                 ptlrpc_init_svc(MDS_NBUFS, MDS_BUFSIZE, MDS_MAXREQSIZE,
2067                                 MDS_SETATTR_PORTAL, MDC_REPLY_PORTAL,
2068                                 MDS_SERVICE_WATCHDOG_TIMEOUT,
2069                                 mds_handle, "mds_setattr",
2070                                 obd->obd_proc_entry, NULL);
2071         if (!mds->mds_setattr_service) {
2072                 CERROR("failed to start getattr service\n");
2073                 GOTO(err_thread, rc = -ENOMEM);
2074         }
2075
2076         rc = ptlrpc_start_n_threads(obd, mds->mds_setattr_service,
2077                                     MDT_NUM_THREADS, "ll_mdt_attr");
2078         if (rc)
2079                 GOTO(err_thread2, rc);
2080
2081         mds->mds_readpage_service =
2082                 ptlrpc_init_svc(MDS_NBUFS, MDS_BUFSIZE, MDS_MAXREQSIZE,
2083                                 MDS_READPAGE_PORTAL, MDC_REPLY_PORTAL,
2084                                 MDS_SERVICE_WATCHDOG_TIMEOUT,
2085                                 mds_handle, "mds_readpage",
2086                                 obd->obd_proc_entry, NULL);
2087         if (!mds->mds_readpage_service) {
2088                 CERROR("failed to start readpage service\n");
2089                 GOTO(err_thread2, rc = -ENOMEM);
2090         }
2091
2092         rc = ptlrpc_start_n_threads(obd, mds->mds_readpage_service,
2093                                     MDT_NUM_THREADS, "ll_mdt_rdpg");
2094
2095         if (rc)
2096                 GOTO(err_thread3, rc);
2097
2098         RETURN(0);
2099
2100 err_thread3:
2101         ptlrpc_unregister_service(mds->mds_readpage_service);
2102 err_thread2:
2103         ptlrpc_unregister_service(mds->mds_setattr_service);
2104 err_thread:
2105         ptlrpc_unregister_service(mds->mds_service);
2106 err_lprocfs:
2107         lprocfs_obd_cleanup(obd);
2108         return rc;
2109 }
2110
2111 static int mdt_cleanup(struct obd_device *obd)
2112 {
2113         struct mds_obd *mds = &obd->u.mds;
2114         ENTRY;
2115
2116         ptlrpc_unregister_service(mds->mds_readpage_service);
2117         ptlrpc_unregister_service(mds->mds_setattr_service);
2118         ptlrpc_unregister_service(mds->mds_service);
2119
2120         lprocfs_obd_cleanup(obd);
2121
2122         RETURN(0);
2123 }
2124
2125 static struct dentry *mds_lvfs_fid2dentry(__u64 id, __u32 gen, __u64 gr,
2126                                           void *data)
2127 {
2128         struct obd_device *obd = data;
2129         struct ll_fid fid;
2130         fid.id = id;
2131         fid.generation = gen;
2132         return mds_fid2dentry(&obd->u.mds, &fid, NULL);
2133 }
2134
2135 struct lvfs_callback_ops mds_lvfs_ops = {
2136         l_fid2dentry:     mds_lvfs_fid2dentry,
2137 };
2138
2139 /* use obd ops to offer management infrastructure */
2140 static struct obd_ops mds_obd_ops = {
2141         .o_owner           = THIS_MODULE,
2142         .o_connect         = mds_connect,
2143         .o_init_export     = mds_init_export,
2144         .o_destroy_export  = mds_destroy_export,
2145         .o_disconnect      = mds_disconnect,
2146         .o_setup           = mds_setup,
2147         .o_precleanup      = mds_precleanup,
2148         .o_cleanup         = mds_cleanup,
2149         .o_postrecov       = mds_postrecov,
2150         .o_statfs          = mds_obd_statfs,
2151         .o_iocontrol       = mds_iocontrol,
2152         .o_create          = mds_obd_create,
2153         .o_destroy         = mds_obd_destroy,
2154         .o_llog_init       = mds_llog_init,
2155         .o_llog_finish     = mds_llog_finish,
2156         .o_notify          = mds_notify,
2157 };
2158
2159 static struct obd_ops mdt_obd_ops = {
2160         .o_owner           = THIS_MODULE,
2161         .o_setup           = mdt_setup,
2162         .o_cleanup         = mdt_cleanup,
2163 };
2164
2165 static int __init mds_init(void)
2166 {
2167         int rc;
2168         struct lprocfs_static_vars lvars;
2169
2170         rc = lustre_dquot_init();
2171         if (rc)
2172                 return rc;
2173
2174         lprocfs_init_vars(mds, &lvars);
2175         class_register_type(&mds_obd_ops, lvars.module_vars, LUSTRE_MDS_NAME);
2176         lprocfs_init_vars(mdt, &lvars);
2177         class_register_type(&mdt_obd_ops, lvars.module_vars, LUSTRE_MDT_NAME);
2178
2179         return 0;
2180 }
2181
2182 static void /*__exit*/ mds_exit(void)
2183 {
2184         lustre_dquot_exit();
2185
2186         class_unregister_type(LUSTRE_MDS_NAME);
2187         class_unregister_type(LUSTRE_MDT_NAME);
2188 }
2189
2190 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
2191 MODULE_DESCRIPTION("Lustre Metadata Server (MDS)");
2192 MODULE_LICENSE("GPL");
2193
2194 module_init(mds_init);
2195 module_exit(mds_exit);