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