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