lustre/mds/handler.c

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