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