lustre/mdt/mdt_handler.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2012, Whamcloud, Inc.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/mdt/mdt_handler.c
  37  *
  38  * Lustre Metadata Target (mdt) request handler
  39  *
  40  * Author: Peter Braam <braam@clusterfs.com>
  41  * Author: Andreas Dilger <adilger@clusterfs.com>
  42  * Author: Phil Schwan <phil@clusterfs.com>
  43  * Author: Mike Shaver <shaver@clusterfs.com>
  44  * Author: Nikita Danilov <nikita@clusterfs.com>
  45  * Author: Huang Hua <huanghua@clusterfs.com>
  46  * Author: Yury Umanets <umka@clusterfs.com>
  47  */
  48
  49 #ifndef EXPORT_SYMTAB
  50 # define EXPORT_SYMTAB
  51 #endif
  52 #define DEBUG_SUBSYSTEM S_MDS
  53
  54 #include <linux/module.h>
  55 /*
  56  * struct OBD_{ALLOC,FREE}*()
  57  */
  58 #include <obd_support.h>
  59 /* struct ptlrpc_request */
  60 #include <lustre_net.h>
  61 /* struct obd_export */
  62 #include <lustre_export.h>
  63 /* struct obd_device */
  64 #include <obd.h>
  65 /* lu2dt_dev() */
  66 #include <dt_object.h>
  67 #include <lustre_mds.h>
  68 #include <lustre_mdt.h>
  69 #include "mdt_internal.h"
  70 #ifdef HAVE_QUOTA_SUPPORT
  71 # include <lustre_quota.h>
  72 #endif
  73 #include <lustre_acl.h>
  74 #include <lustre_param.h>
  75 #include <lustre_fsfilt.h>
  76
  77 mdl_mode_t mdt_mdl_lock_modes[] = {
  78         [LCK_MINMODE] = MDL_MINMODE,
  79         [LCK_EX]      = MDL_EX,
  80         [LCK_PW]      = MDL_PW,
  81         [LCK_PR]      = MDL_PR,
  82         [LCK_CW]      = MDL_CW,
  83         [LCK_CR]      = MDL_CR,
  84         [LCK_NL]      = MDL_NL,
  85         [LCK_GROUP]   = MDL_GROUP
  86 };
  87
  88 ldlm_mode_t mdt_dlm_lock_modes[] = {
  89         [MDL_MINMODE] = LCK_MINMODE,
  90         [MDL_EX]      = LCK_EX,
  91         [MDL_PW]      = LCK_PW,
  92         [MDL_PR]      = LCK_PR,
  93         [MDL_CW]      = LCK_CW,
  94         [MDL_CR]      = LCK_CR,
  95         [MDL_NL]      = LCK_NL,
  96         [MDL_GROUP]   = LCK_GROUP
  97 };
  98
  99 /*
 100  * Initialized in mdt_mod_init().
 101  */
 102 static unsigned long mdt_num_threads;
 103 static unsigned long mdt_min_threads;
 104 static unsigned long mdt_max_threads;
 105
 106 /* ptlrpc request handler for MDT. All handlers are
 107  * grouped into several slices - struct mdt_opc_slice,
 108  * and stored in an array - mdt_handlers[].
 109  */
 110 struct mdt_handler {
 111         /* The name of this handler. */
 112         const char *mh_name;
 113         /* Fail id for this handler, checked at the beginning of this handler*/
 114         int         mh_fail_id;
 115         /* Operation code for this handler */
 116         __u32       mh_opc;
 117         /* flags are listed in enum mdt_handler_flags below. */
 118         __u32       mh_flags;
 119         /* The actual handler function to execute. */
 120         int (*mh_act)(struct mdt_thread_info *info);
 121         /* Request format for this request. */
 122         const struct req_format *mh_fmt;
 123 };
 124
 125 enum mdt_handler_flags {
 126         /*
 127          * struct mdt_body is passed in the incoming message, and object
 128          * identified by this fid exists on disk.
 129          *
 130          * "habeo corpus" == "I have a body"
 131          */
 132         HABEO_CORPUS = (1 << 0),
 133         /*
 134          * struct ldlm_request is passed in the incoming message.
 135          *
 136          * "habeo clavis" == "I have a key"
 137          */
 138         HABEO_CLAVIS = (1 << 1),
 139         /*
 140          * this request has fixed reply format, so that reply message can be
 141          * packed by generic code.
 142          *
 143          * "habeo refero" == "I have a reply"
 144          */
 145         HABEO_REFERO = (1 << 2),
 146         /*
 147          * this request will modify something, so check whether the filesystem
 148          * is readonly or not, then return -EROFS to client asap if necessary.
 149          *
 150          * "mutabor" == "I shall modify"
 151          */
 152         MUTABOR      = (1 << 3)
 153 };
 154
 155 struct mdt_opc_slice {
 156         __u32               mos_opc_start;
 157         int                 mos_opc_end;
 158         struct mdt_handler *mos_hs;
 159 };
 160
 161 static struct mdt_opc_slice mdt_regular_handlers[];
 162 static struct mdt_opc_slice mdt_readpage_handlers[];
 163 static struct mdt_opc_slice mdt_xmds_handlers[];
 164 static struct mdt_opc_slice mdt_seq_handlers[];
 165 static struct mdt_opc_slice mdt_fld_handlers[];
 166
 167 static struct mdt_device *mdt_dev(struct lu_device *d);
 168 static int mdt_regular_handle(struct ptlrpc_request *req);
 169 static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags);
 170 static int mdt_fid2path(const struct lu_env *env, struct mdt_device *mdt,
 171                         struct getinfo_fid2path *fp);
 172
 173 static const struct lu_object_operations mdt_obj_ops;
 174
 175 int mdt_get_disposition(struct ldlm_reply *rep, int flag)
 176 {
 177         if (!rep)
 178                 return 0;
 179         return (rep->lock_policy_res1 & flag);
 180 }
 181
 182 void mdt_clear_disposition(struct mdt_thread_info *info,
 183                            struct ldlm_reply *rep, int flag)
 184 {
 185         if (info)
 186                 info->mti_opdata &= ~flag;
 187         if (rep)
 188                 rep->lock_policy_res1 &= ~flag;
 189 }
 190
 191 void mdt_set_disposition(struct mdt_thread_info *info,
 192                          struct ldlm_reply *rep, int flag)
 193 {
 194         if (info)
 195                 info->mti_opdata |= flag;
 196         if (rep)
 197                 rep->lock_policy_res1 |= flag;
 198 }
 199
 200 void mdt_lock_reg_init(struct mdt_lock_handle *lh, ldlm_mode_t lm)
 201 {
 202         lh->mlh_pdo_hash = 0;
 203         lh->mlh_reg_mode = lm;
 204         lh->mlh_type = MDT_REG_LOCK;
 205 }
 206
 207 void mdt_lock_pdo_init(struct mdt_lock_handle *lh, ldlm_mode_t lm,
 208                        const char *name, int namelen)
 209 {
 210         lh->mlh_reg_mode = lm;
 211         lh->mlh_type = MDT_PDO_LOCK;
 212
 213         if (name != NULL && (name[0] != '\0')) {
 214                 LASSERT(namelen > 0);
 215                 lh->mlh_pdo_hash = full_name_hash(name, namelen);
 216         } else {
 217                 LASSERT(namelen == 0);
 218                 lh->mlh_pdo_hash = 0ull;
 219         }
 220 }
 221
 222 static void mdt_lock_pdo_mode(struct mdt_thread_info *info, struct mdt_object *o,
 223                               struct mdt_lock_handle *lh)
 224 {
 225         mdl_mode_t mode;
 226         ENTRY;
 227
 228         /*
 229          * Any dir access needs couple of locks:
 230          *
 231          * 1) on part of dir we gonna take lookup/modify;
 232          *
 233          * 2) on whole dir to protect it from concurrent splitting and/or to
 234          * flush client's cache for readdir().
 235          *
 236          * so, for a given mode and object this routine decides what lock mode
 237          * to use for lock #2:
 238          *
 239          * 1) if caller's gonna lookup in dir then we need to protect dir from
 240          * being splitted only - LCK_CR
 241          *
 242          * 2) if caller's gonna modify dir then we need to protect dir from
 243          * being splitted and to flush cache - LCK_CW
 244          *
 245          * 3) if caller's gonna modify dir and that dir seems ready for
 246          * splitting then we need to protect it from any type of access
 247          * (lookup/modify/split) - LCK_EX --bzzz
 248          */
 249
 250         LASSERT(lh->mlh_reg_mode != LCK_MINMODE);
 251         LASSERT(lh->mlh_pdo_mode == LCK_MINMODE);
 252
 253         /*
 254          * Ask underlaying level its opinion about preferable PDO lock mode
 255          * having access type passed as regular lock mode:
 256          *
 257          * - MDL_MINMODE means that lower layer does not want to specify lock
 258          * mode;
 259          *
 260          * - MDL_NL means that no PDO lock should be taken. This is used in some
 261          * cases. Say, for non-splittable directories no need to use PDO locks
 262          * at all.
 263          */
 264         mode = mdo_lock_mode(info->mti_env, mdt_object_child(o),
 265                              mdt_dlm_mode2mdl_mode(lh->mlh_reg_mode));
 266
 267         if (mode != MDL_MINMODE) {
 268                 lh->mlh_pdo_mode = mdt_mdl_mode2dlm_mode(mode);
 269         } else {
 270                 /*
 271                  * Lower layer does not want to specify locking mode. We do it
 272                  * our selves. No special protection is needed, just flush
 273                  * client's cache on modification and allow concurrent
 274                  * mondification.
 275                  */
 276                 switch (lh->mlh_reg_mode) {
 277                 case LCK_EX:
 278                         lh->mlh_pdo_mode = LCK_EX;
 279                         break;
 280                 case LCK_PR:
 281                         lh->mlh_pdo_mode = LCK_CR;
 282                         break;
 283                 case LCK_PW:
 284                         lh->mlh_pdo_mode = LCK_CW;
 285                         break;
 286                 default:
 287                         CERROR("Not expected lock type (0x%x)\n",
 288                                (int)lh->mlh_reg_mode);
 289                         LBUG();
 290                 }
 291         }
 292
 293         LASSERT(lh->mlh_pdo_mode != LCK_MINMODE);
 294         EXIT;
 295 }
 296
 297 static int mdt_getstatus(struct mdt_thread_info *info)
 298 {
 299         struct mdt_device *mdt  = info->mti_mdt;
 300         struct md_device  *next = mdt->mdt_child;
 301         struct mdt_body   *repbody;
 302         int                rc;
 303
 304         ENTRY;
 305
 306         rc = mdt_check_ucred(info);
 307         if (rc)
 308                 RETURN(err_serious(rc));
 309
 310         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETSTATUS_PACK))
 311                 RETURN(err_serious(-ENOMEM));
 312
 313         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 314         rc = next->md_ops->mdo_root_get(info->mti_env, next, &repbody->fid1);
 315         if (rc != 0)
 316                 RETURN(rc);
 317
 318         repbody->valid |= OBD_MD_FLID;
 319
 320         if (mdt->mdt_opts.mo_mds_capa &&
 321             info->mti_exp->exp_connect_flags & OBD_CONNECT_MDS_CAPA) {
 322                 struct mdt_object  *root;
 323                 struct lustre_capa *capa;
 324
 325                 root = mdt_object_find(info->mti_env, mdt, &repbody->fid1);
 326                 if (IS_ERR(root))
 327                         RETURN(PTR_ERR(root));
 328
 329                 capa = req_capsule_server_get(info->mti_pill, &RMF_CAPA1);
 330                 LASSERT(capa);
 331                 capa->lc_opc = CAPA_OPC_MDS_DEFAULT;
 332                 rc = mo_capa_get(info->mti_env, mdt_object_child(root), capa,
 333                                  0);
 334                 mdt_object_put(info->mti_env, root);
 335                 if (rc == 0)
 336                         repbody->valid |= OBD_MD_FLMDSCAPA;
 337         }
 338
 339         RETURN(rc);
 340 }
 341
 342 static int mdt_statfs(struct mdt_thread_info *info)
 343 {
 344         struct ptlrpc_request *req = mdt_info_req(info);
 345         struct md_device      *next  = info->mti_mdt->mdt_child;
 346         struct ptlrpc_service *svc;
 347         struct obd_statfs     *osfs;
 348         int                    rc;
 349
 350         ENTRY;
 351
 352         svc = info->mti_pill->rc_req->rq_rqbd->rqbd_service;
 353
 354         /* This will trigger a watchdog timeout */
 355         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_STATFS_LCW_SLEEP,
 356                          (MDT_SERVICE_WATCHDOG_FACTOR *
 357                           at_get(&svc->srv_at_estimate)) + 1);
 358
 359         rc = mdt_check_ucred(info);
 360         if (rc)
 361                 RETURN(err_serious(rc));
 362
 363         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_STATFS_PACK)) {
 364                 rc = err_serious(-ENOMEM);
 365         } else {
 366                 osfs = req_capsule_server_get(info->mti_pill, &RMF_OBD_STATFS);
 367                 rc = next->md_ops->mdo_statfs(info->mti_env, next, osfs);
 368         }
 369
 370         if (rc == 0)
 371                 mdt_counter_incr(req, LPROC_MDT_STATFS);
 372
 373         RETURN(rc);
 374 }
 375
 376 /**
 377  * Pack SOM attributes into the reply.
 378  * Call under a DLM UPDATE lock.
 379  */
 380 static void mdt_pack_size2body(struct mdt_thread_info *info,
 381                                struct mdt_object *mo)
 382 {
 383         struct mdt_body *b;
 384         struct md_attr *ma = &info->mti_attr;
 385
 386         LASSERT(ma->ma_attr.la_valid & LA_MODE);
 387         b = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 388
 389         /* Check if Size-on-MDS is supported, if this is a regular file,
 390          * if SOM is enabled on the object and if SOM cache exists and valid.
 391          * Otherwise do not pack Size-on-MDS attributes to the reply. */
 392         if (!(mdt_conn_flags(info) & OBD_CONNECT_SOM) ||
 393             !S_ISREG(ma->ma_attr.la_mode) ||
 394             !mdt_object_is_som_enabled(mo) ||
 395             !(ma->ma_valid & MA_SOM))
 396                 return;
 397
 398         b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
 399         b->size = ma->ma_som->msd_size;
 400         b->blocks = ma->ma_som->msd_blocks;
 401 }
 402
 403 void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b,
 404                         const struct lu_attr *attr, const struct lu_fid *fid)
 405 {
 406         struct md_attr *ma = &info->mti_attr;
 407
 408         LASSERT(ma->ma_valid & MA_INODE);
 409
 410         b->atime      = attr->la_atime;
 411         b->mtime      = attr->la_mtime;
 412         b->ctime      = attr->la_ctime;
 413         b->mode       = attr->la_mode;
 414         b->size       = attr->la_size;
 415         b->blocks     = attr->la_blocks;
 416         b->uid        = attr->la_uid;
 417         b->gid        = attr->la_gid;
 418         b->flags      = attr->la_flags;
 419         b->nlink      = attr->la_nlink;
 420         b->rdev       = attr->la_rdev;
 421
 422         /*XXX should pack the reply body according to lu_valid*/
 423         b->valid |= OBD_MD_FLCTIME | OBD_MD_FLUID   |
 424                     OBD_MD_FLGID   | OBD_MD_FLTYPE  |
 425                     OBD_MD_FLMODE  | OBD_MD_FLNLINK | OBD_MD_FLFLAGS |
 426                     OBD_MD_FLATIME | OBD_MD_FLMTIME ;
 427
 428         if (!S_ISREG(attr->la_mode)) {
 429                 b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS | OBD_MD_FLRDEV;
 430         } else if (ma->ma_need & MA_LOV && ma->ma_lmm_size == 0) {
 431                 /* means no objects are allocated on osts. */
 432                 LASSERT(!(ma->ma_valid & MA_LOV));
 433                 /* just ignore blocks occupied by extend attributes on MDS */
 434                 b->blocks = 0;
 435                 /* if no object is allocated on osts, the size on mds is valid. b=22272 */
 436                 b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
 437         }
 438
 439         if (fid) {
 440                 b->fid1 = *fid;
 441                 b->valid |= OBD_MD_FLID;
 442
 443                 /* FIXME: these should be fixed when new igif ready.*/
 444                 b->ino  =  fid_oid(fid);       /* 1.6 compatibility */
 445                 b->generation = fid_ver(fid);  /* 1.6 compatibility */
 446                 b->valid |= OBD_MD_FLGENER;    /* 1.6 compatibility */
 447
 448                 CDEBUG(D_INODE, DFID": nlink=%d, mode=%o, size="LPU64"\n",
 449                                 PFID(fid), b->nlink, b->mode, b->size);
 450         }
 451
 452         if (info)
 453                 mdt_body_reverse_idmap(info, b);
 454
 455         if (b->valid & OBD_MD_FLSIZE)
 456                 CDEBUG(D_VFSTRACE, DFID": returning size %llu\n",
 457                        PFID(fid), (unsigned long long)b->size);
 458 }
 459
 460 static inline int mdt_body_has_lov(const struct lu_attr *la,
 461                                    const struct mdt_body *body)
 462 {
 463         return ((S_ISREG(la->la_mode) && (body->valid & OBD_MD_FLEASIZE)) ||
 464                 (S_ISDIR(la->la_mode) && (body->valid & OBD_MD_FLDIREA )) );
 465 }
 466
 467 void mdt_client_compatibility(struct mdt_thread_info *info)
 468 {
 469         struct mdt_body       *body;
 470         struct ptlrpc_request *req = mdt_info_req(info);
 471         struct obd_export     *exp = req->rq_export;
 472         struct md_attr        *ma = &info->mti_attr;
 473         struct lu_attr        *la = &ma->ma_attr;
 474         ENTRY;
 475
 476         if (exp->exp_connect_flags & OBD_CONNECT_LAYOUTLOCK)
 477                 /* the client can deal with 16-bit lmm_stripe_count */
 478                 RETURN_EXIT;
 479
 480         body = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 481
 482         if (!mdt_body_has_lov(la, body))
 483                 RETURN_EXIT;
 484
 485         /* now we have a reply with a lov for a client not compatible with the
 486          * layout lock so we have to clean the layout generation number */
 487         if (S_ISREG(la->la_mode))
 488                 ma->ma_lmm->lmm_layout_gen = 0;
 489         EXIT;
 490 }
 491
 492
 493 static int mdt_getattr_internal(struct mdt_thread_info *info,
 494                                 struct mdt_object *o, int ma_need)
 495 {
 496         struct md_object        *next = mdt_object_child(o);
 497         const struct mdt_body   *reqbody = info->mti_body;
 498         struct ptlrpc_request   *req = mdt_info_req(info);
 499         struct md_attr          *ma = &info->mti_attr;
 500         struct lu_attr          *la = &ma->ma_attr;
 501         struct req_capsule      *pill = info->mti_pill;
 502         const struct lu_env     *env = info->mti_env;
 503         struct mdt_body         *repbody;
 504         struct lu_buf           *buffer = &info->mti_buf;
 505         int                     rc;
 506         ENTRY;
 507
 508         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK))
 509                 RETURN(err_serious(-ENOMEM));
 510
 511         repbody = req_capsule_server_get(pill, &RMF_MDT_BODY);
 512
 513         ma->ma_valid = 0;
 514
 515         rc = mdt_object_exists(o);
 516         if (rc < 0) {
 517                 /* This object is located on remote node.*/
 518                 repbody->fid1 = *mdt_object_fid(o);
 519                 repbody->valid = OBD_MD_FLID | OBD_MD_MDS;
 520                 GOTO(out, rc = 0);
 521         }
 522
 523         buffer->lb_buf = req_capsule_server_get(pill, &RMF_MDT_MD);
 524         buffer->lb_len = req_capsule_get_size(pill, &RMF_MDT_MD, RCL_SERVER);
 525
 526         /* If it is dir object and client require MEA, then we got MEA */
 527         if (S_ISDIR(lu_object_attr(&next->mo_lu)) &&
 528             reqbody->valid & OBD_MD_MEA) {
 529                 /* Assumption: MDT_MD size is enough for lmv size. */
 530                 ma->ma_lmv = buffer->lb_buf;
 531                 ma->ma_lmv_size = buffer->lb_len;
 532                 ma->ma_need = MA_LMV | MA_INODE;
 533         } else {
 534                 ma->ma_lmm = buffer->lb_buf;
 535                 ma->ma_lmm_size = buffer->lb_len;
 536                 ma->ma_need = MA_LOV | MA_INODE;
 537         }
 538
 539         if (S_ISDIR(lu_object_attr(&next->mo_lu)) &&
 540             reqbody->valid & OBD_MD_FLDIREA  &&
 541             lustre_msg_get_opc(req->rq_reqmsg) == MDS_GETATTR) {
 542                 /* get default stripe info for this dir. */
 543                 ma->ma_need |= MA_LOV_DEF;
 544         }
 545         ma->ma_need |= ma_need;
 546         if (ma->ma_need & MA_SOM)
 547                 ma->ma_som = &info->mti_u.som.data;
 548
 549         rc = mo_attr_get(env, next, ma);
 550         if (unlikely(rc)) {
 551                 CERROR("getattr error for "DFID": %d\n",
 552                         PFID(mdt_object_fid(o)), rc);
 553                 RETURN(rc);
 554         }
 555
 556         if (likely(ma->ma_valid & MA_INODE))
 557                 mdt_pack_attr2body(info, repbody, la, mdt_object_fid(o));
 558         else
 559                 RETURN(-EFAULT);
 560
 561         if (mdt_body_has_lov(la, reqbody)) {
 562                 if (ma->ma_valid & MA_LOV) {
 563                         LASSERT(ma->ma_lmm_size);
 564                         mdt_dump_lmm(D_INFO, ma->ma_lmm);
 565                         repbody->eadatasize = ma->ma_lmm_size;
 566                         if (S_ISDIR(la->la_mode))
 567                                 repbody->valid |= OBD_MD_FLDIREA;
 568                         else
 569                                 repbody->valid |= OBD_MD_FLEASIZE;
 570                 }
 571                 if (ma->ma_valid & MA_LMV) {
 572                         LASSERT(S_ISDIR(la->la_mode));
 573                         repbody->eadatasize = ma->ma_lmv_size;
 574                         repbody->valid |= (OBD_MD_FLDIREA|OBD_MD_MEA);
 575                 }
 576         } else if (S_ISLNK(la->la_mode) &&
 577                    reqbody->valid & OBD_MD_LINKNAME) {
 578                 buffer->lb_buf = ma->ma_lmm;
 579                 /* eadatasize from client includes NULL-terminator, so
 580                  * there is no need to read it */
 581                 buffer->lb_len = reqbody->eadatasize - 1;
 582                 rc = mo_readlink(env, next, buffer);
 583                 if (unlikely(rc <= 0)) {
 584                         CERROR("readlink failed: %d\n", rc);
 585                         rc = -EFAULT;
 586                 } else {
 587                         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_READLINK_EPROTO))
 588                                  rc -= 2;
 589                         repbody->valid |= OBD_MD_LINKNAME;
 590                         /* we need to report back size with NULL-terminator
 591                          * because client expects that */
 592                         repbody->eadatasize = rc + 1;
 593                         if (repbody->eadatasize != reqbody->eadatasize)
 594                                 CERROR("Read shorter link %d than expected "
 595                                        "%d\n", rc, reqbody->eadatasize - 1);
 596                         /* NULL terminate */
 597                         ((char*)ma->ma_lmm)[rc] = 0;
 598                         CDEBUG(D_INODE, "symlink dest %s, len = %d\n",
 599                                (char*)ma->ma_lmm, rc);
 600                         rc = 0;
 601                 }
 602         }
 603
 604         if (reqbody->valid & OBD_MD_FLMODEASIZE) {
 605                 repbody->max_cookiesize = info->mti_mdt->mdt_max_cookiesize;
 606                 repbody->max_mdsize = info->mti_mdt->mdt_max_mdsize;
 607                 repbody->valid |= OBD_MD_FLMODEASIZE;
 608                 CDEBUG(D_INODE, "I am going to change the MAX_MD_SIZE & "
 609                        "MAX_COOKIE to : %d:%d\n", repbody->max_mdsize,
 610                        repbody->max_cookiesize);
 611         }
 612
 613         if (exp_connect_rmtclient(info->mti_exp) &&
 614             reqbody->valid & OBD_MD_FLRMTPERM) {
 615                 void *buf = req_capsule_server_get(pill, &RMF_ACL);
 616
 617                 /* mdt_getattr_lock only */
 618                 rc = mdt_pack_remote_perm(info, o, buf);
 619                 if (rc) {
 620                         repbody->valid &= ~OBD_MD_FLRMTPERM;
 621                         repbody->aclsize = 0;
 622                         RETURN(rc);
 623                 } else {
 624                         repbody->valid |= OBD_MD_FLRMTPERM;
 625                         repbody->aclsize = sizeof(struct mdt_remote_perm);
 626                 }
 627         }
 628 #ifdef CONFIG_FS_POSIX_ACL
 629         else if ((req->rq_export->exp_connect_flags & OBD_CONNECT_ACL) &&
 630                  (reqbody->valid & OBD_MD_FLACL)) {
 631                 buffer->lb_buf = req_capsule_server_get(pill, &RMF_ACL);
 632                 buffer->lb_len = req_capsule_get_size(pill,
 633                                                       &RMF_ACL, RCL_SERVER);
 634                 if (buffer->lb_len > 0) {
 635                         rc = mo_xattr_get(env, next, buffer,
 636                                           XATTR_NAME_ACL_ACCESS);
 637                         if (rc < 0) {
 638                                 if (rc == -ENODATA) {
 639                                         repbody->aclsize = 0;
 640                                         repbody->valid |= OBD_MD_FLACL;
 641                                         rc = 0;
 642                                 } else if (rc == -EOPNOTSUPP) {
 643                                         rc = 0;
 644                                 } else {
 645                                         CERROR("got acl size: %d\n", rc);
 646                                 }
 647                         } else {
 648                                 repbody->aclsize = rc;
 649                                 repbody->valid |= OBD_MD_FLACL;
 650                                 rc = 0;
 651                         }
 652                 }
 653         }
 654 #endif
 655
 656         if (reqbody->valid & OBD_MD_FLMDSCAPA &&
 657             info->mti_mdt->mdt_opts.mo_mds_capa &&
 658             info->mti_exp->exp_connect_flags & OBD_CONNECT_MDS_CAPA) {
 659                 struct lustre_capa *capa;
 660
 661                 capa = req_capsule_server_get(pill, &RMF_CAPA1);
 662                 LASSERT(capa);
 663                 capa->lc_opc = CAPA_OPC_MDS_DEFAULT;
 664                 rc = mo_capa_get(env, next, capa, 0);
 665                 if (rc)
 666                         RETURN(rc);
 667                 repbody->valid |= OBD_MD_FLMDSCAPA;
 668         }
 669
 670 out:
 671         if (rc == 0)
 672                 mdt_counter_incr(req, LPROC_MDT_GETATTR);
 673
 674         RETURN(rc);
 675 }
 676
 677 static int mdt_renew_capa(struct mdt_thread_info *info)
 678 {
 679         struct mdt_object  *obj = info->mti_object;
 680         struct mdt_body    *body;
 681         struct lustre_capa *capa, *c;
 682         int rc;
 683         ENTRY;
 684
 685         /* if object doesn't exist, or server has disabled capability,
 686          * return directly, client will find body->valid OBD_MD_FLOSSCAPA
 687          * flag not set.
 688          */
 689         if (!obj || !info->mti_mdt->mdt_opts.mo_oss_capa ||
 690             !(info->mti_exp->exp_connect_flags & OBD_CONNECT_OSS_CAPA))
 691                 RETURN(0);
 692
 693         body = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 694         LASSERT(body != NULL);
 695
 696         c = req_capsule_client_get(info->mti_pill, &RMF_CAPA1);
 697         LASSERT(c);
 698
 699         capa = req_capsule_server_get(info->mti_pill, &RMF_CAPA2);
 700         LASSERT(capa);
 701
 702         *capa = *c;
 703         rc = mo_capa_get(info->mti_env, mdt_object_child(obj), capa, 1);
 704         if (rc == 0)
 705                 body->valid |= OBD_MD_FLOSSCAPA;
 706         RETURN(rc);
 707 }
 708
 709 static int mdt_getattr(struct mdt_thread_info *info)
 710 {
 711         struct mdt_object       *obj = info->mti_object;
 712         struct req_capsule      *pill = info->mti_pill;
 713         struct mdt_body         *reqbody;
 714         struct mdt_body         *repbody;
 715         mode_t                   mode;
 716         int rc, rc2;
 717         ENTRY;
 718
 719         reqbody = req_capsule_client_get(pill, &RMF_MDT_BODY);
 720         LASSERT(reqbody);
 721
 722         if (reqbody->valid & OBD_MD_FLOSSCAPA) {
 723                 rc = req_capsule_server_pack(pill);
 724                 if (unlikely(rc))
 725                         RETURN(err_serious(rc));
 726                 rc = mdt_renew_capa(info);
 727                 GOTO(out_shrink, rc);
 728         }
 729
 730         LASSERT(obj != NULL);
 731         LASSERT(lu_object_assert_exists(&obj->mot_obj.mo_lu));
 732
 733         mode = lu_object_attr(&obj->mot_obj.mo_lu);
 734
 735         /* old clients may not report needed easize, use max value then */
 736         req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER,
 737                              reqbody->eadatasize == 0 ?
 738                              info->mti_mdt->mdt_max_mdsize :
 739                              reqbody->eadatasize);
 740
 741         rc = req_capsule_server_pack(pill);
 742         if (unlikely(rc != 0))
 743                 RETURN(err_serious(rc));
 744
 745         repbody = req_capsule_server_get(pill, &RMF_MDT_BODY);
 746         LASSERT(repbody != NULL);
 747         repbody->eadatasize = 0;
 748         repbody->aclsize = 0;
 749
 750         if (reqbody->valid & OBD_MD_FLRMTPERM)
 751                 rc = mdt_init_ucred(info, reqbody);
 752         else
 753                 rc = mdt_check_ucred(info);
 754         if (unlikely(rc))
 755                 GOTO(out_shrink, rc);
 756
 757         info->mti_spec.sp_ck_split = !!(reqbody->valid & OBD_MD_FLCKSPLIT);
 758         info->mti_cross_ref = !!(reqbody->valid & OBD_MD_FLCROSSREF);
 759
 760         /*
 761          * Don't check capability at all, because rename might getattr for
 762          * remote obj, and at that time no capability is available.
 763          */
 764         mdt_set_capainfo(info, 1, &reqbody->fid1, BYPASS_CAPA);
 765         rc = mdt_getattr_internal(info, obj, 0);
 766         if (reqbody->valid & OBD_MD_FLRMTPERM)
 767                 mdt_exit_ucred(info);
 768         EXIT;
 769 out_shrink:
 770         mdt_client_compatibility(info);
 771         rc2 = mdt_fix_reply(info);
 772         if (rc == 0)
 773                 rc = rc2;
 774         return rc;
 775 }
 776
 777 static int mdt_is_subdir(struct mdt_thread_info *info)
 778 {
 779         struct mdt_object     *o = info->mti_object;
 780         struct req_capsule    *pill = info->mti_pill;
 781         const struct mdt_body *body = info->mti_body;
 782         struct mdt_body       *repbody;
 783         int                    rc;
 784         ENTRY;
 785
 786         LASSERT(o != NULL);
 787
 788         repbody = req_capsule_server_get(pill, &RMF_MDT_BODY);
 789
 790         /*
 791          * We save last checked parent fid to @repbody->fid1 for remote
 792          * directory case.
 793          */
 794         LASSERT(fid_is_sane(&body->fid2));
 795         LASSERT(mdt_object_exists(o) > 0);
 796         rc = mdo_is_subdir(info->mti_env, mdt_object_child(o),
 797                            &body->fid2, &repbody->fid1);
 798         if (rc == 0 || rc == -EREMOTE)
 799                 repbody->valid |= OBD_MD_FLID;
 800
 801         RETURN(rc);
 802 }
 803
 804 static int mdt_raw_lookup(struct mdt_thread_info *info,
 805                           struct mdt_object *parent,
 806                           const struct lu_name *lname,
 807                           struct ldlm_reply *ldlm_rep)
 808 {
 809         struct md_object *next = mdt_object_child(info->mti_object);
 810         const struct mdt_body *reqbody = info->mti_body;
 811         struct lu_fid *child_fid = &info->mti_tmp_fid1;
 812         struct mdt_body *repbody;
 813         int rc;
 814         ENTRY;
 815
 816         if (reqbody->valid != OBD_MD_FLID)
 817                 RETURN(0);
 818
 819         LASSERT(!info->mti_cross_ref);
 820
 821         /* Only got the fid of this obj by name */
 822         fid_zero(child_fid);
 823         rc = mdo_lookup(info->mti_env, next, lname, child_fid,
 824                         &info->mti_spec);
 825 #if 0
 826         /* XXX is raw_lookup possible as intent operation? */
 827         if (rc != 0) {
 828                 if (rc == -ENOENT)
 829                         mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG);
 830                 RETURN(rc);
 831         } else
 832                 mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS);
 833
 834         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 835 #endif
 836         if (rc == 0) {
 837                 repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 838                 repbody->fid1 = *child_fid;
 839                 repbody->valid = OBD_MD_FLID;
 840         }
 841         RETURN(1);
 842 }
 843
 844 /*
 845  * UPDATE lock should be taken against parent, and be release before exit;
 846  * child_bits lock should be taken against child, and be returned back:
 847  *            (1)normal request should release the child lock;
 848  *            (2)intent request will grant the lock to client.
 849  */
 850 static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 851                                  struct mdt_lock_handle *lhc,
 852                                  __u64 child_bits,
 853                                  struct ldlm_reply *ldlm_rep)
 854 {
 855         struct ptlrpc_request  *req       = mdt_info_req(info);
 856         struct mdt_body        *reqbody   = NULL;
 857         struct mdt_object      *parent    = info->mti_object;
 858         struct mdt_object      *child;
 859         struct md_object       *next      = mdt_object_child(parent);
 860         struct lu_fid          *child_fid = &info->mti_tmp_fid1;
 861         struct lu_name         *lname     = NULL;
 862         const char             *name      = NULL;
 863         int                     namelen   = 0;
 864         struct mdt_lock_handle *lhp       = NULL;
 865         struct ldlm_lock       *lock;
 866         struct ldlm_res_id     *res_id;
 867         int                     is_resent;
 868         int                     ma_need = 0;
 869         int                     rc;
 870
 871         ENTRY;
 872
 873         is_resent = lustre_handle_is_used(&lhc->mlh_reg_lh);
 874         LASSERT(ergo(is_resent,
 875                      lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT));
 876
 877         LASSERT(parent != NULL);
 878         name = req_capsule_client_get(info->mti_pill, &RMF_NAME);
 879         if (name == NULL)
 880                 RETURN(err_serious(-EFAULT));
 881
 882         namelen = req_capsule_get_size(info->mti_pill, &RMF_NAME,
 883                                        RCL_CLIENT) - 1;
 884         if (!info->mti_cross_ref) {
 885                 /*
 886                  * XXX: Check for "namelen == 0" is for getattr by fid
 887                  * (OBD_CONNECT_ATTRFID), otherwise do not allow empty name,
 888                  * that is the name must contain at least one character and
 889                  * the terminating '\0'
 890                  */
 891                 if (namelen == 0) {
 892                         reqbody = req_capsule_client_get(info->mti_pill,
 893                                                          &RMF_MDT_BODY);
 894                         if (unlikely(reqbody == NULL))
 895                                 RETURN(err_serious(-EFAULT));
 896
 897                         if (unlikely(!fid_is_sane(&reqbody->fid2)))
 898                                 RETURN(err_serious(-EINVAL));
 899
 900                         name = NULL;
 901                         CDEBUG(D_INODE, "getattr with lock for "DFID"/"DFID", "
 902                                "ldlm_rep = %p\n",
 903                                PFID(mdt_object_fid(parent)),
 904                                PFID(&reqbody->fid2), ldlm_rep);
 905                 } else {
 906                         lname = mdt_name(info->mti_env, (char *)name, namelen);
 907                         CDEBUG(D_INODE, "getattr with lock for "DFID"/%s, "
 908                                "ldlm_rep = %p\n", PFID(mdt_object_fid(parent)),
 909                                name, ldlm_rep);
 910                 }
 911         }
 912         mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_EXECD);
 913
 914         rc = mdt_object_exists(parent);
 915         if (unlikely(rc == 0)) {
 916                 LU_OBJECT_DEBUG(D_INODE, info->mti_env,
 917                                 &parent->mot_obj.mo_lu,
 918                                 "Parent doesn't exist!\n");
 919                 RETURN(-ESTALE);
 920         } else if (!info->mti_cross_ref) {
 921                 LASSERTF(rc > 0, "Parent "DFID" is on remote server\n",
 922                          PFID(mdt_object_fid(parent)));
 923         }
 924         if (lname) {
 925                 rc = mdt_raw_lookup(info, parent, lname, ldlm_rep);
 926                 if (rc != 0) {
 927                         if (rc > 0)
 928                                 rc = 0;
 929                         RETURN(rc);
 930                 }
 931         }
 932
 933         if (info->mti_cross_ref) {
 934                 /* Only getattr on the child. Parent is on another node. */
 935                 mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS);
 936                 child = parent;
 937                 CDEBUG(D_INODE, "partial getattr_name child_fid = "DFID", "
 938                        "ldlm_rep=%p\n", PFID(mdt_object_fid(child)), ldlm_rep);
 939
 940                 if (is_resent) {
 941                         /* Do not take lock for resent case. */
 942                         lock = ldlm_handle2lock(&lhc->mlh_reg_lh);
 943                         LASSERTF(lock != NULL, "Invalid lock handle "LPX64"\n",
 944                                  lhc->mlh_reg_lh.cookie);
 945                         LASSERT(fid_res_name_eq(mdt_object_fid(child),
 946                                                 &lock->l_resource->lr_name));
 947                         LDLM_LOCK_PUT(lock);
 948                         rc = 0;
 949                 } else {
 950                         mdt_lock_handle_init(lhc);
 951                         mdt_lock_reg_init(lhc, LCK_PR);
 952
 953                         /*
 954                          * Object's name is on another MDS, no lookup lock is
 955                          * needed here but update is.
 956                          */
 957                         child_bits &= ~MDS_INODELOCK_LOOKUP;
 958                         child_bits |= MDS_INODELOCK_UPDATE;
 959
 960                         rc = mdt_object_lock(info, child, lhc, child_bits,
 961                                              MDT_LOCAL_LOCK);
 962                 }
 963                 if (rc == 0) {
 964                         /* Finally, we can get attr for child. */
 965                         mdt_set_capainfo(info, 0, mdt_object_fid(child),
 966                                          BYPASS_CAPA);
 967                         rc = mdt_getattr_internal(info, child, 0);
 968                         if (unlikely(rc != 0))
 969                                 mdt_object_unlock(info, child, lhc, 1);
 970                 }
 971                 RETURN(rc);
 972         }
 973
 974         if (lname) {
 975                 /* step 1: lock parent only if parent is a directory */
 976                 if (S_ISDIR(lu_object_attr(&parent->mot_obj.mo_lu))) {
 977                         lhp = &info->mti_lh[MDT_LH_PARENT];
 978                         mdt_lock_pdo_init(lhp, LCK_PR, name, namelen);
 979                         rc = mdt_object_lock(info, parent, lhp,
 980                                              MDS_INODELOCK_UPDATE,
 981                                              MDT_LOCAL_LOCK);
 982                         if (unlikely(rc != 0))
 983                                 RETURN(rc);
 984                 }
 985
 986                 /* step 2: lookup child's fid by name */
 987                 fid_zero(child_fid);
 988                 rc = mdo_lookup(info->mti_env, next, lname, child_fid,
 989                                 &info->mti_spec);
 990
 991                 if (rc != 0) {
 992                         if (rc == -ENOENT)
 993                                 mdt_set_disposition(info, ldlm_rep,
 994                                                     DISP_LOOKUP_NEG);
 995                         GOTO(out_parent, rc);
 996                 } else
 997                         mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS);
 998         } else {
 999                 *child_fid = reqbody->fid2;
1000                 mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS);
1001         }
1002
1003         /*
1004          *step 3: find the child object by fid & lock it.
1005          *        regardless if it is local or remote.
1006          */
1007         child = mdt_object_find(info->mti_env, info->mti_mdt, child_fid);
1008
1009         if (unlikely(IS_ERR(child)))
1010                 GOTO(out_parent, rc = PTR_ERR(child));
1011         if (is_resent) {
1012                 /* Do not take lock for resent case. */
1013                 lock = ldlm_handle2lock(&lhc->mlh_reg_lh);
1014                 LASSERTF(lock != NULL, "Invalid lock handle "LPX64"\n",
1015                          lhc->mlh_reg_lh.cookie);
1016
1017                 res_id = &lock->l_resource->lr_name;
1018                 if (!fid_res_name_eq(mdt_object_fid(child),
1019                                     &lock->l_resource->lr_name)) {
1020                          LASSERTF(fid_res_name_eq(mdt_object_fid(parent),
1021                                                  &lock->l_resource->lr_name),
1022                                  "Lock res_id: %lu/%lu/%lu, Fid: "DFID".\n",
1023                                  (unsigned long)res_id->name[0],
1024                                  (unsigned long)res_id->name[1],
1025                                  (unsigned long)res_id->name[2],
1026                                  PFID(mdt_object_fid(parent)));
1027                           CWARN("Although resent, but still not get child lock"
1028                                 "parent:"DFID" child:"DFID"\n",
1029                                 PFID(mdt_object_fid(parent)),
1030                                 PFID(mdt_object_fid(child)));
1031                           lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
1032                           LDLM_LOCK_PUT(lock);
1033                           GOTO(relock, 0);
1034                 }
1035                 LDLM_LOCK_PUT(lock);
1036                 rc = 0;
1037         } else {
1038 relock:
1039                 OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RESEND, obd_timeout*2);
1040                 mdt_lock_handle_init(lhc);
1041                 if (child_bits == MDS_INODELOCK_LAYOUT)
1042                         mdt_lock_reg_init(lhc, LCK_CR);
1043                 else
1044                         mdt_lock_reg_init(lhc, LCK_PR);
1045
1046                 if (mdt_object_exists(child) == 0) {
1047                         LU_OBJECT_DEBUG(D_INODE, info->mti_env,
1048                                         &child->mot_obj.mo_lu,
1049                                         "Object doesn't exist!\n");
1050                         GOTO(out_child, rc = -ENOENT);
1051                 }
1052
1053                 if (!(child_bits & MDS_INODELOCK_UPDATE)) {
1054                         struct md_attr *ma = &info->mti_attr;
1055
1056                         ma->ma_valid = 0;
1057                         ma->ma_need = MA_INODE;
1058                         rc = mo_attr_get(info->mti_env,
1059                                          mdt_object_child(child), ma);
1060                         if (unlikely(rc != 0))
1061                                 GOTO(out_child, rc);
1062
1063                         /* layout lock is used only on regular files */
1064                         if ((ma->ma_valid & MA_INODE) &&
1065                             (ma->ma_attr.la_valid & LA_MODE) &&
1066                             !S_ISREG(ma->ma_attr.la_mode))
1067                                 child_bits &= ~MDS_INODELOCK_LAYOUT;
1068
1069                         /* If the file has not been changed for some time, we
1070                          * return not only a LOOKUP lock, but also an UPDATE
1071                          * lock and this might save us RPC on later STAT. For
1072                          * directories, it also let negative dentry starts
1073                          * working for this dir. */
1074                         if (ma->ma_valid & MA_INODE &&
1075                             ma->ma_attr.la_valid & LA_CTIME &&
1076                             info->mti_mdt->mdt_namespace->ns_ctime_age_limit +
1077                                 ma->ma_attr.la_ctime < cfs_time_current_sec())
1078                                 child_bits |= MDS_INODELOCK_UPDATE;
1079                 }
1080
1081                 rc = mdt_object_lock(info, child, lhc, child_bits,
1082                                      MDT_CROSS_LOCK);
1083
1084                 if (unlikely(rc != 0))
1085                         GOTO(out_child, rc);
1086         }
1087
1088         lock = ldlm_handle2lock(&lhc->mlh_reg_lh);
1089         /* Get MA_SOM attributes if update lock is given. */
1090         if (lock &&
1091             lock->l_policy_data.l_inodebits.bits & MDS_INODELOCK_UPDATE &&
1092             S_ISREG(lu_object_attr(&mdt_object_child(child)->mo_lu)))
1093                 ma_need = MA_SOM;
1094
1095         /* finally, we can get attr for child. */
1096         mdt_set_capainfo(info, 1, child_fid, BYPASS_CAPA);
1097         rc = mdt_getattr_internal(info, child, ma_need);
1098         if (unlikely(rc != 0)) {
1099                 mdt_object_unlock(info, child, lhc, 1);
1100         } else if (lock) {
1101                 /* Debugging code. */
1102                 res_id = &lock->l_resource->lr_name;
1103                 LDLM_DEBUG(lock, "Returning lock to client");
1104                 LASSERTF(fid_res_name_eq(mdt_object_fid(child),
1105                                          &lock->l_resource->lr_name),
1106                          "Lock res_id: %lu/%lu/%lu, Fid: "DFID".\n",
1107                          (unsigned long)res_id->name[0],
1108                          (unsigned long)res_id->name[1],
1109                          (unsigned long)res_id->name[2],
1110                          PFID(mdt_object_fid(child)));
1111                 mdt_pack_size2body(info, child);
1112         }
1113         if (lock)
1114                 LDLM_LOCK_PUT(lock);
1115
1116         EXIT;
1117 out_child:
1118         mdt_object_put(info->mti_env, child);
1119 out_parent:
1120         if (lhp)
1121                 mdt_object_unlock(info, parent, lhp, 1);
1122         return rc;
1123 }
1124
1125 /* normal handler: should release the child lock */
1126 static int mdt_getattr_name(struct mdt_thread_info *info)
1127 {
1128         struct mdt_lock_handle *lhc = &info->mti_lh[MDT_LH_CHILD];
1129         struct mdt_body        *reqbody;
1130         struct mdt_body        *repbody;
1131         int rc, rc2;
1132         ENTRY;
1133
1134         reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY);
1135         LASSERT(reqbody != NULL);
1136         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
1137         LASSERT(repbody != NULL);
1138
1139         info->mti_spec.sp_ck_split = !!(reqbody->valid & OBD_MD_FLCKSPLIT);
1140         info->mti_cross_ref = !!(reqbody->valid & OBD_MD_FLCROSSREF);
1141         repbody->eadatasize = 0;
1142         repbody->aclsize = 0;
1143
1144         rc = mdt_init_ucred(info, reqbody);
1145         if (unlikely(rc))
1146                 GOTO(out_shrink, rc);
1147
1148         rc = mdt_getattr_name_lock(info, lhc, MDS_INODELOCK_UPDATE, NULL);
1149         if (lustre_handle_is_used(&lhc->mlh_reg_lh)) {
1150                 ldlm_lock_decref(&lhc->mlh_reg_lh, lhc->mlh_reg_mode);
1151                 lhc->mlh_reg_lh.cookie = 0;
1152         }
1153         mdt_exit_ucred(info);
1154         EXIT;
1155 out_shrink:
1156         mdt_client_compatibility(info);
1157         rc2 = mdt_fix_reply(info);
1158         if (rc == 0)
1159                 rc = rc2;
1160         return rc;
1161 }
1162
1163 static const struct lu_device_operations mdt_lu_ops;
1164
1165 static int lu_device_is_mdt(struct lu_device *d)
1166 {
1167         return ergo(d != NULL && d->ld_ops != NULL, d->ld_ops == &mdt_lu_ops);
1168 }
1169
1170 static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
1171                          void *karg, void *uarg);
1172
1173 static int mdt_set_info(struct mdt_thread_info *info)
1174 {
1175         struct ptlrpc_request *req = mdt_info_req(info);
1176         char *key;
1177         void *val;
1178         int keylen, vallen, rc = 0;
1179         ENTRY;
1180
1181         rc = req_capsule_server_pack(info->mti_pill);
1182         if (rc)
1183                 RETURN(rc);
1184
1185         key = req_capsule_client_get(info->mti_pill, &RMF_SETINFO_KEY);
1186         if (key == NULL) {
1187                 DEBUG_REQ(D_HA, req, "no set_info key");
1188                 RETURN(-EFAULT);
1189         }
1190
1191         keylen = req_capsule_get_size(info->mti_pill, &RMF_SETINFO_KEY,
1192                                       RCL_CLIENT);
1193
1194         val = req_capsule_client_get(info->mti_pill, &RMF_SETINFO_VAL);
1195         if (val == NULL) {
1196                 DEBUG_REQ(D_HA, req, "no set_info val");
1197                 RETURN(-EFAULT);
1198         }
1199
1200         vallen = req_capsule_get_size(info->mti_pill, &RMF_SETINFO_VAL,
1201                                       RCL_CLIENT);
1202
1203         /* Swab any part of val you need to here */
1204         if (KEY_IS(KEY_READ_ONLY)) {
1205                 req->rq_status = 0;
1206                 lustre_msg_set_status(req->rq_repmsg, 0);
1207
1208                 cfs_spin_lock(&req->rq_export->exp_lock);
1209                 if (*(__u32 *)val)
1210                         req->rq_export->exp_connect_flags |= OBD_CONNECT_RDONLY;
1211                 else
1212                         req->rq_export->exp_connect_flags &=~OBD_CONNECT_RDONLY;
1213                 cfs_spin_unlock(&req->rq_export->exp_lock);
1214
1215         } else if (KEY_IS(KEY_CHANGELOG_CLEAR)) {
1216                 struct changelog_setinfo *cs =
1217                         (struct changelog_setinfo *)val;
1218                 if (vallen != sizeof(*cs)) {
1219                         CERROR("Bad changelog_clear setinfo size %d\n", vallen);
1220                         RETURN(-EINVAL);
1221                 }
1222                 if (ptlrpc_req_need_swab(req)) {
1223                         __swab64s(&cs->cs_recno);
1224                         __swab32s(&cs->cs_id);
1225                 }
1226
1227                 rc = mdt_iocontrol(OBD_IOC_CHANGELOG_CLEAR, info->mti_exp,
1228                                    vallen, val, NULL);
1229                 lustre_msg_set_status(req->rq_repmsg, rc);
1230
1231         } else {
1232                 RETURN(-EINVAL);
1233         }
1234         RETURN(0);
1235 }
1236
1237 static int mdt_connect(struct mdt_thread_info *info)
1238 {
1239         int rc;
1240         struct ptlrpc_request *req;
1241
1242         req = mdt_info_req(info);
1243         rc = target_handle_connect(req);
1244         if (rc == 0) {
1245                 LASSERT(req->rq_export != NULL);
1246                 info->mti_mdt = mdt_dev(req->rq_export->exp_obd->obd_lu_dev);
1247                 rc = mdt_init_sec_level(info);
1248                 if (rc == 0)
1249                         rc = mdt_init_idmap(info);
1250                 if (rc != 0)
1251                         obd_disconnect(class_export_get(req->rq_export));
1252         } else {
1253                 rc = err_serious(rc);
1254         }
1255         return rc;
1256 }
1257
1258 static int mdt_disconnect(struct mdt_thread_info *info)
1259 {
1260         int rc;
1261         ENTRY;
1262
1263         rc = target_handle_disconnect(mdt_info_req(info));
1264         if (rc)
1265                 rc = err_serious(rc);
1266         RETURN(rc);
1267 }
1268
1269 static int mdt_sendpage(struct mdt_thread_info *info,
1270                         struct lu_rdpg *rdpg, int nob)
1271 {
1272         struct ptlrpc_request   *req = mdt_info_req(info);
1273         struct obd_export       *exp = req->rq_export;
1274         struct ptlrpc_bulk_desc *desc;
1275         struct l_wait_info      *lwi = &info->mti_u.rdpg.mti_wait_info;
1276         int                      tmpcount;
1277         int                      tmpsize;
1278         int                      i;
1279         int                      rc;
1280         ENTRY;
1281
1282         desc = ptlrpc_prep_bulk_exp(req, rdpg->rp_npages, BULK_PUT_SOURCE,
1283                                     MDS_BULK_PORTAL);
1284         if (desc == NULL)
1285                 RETURN(-ENOMEM);
1286
1287         for (i = 0, tmpcount = nob;
1288                 i < rdpg->rp_npages && tmpcount > 0; i++, tmpcount -= tmpsize) {
1289                 tmpsize = min_t(int, tmpcount, CFS_PAGE_SIZE);
1290                 ptlrpc_prep_bulk_page(desc, rdpg->rp_pages[i], 0, tmpsize);
1291         }
1292
1293         LASSERT(desc->bd_nob == nob);
1294         rc = target_bulk_io(exp, desc, lwi);
1295         ptlrpc_free_bulk(desc);
1296         RETURN(rc);
1297 }
1298
1299 #ifdef HAVE_SPLIT_SUPPORT
1300 /*
1301  * Retrieve dir entry from the page and insert it to the slave object, actually,
1302  * this should be in osd layer, but since it will not in the final product, so
1303  * just do it here and do not define more moo api anymore for this.
1304  */
1305 static int mdt_write_dir_page(struct mdt_thread_info *info, struct page *page,
1306                               int size)
1307 {
1308         struct mdt_object *object = info->mti_object;
1309         struct lu_fid *lf = &info->mti_tmp_fid2;
1310         struct md_attr *ma = &info->mti_attr;
1311         struct lu_dirpage *dp;
1312         struct lu_dirent *ent;
1313         int rc = 0, offset = 0;
1314         ENTRY;
1315
1316         /* Make sure we have at least one entry. */
1317         if (size == 0)
1318                 RETURN(-EINVAL);
1319
1320         /*
1321          * Disable trans for this name insert, since it will include many trans
1322          * for this.
1323          */
1324         info->mti_no_need_trans = 1;
1325         /*
1326          * When write_dir_page, no need update parent's ctime,
1327          * and no permission check for name_insert.
1328          */
1329         ma->ma_attr.la_ctime = 0;
1330         ma->ma_attr.la_valid = LA_MODE;
1331         ma->ma_valid = MA_INODE;
1332
1333         cfs_kmap(page);
1334         dp = page_address(page);
1335         offset = (int)((__u32)lu_dirent_start(dp) - (__u32)dp);
1336
1337         for (ent = lu_dirent_start(dp); ent != NULL;
1338              ent = lu_dirent_next(ent)) {
1339                 struct lu_name *lname;
1340                 char *name;
1341
1342                 if (le16_to_cpu(ent->lde_namelen) == 0)
1343                         continue;
1344
1345                 fid_le_to_cpu(lf, &ent->lde_fid);
1346                 if (le64_to_cpu(ent->lde_hash) & MAX_HASH_HIGHEST_BIT)
1347                         ma->ma_attr.la_mode = S_IFDIR;
1348                 else
1349                         ma->ma_attr.la_mode = 0;
1350                 OBD_ALLOC(name, le16_to_cpu(ent->lde_namelen) + 1);
1351                 if (name == NULL)
1352                         GOTO(out, rc = -ENOMEM);
1353
1354                 memcpy(name, ent->lde_name, le16_to_cpu(ent->lde_namelen));
1355                 lname = mdt_name(info->mti_env, name,
1356                                  le16_to_cpu(ent->lde_namelen));
1357                 ma->ma_attr_flags |= (MDS_PERM_BYPASS | MDS_QUOTA_IGNORE);
1358                 rc = mdo_name_insert(info->mti_env,
1359                                      md_object_next(&object->mot_obj),
1360                                      lname, lf, ma);
1361                 OBD_FREE(name, le16_to_cpu(ent->lde_namelen) + 1);
1362                 if (rc) {
1363                         CERROR("Can't insert %*.*s, rc %d\n",
1364                                le16_to_cpu(ent->lde_namelen),
1365                                le16_to_cpu(ent->lde_namelen),
1366                                ent->lde_name, rc);
1367                         GOTO(out, rc);
1368                 }
1369
1370                 offset += lu_dirent_size(ent);
1371                 if (offset >= size)
1372                         break;
1373         }
1374         EXIT;
1375 out:
1376         cfs_kunmap(page);
1377         return rc;
1378 }
1379
1380 static int mdt_bulk_timeout(void *data)
1381 {
1382         ENTRY;
1383
1384         CERROR("mdt bulk transfer timeout \n");
1385
1386         RETURN(1);
1387 }
1388
1389 static int mdt_writepage(struct mdt_thread_info *info)
1390 {
1391         struct ptlrpc_request   *req = mdt_info_req(info);
1392         struct mdt_body         *reqbody;
1393         struct l_wait_info      *lwi;
1394         struct ptlrpc_bulk_desc *desc;
1395         struct page             *page;
1396         int                rc;
1397         ENTRY;
1398
1399
1400         reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY);
1401         if (reqbody == NULL)
1402                 RETURN(err_serious(-EFAULT));
1403
1404         desc = ptlrpc_prep_bulk_exp(req, 1, BULK_GET_SINK, MDS_BULK_PORTAL);
1405         if (desc == NULL)
1406                 RETURN(err_serious(-ENOMEM));
1407
1408         /* allocate the page for the desc */
1409         page = cfs_alloc_page(CFS_ALLOC_STD);
1410         if (page == NULL)
1411                 GOTO(desc_cleanup, rc = -ENOMEM);
1412
1413         CDEBUG(D_INFO, "Received page offset %d size %d \n",
1414                (int)reqbody->size, (int)reqbody->nlink);
1415
1416         ptlrpc_prep_bulk_page(desc, page, (int)reqbody->size,
1417                               (int)reqbody->nlink);
1418
1419         rc = sptlrpc_svc_prep_bulk(req, desc);
1420         if (rc != 0)
1421                 GOTO(cleanup_page, rc);
1422         /*
1423          * Check if client was evicted while we were doing i/o before touching
1424          * network.
1425          */
1426         OBD_ALLOC_PTR(lwi);
1427         if (!lwi)
1428                 GOTO(cleanup_page, rc = -ENOMEM);
1429
1430         if (desc->bd_export->exp_failed)
1431                 rc = -ENOTCONN;
1432         else
1433                 rc = ptlrpc_start_bulk_transfer (desc);
1434         if (rc == 0) {
1435                 *lwi = LWI_TIMEOUT_INTERVAL(obd_timeout * CFS_HZ / 4, CFS_HZ,
1436                                             mdt_bulk_timeout, desc);
1437                 rc = l_wait_event(desc->bd_waitq, !ptlrpc_bulk_active(desc) ||
1438                                   desc->bd_export->exp_failed, lwi);
1439                 LASSERT(rc == 0 || rc == -ETIMEDOUT);
1440                 if (rc == -ETIMEDOUT) {
1441                         DEBUG_REQ(D_ERROR, req, "timeout on bulk GET");
1442                         ptlrpc_abort_bulk(desc);
1443                 } else if (desc->bd_export->exp_failed) {
1444                         DEBUG_REQ(D_ERROR, req, "Eviction on bulk GET");
1445                         rc = -ENOTCONN;
1446                         ptlrpc_abort_bulk(desc);
1447                 } else if (!desc->bd_success ||
1448                            desc->bd_nob_transferred != desc->bd_nob) {
1449                         DEBUG_REQ(D_ERROR, req, "%s bulk GET %d(%d)",
1450                                   desc->bd_success ?
1451                                   "truncated" : "network error on",
1452                                   desc->bd_nob_transferred, desc->bd_nob);
1453                         /* XXX should this be a different errno? */
1454                         rc = -ETIMEDOUT;
1455                 }
1456         } else {
1457                 DEBUG_REQ(D_ERROR, req, "ptlrpc_bulk_get failed: rc %d", rc);
1458         }
1459         if (rc)
1460                 GOTO(cleanup_lwi, rc);
1461         rc = mdt_write_dir_page(info, page, reqbody->nlink);
1462
1463 cleanup_lwi:
1464         OBD_FREE_PTR(lwi);
1465 cleanup_page:
1466         cfs_free_page(page);
1467 desc_cleanup:
1468         ptlrpc_free_bulk(desc);
1469         RETURN(rc);
1470 }
1471 #endif
1472
1473 static int mdt_readpage(struct mdt_thread_info *info)
1474 {
1475         struct mdt_object *object = info->mti_object;
1476         struct lu_rdpg    *rdpg = &info->mti_u.rdpg.mti_rdpg;
1477         struct mdt_body   *reqbody;
1478         struct mdt_body   *repbody;
1479         int                rc;
1480         int                i;
1481         ENTRY;
1482
1483         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_READPAGE_PACK))
1484                 RETURN(err_serious(-ENOMEM));
1485
1486         reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY);
1487         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
1488         if (reqbody == NULL || repbody == NULL)
1489                 RETURN(err_serious(-EFAULT));
1490
1491         /*
1492          * prepare @rdpg before calling lower layers and transfer itself. Here
1493          * reqbody->size contains offset of where to start to read and
1494          * reqbody->nlink contains number bytes to read.
1495          */
1496         rdpg->rp_hash = reqbody->size;
1497         if (rdpg->rp_hash != reqbody->size) {
1498                 CERROR("Invalid hash: "LPX64" != "LPX64"\n",
1499                        rdpg->rp_hash, reqbody->size);
1500                 RETURN(-EFAULT);
1501         }
1502
1503         rdpg->rp_attrs = reqbody->mode;
1504         if (info->mti_exp->exp_connect_flags & OBD_CONNECT_64BITHASH)
1505                 rdpg->rp_attrs |= LUDA_64BITHASH;
1506         rdpg->rp_count  = min_t(unsigned int, reqbody->nlink,
1507                                 PTLRPC_MAX_BRW_SIZE);
1508         rdpg->rp_npages = (rdpg->rp_count + CFS_PAGE_SIZE - 1) >>
1509                           CFS_PAGE_SHIFT;
1510         OBD_ALLOC(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]);
1511         if (rdpg->rp_pages == NULL)
1512                 RETURN(-ENOMEM);
1513
1514         for (i = 0; i < rdpg->rp_npages; ++i) {
1515                 rdpg->rp_pages[i] = cfs_alloc_page(CFS_ALLOC_STD);
1516                 if (rdpg->rp_pages[i] == NULL)
1517                         GOTO(free_rdpg, rc = -ENOMEM);
1518         }
1519
1520         /* call lower layers to fill allocated pages with directory data */
1521         rc = mo_readpage(info->mti_env, mdt_object_child(object), rdpg);
1522         if (rc < 0)
1523                 GOTO(free_rdpg, rc);
1524
1525         /* send pages to client */
1526         rc = mdt_sendpage(info, rdpg, rc);
1527
1528         EXIT;
1529 free_rdpg:
1530
1531         for (i = 0; i < rdpg->rp_npages; i++)
1532                 if (rdpg->rp_pages[i] != NULL)
1533                         cfs_free_page(rdpg->rp_pages[i]);
1534         OBD_FREE(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]);
1535
1536         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE))
1537                 RETURN(0);
1538
1539         return rc;
1540 }
1541
1542 static int mdt_reint_internal(struct mdt_thread_info *info,
1543                               struct mdt_lock_handle *lhc,
1544                               __u32 op)
1545 {
1546         struct req_capsule      *pill = info->mti_pill;
1547         struct md_quota         *mq = md_quota(info->mti_env);
1548         struct mdt_body         *repbody;
1549         int                      rc = 0, rc2;
1550         ENTRY;
1551
1552
1553         rc = mdt_reint_unpack(info, op);
1554         if (rc != 0) {
1555                 CERROR("Can't unpack reint, rc %d\n", rc);
1556                 RETURN(err_serious(rc));
1557         }
1558
1559         /* for replay (no_create) lmm is not needed, client has it already */
1560         if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER))
1561                 req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER,
1562                                      info->mti_rr.rr_eadatalen);
1563
1564         if (req_capsule_has_field(pill, &RMF_LOGCOOKIES, RCL_SERVER))
1565                 req_capsule_set_size(pill, &RMF_LOGCOOKIES, RCL_SERVER,
1566                                      info->mti_mdt->mdt_max_cookiesize);
1567
1568         rc = req_capsule_server_pack(pill);
1569         if (rc != 0) {
1570                 CERROR("Can't pack response, rc %d\n", rc);
1571                 RETURN(err_serious(rc));
1572         }
1573
1574         if (req_capsule_has_field(pill, &RMF_MDT_BODY, RCL_SERVER)) {
1575                 repbody = req_capsule_server_get(pill, &RMF_MDT_BODY);
1576                 LASSERT(repbody);
1577                 repbody->eadatasize = 0;
1578                 repbody->aclsize = 0;
1579         }
1580
1581         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_REINT_DELAY, 10);
1582
1583         /* for replay no cookkie / lmm need, because client have this already */
1584         if (info->mti_spec.no_create)
1585                 if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER))
1586                         req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, 0);
1587
1588         rc = mdt_init_ucred_reint(info);
1589         if (rc)
1590                 GOTO(out_shrink, rc);
1591
1592         rc = mdt_fix_attr_ucred(info, op);
1593         if (rc != 0)
1594                 GOTO(out_ucred, rc = err_serious(rc));
1595
1596         if (mdt_check_resent(info, mdt_reconstruct, lhc)) {
1597                 rc = lustre_msg_get_status(mdt_info_req(info)->rq_repmsg);
1598                 GOTO(out_ucred, rc);
1599         }
1600         mq->mq_exp = info->mti_exp;
1601         rc = mdt_reint_rec(info, lhc);
1602         EXIT;
1603 out_ucred:
1604         mdt_exit_ucred(info);
1605 out_shrink:
1606         mdt_client_compatibility(info);
1607         rc2 = mdt_fix_reply(info);
1608         if (rc == 0)
1609                 rc = rc2;
1610         return rc;
1611 }
1612
1613 static long mdt_reint_opcode(struct mdt_thread_info *info,
1614                              const struct req_format **fmt)
1615 {
1616         struct mdt_rec_reint *rec;
1617         long opc;
1618
1619         opc = err_serious(-EFAULT);
1620         rec = req_capsule_client_get(info->mti_pill, &RMF_REC_REINT);
1621         if (rec != NULL) {
1622                 opc = rec->rr_opcode;
1623                 DEBUG_REQ(D_INODE, mdt_info_req(info), "reint opt = %ld", opc);
1624                 if (opc < REINT_MAX && fmt[opc] != NULL)
1625                         req_capsule_extend(info->mti_pill, fmt[opc]);
1626                 else {
1627                         CERROR("Unsupported opc: %ld\n", opc);
1628                         opc = err_serious(opc);
1629                 }
1630         }
1631         return opc;
1632 }
1633
1634 static int mdt_reint(struct mdt_thread_info *info)
1635 {
1636         long opc;
1637         int  rc;
1638
1639         static const struct req_format *reint_fmts[REINT_MAX] = {
1640                 [REINT_SETATTR]  = &RQF_MDS_REINT_SETATTR,
1641                 [REINT_CREATE]   = &RQF_MDS_REINT_CREATE,
1642                 [REINT_LINK]     = &RQF_MDS_REINT_LINK,
1643                 [REINT_UNLINK]   = &RQF_MDS_REINT_UNLINK,
1644                 [REINT_RENAME]   = &RQF_MDS_REINT_RENAME,
1645                 [REINT_OPEN]     = &RQF_MDS_REINT_OPEN,
1646                 [REINT_SETXATTR] = &RQF_MDS_REINT_SETXATTR
1647         };
1648
1649         ENTRY;
1650
1651         opc = mdt_reint_opcode(info, reint_fmts);
1652         if (opc >= 0) {
1653                 /*
1654                  * No lock possible here from client to pass it to reint code
1655                  * path.
1656                  */
1657                 rc = mdt_reint_internal(info, NULL, opc);
1658         } else {
1659                 rc = opc;
1660         }
1661
1662         info->mti_fail_id = OBD_FAIL_MDS_REINT_NET_REP;
1663         RETURN(rc);
1664 }
1665
1666 /* this should sync the whole device */
1667 static int mdt_device_sync(const struct lu_env *env, struct mdt_device *mdt)
1668 {
1669         struct dt_device *dt = mdt->mdt_bottom;
1670         int rc;
1671         ENTRY;
1672
1673         rc = dt->dd_ops->dt_sync(env, dt);
1674         RETURN(rc);
1675 }
1676
1677 /* this should sync this object */
1678 static int mdt_object_sync(struct mdt_thread_info *info)
1679 {
1680         struct md_object *next;
1681         int rc;
1682         ENTRY;
1683
1684         if (!mdt_object_exists(info->mti_object)) {
1685                 CWARN("Non existing object  "DFID"!\n",
1686                       PFID(mdt_object_fid(info->mti_object)));
1687                 RETURN(-ESTALE);
1688         }
1689         next = mdt_object_child(info->mti_object);
1690         rc = mo_object_sync(info->mti_env, next);
1691
1692         RETURN(rc);
1693 }
1694
1695 static int mdt_sync(struct mdt_thread_info *info)
1696 {
1697         struct ptlrpc_request *req = mdt_info_req(info);
1698         struct req_capsule *pill = info->mti_pill;
1699         struct mdt_body *body;
1700         int rc;
1701         ENTRY;
1702
1703         /* The fid may be zero, so we req_capsule_set manually */
1704         req_capsule_set(pill, &RQF_MDS_SYNC);
1705
1706         body = req_capsule_client_get(pill, &RMF_MDT_BODY);
1707         if (body == NULL)
1708                 RETURN(err_serious(-EINVAL));
1709
1710         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SYNC_PACK))
1711                 RETURN(err_serious(-ENOMEM));
1712
1713         if (fid_seq(&body->fid1) == 0) {
1714                 /* sync the whole device */
1715                 rc = req_capsule_server_pack(pill);
1716                 if (rc == 0)
1717                         rc = mdt_device_sync(info->mti_env, info->mti_mdt);
1718                 else
1719                         rc = err_serious(rc);
1720         } else {
1721                 /* sync an object */
1722                 rc = mdt_unpack_req_pack_rep(info, HABEO_CORPUS|HABEO_REFERO);
1723                 if (rc == 0) {
1724                         rc = mdt_object_sync(info);
1725                         if (rc == 0) {
1726                                 struct md_object *next;
1727                                 const struct lu_fid *fid;
1728                                 struct lu_attr *la = &info->mti_attr.ma_attr;
1729
1730                                 next = mdt_object_child(info->mti_object);
1731                                 info->mti_attr.ma_need = MA_INODE;
1732                                 info->mti_attr.ma_valid = 0;
1733                                 rc = mo_attr_get(info->mti_env, next,
1734                                                  &info->mti_attr);
1735                                 if (rc == 0) {
1736                                         body = req_capsule_server_get(pill,
1737                                                                 &RMF_MDT_BODY);
1738                                         fid = mdt_object_fid(info->mti_object);
1739                                         mdt_pack_attr2body(info, body, la, fid);
1740                                 }
1741                         }
1742                 } else
1743                         rc = err_serious(rc);
1744         }
1745         if (rc == 0)
1746                 mdt_counter_incr(req, LPROC_MDT_SYNC);
1747
1748         RETURN(rc);
1749 }
1750
1751 #ifdef HAVE_QUOTA_SUPPORT
1752 static int mdt_quotacheck_handle(struct mdt_thread_info *info)
1753 {
1754         struct obd_quotactl *oqctl;
1755         struct req_capsule *pill = info->mti_pill;
1756         struct obd_export *exp = info->mti_exp;
1757         struct md_quota *mq = md_quota(info->mti_env);
1758         struct md_device *next = info->mti_mdt->mdt_child;
1759         int rc;
1760         ENTRY;
1761
1762         oqctl = req_capsule_client_get(pill, &RMF_OBD_QUOTACTL);
1763         if (oqctl == NULL)
1764                 RETURN(-EPROTO);
1765
1766         /* remote client has no permission for quotacheck */
1767         if (unlikely(exp_connect_rmtclient(exp)))
1768                 RETURN(-EPERM);
1769
1770         rc = req_capsule_server_pack(pill);
1771         if (rc)
1772                 RETURN(rc);
1773
1774         mq->mq_exp = exp;
1775         rc = next->md_ops->mdo_quota.mqo_check(info->mti_env, next,
1776                                                oqctl->qc_type);
1777         RETURN(rc);
1778 }
1779
1780 static int mdt_quotactl_handle(struct mdt_thread_info *info)
1781 {
1782         struct obd_quotactl *oqctl, *repoqc;
1783         struct req_capsule *pill = info->mti_pill;
1784         struct obd_export *exp = info->mti_exp;
1785         struct md_quota *mq = md_quota(info->mti_env);
1786         struct md_device *next = info->mti_mdt->mdt_child;
1787         const struct md_quota_operations *mqo = &next->md_ops->mdo_quota;
1788         int id, rc;
1789         ENTRY;
1790
1791         oqctl = req_capsule_client_get(pill, &RMF_OBD_QUOTACTL);
1792         if (oqctl == NULL)
1793                 RETURN(-EPROTO);
1794
1795         id = oqctl->qc_id;
1796         if (exp_connect_rmtclient(exp)) {
1797                 struct ptlrpc_request *req = mdt_info_req(info);
1798                 struct mdt_export_data *med = mdt_req2med(req);
1799                 struct lustre_idmap_table *idmap = med->med_idmap;
1800
1801                 if (unlikely(oqctl->qc_cmd != Q_GETQUOTA &&
1802                              oqctl->qc_cmd != Q_GETINFO))
1803                         RETURN(-EPERM);
1804
1805
1806                 if (oqctl->qc_type == USRQUOTA)
1807                         id = lustre_idmap_lookup_uid(NULL, idmap, 0,
1808                                                      oqctl->qc_id);
1809                 else if (oqctl->qc_type == GRPQUOTA)
1810                         id = lustre_idmap_lookup_gid(NULL, idmap, 0,
1811                                                      oqctl->qc_id);
1812                 else
1813                         RETURN(-EINVAL);
1814
1815                 if (id == CFS_IDMAP_NOTFOUND) {
1816                         CDEBUG(D_QUOTA, "no mapping for id %u\n",
1817                                oqctl->qc_id);
1818                         RETURN(-EACCES);
1819                 }
1820         }
1821
1822         rc = req_capsule_server_pack(pill);
1823         if (rc)
1824                 RETURN(rc);
1825
1826         repoqc = req_capsule_server_get(pill, &RMF_OBD_QUOTACTL);
1827         LASSERT(repoqc != NULL);
1828
1829         mq->mq_exp = exp;
1830         switch (oqctl->qc_cmd) {
1831         case Q_QUOTAON:
1832                 rc = mqo->mqo_on(info->mti_env, next, oqctl->qc_type);
1833                 break;
1834         case Q_QUOTAOFF:
1835                 rc = mqo->mqo_off(info->mti_env, next, oqctl->qc_type);
1836                 break;
1837         case Q_SETINFO:
1838                 rc = mqo->mqo_setinfo(info->mti_env, next, oqctl->qc_type, id,
1839                                       &oqctl->qc_dqinfo);
1840                 break;
1841         case Q_GETINFO:
1842                 rc = mqo->mqo_getinfo(info->mti_env, next, oqctl->qc_type, id,
1843                                       &oqctl->qc_dqinfo);
1844                 break;
1845         case Q_SETQUOTA:
1846                 rc = mqo->mqo_setquota(info->mti_env, next, oqctl->qc_type, id,
1847                                        &oqctl->qc_dqblk);
1848                 break;
1849         case Q_GETQUOTA:
1850                 rc = mqo->mqo_getquota(info->mti_env, next, oqctl->qc_type, id,
1851                                        &oqctl->qc_dqblk);
1852                 break;
1853         case Q_GETOINFO:
1854                 rc = mqo->mqo_getoinfo(info->mti_env, next, oqctl->qc_type, id,
1855                                        &oqctl->qc_dqinfo);
1856                 break;
1857         case Q_GETOQUOTA:
1858                 rc = mqo->mqo_getoquota(info->mti_env, next, oqctl->qc_type, id,
1859                                         &oqctl->qc_dqblk);
1860                 break;
1861         case LUSTRE_Q_INVALIDATE:
1862                 rc = mqo->mqo_invalidate(info->mti_env, next, oqctl->qc_type);
1863                 break;
1864         case LUSTRE_Q_FINVALIDATE:
1865                 rc = mqo->mqo_finvalidate(info->mti_env, next, oqctl->qc_type);
1866                 break;
1867         default:
1868                 CERROR("unsupported mdt_quotactl command: %d\n",
1869                        oqctl->qc_cmd);
1870                 RETURN(-EFAULT);
1871         }
1872
1873         *repoqc = *oqctl;
1874         RETURN(rc);
1875 }
1876 #endif
1877
1878
1879 /*
1880  * OBD PING and other handlers.
1881  */
1882 static int mdt_obd_ping(struct mdt_thread_info *info)
1883 {
1884         int rc;
1885         ENTRY;
1886
1887         req_capsule_set(info->mti_pill, &RQF_OBD_PING);
1888
1889         rc = target_handle_ping(mdt_info_req(info));
1890         if (rc < 0)
1891                 rc = err_serious(rc);
1892         RETURN(rc);
1893 }
1894
1895 static int mdt_obd_log_cancel(struct mdt_thread_info *info)
1896 {
1897         return err_serious(-EOPNOTSUPP);
1898 }
1899
1900 static int mdt_obd_qc_callback(struct mdt_thread_info *info)
1901 {
1902         return err_serious(-EOPNOTSUPP);
1903 }
1904
1905
1906 /*
1907  * LLOG handlers.
1908  */
1909
1910 /** clone llog ctxt from child (mdd)
1911  * This allows remote llog (replicator) access.
1912  * We can either pass all llog RPCs (eg mdt_llog_create) on to child where the
1913  * context was originally set up, or we can handle them directly.
1914  * I choose the latter, but that means I need any llog
1915  * contexts set up by child to be accessable by the mdt.  So we clone the
1916  * context into our context list here.
1917  */
1918 static int mdt_llog_ctxt_clone(const struct lu_env *env, struct mdt_device *mdt,
1919                                int idx)
1920 {
1921         struct md_device  *next = mdt->mdt_child;
1922         struct llog_ctxt *ctxt;
1923         int rc;
1924
1925         if (!llog_ctxt_null(mdt2obd_dev(mdt), idx))
1926                 return 0;
1927
1928         rc = next->md_ops->mdo_llog_ctxt_get(env, next, idx, (void **)&ctxt);
1929         if (rc || ctxt == NULL) {
1930                 CERROR("Can't get mdd ctxt %d\n", rc);
1931                 return rc;
1932         }
1933
1934         rc = llog_group_set_ctxt(&mdt2obd_dev(mdt)->obd_olg, ctxt, idx);
1935         if (rc)
1936                 CERROR("Can't set mdt ctxt %d\n", rc);
1937
1938         return rc;
1939 }
1940
1941 static int mdt_llog_ctxt_unclone(const struct lu_env *env,
1942                                  struct mdt_device *mdt, int idx)
1943 {
1944         struct llog_ctxt *ctxt;
1945
1946         ctxt = llog_get_context(mdt2obd_dev(mdt), idx);
1947         if (ctxt == NULL)
1948                 return 0;
1949         /* Put once for the get we just did, and once for the clone */
1950         llog_ctxt_put(ctxt);
1951         llog_ctxt_put(ctxt);
1952         return 0;
1953 }
1954
1955 static int mdt_llog_create(struct mdt_thread_info *info)
1956 {
1957         int rc;
1958
1959         req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_CREATE);
1960         rc = llog_origin_handle_create(mdt_info_req(info));
1961         return (rc < 0 ? err_serious(rc) : rc);
1962 }
1963
1964 static int mdt_llog_destroy(struct mdt_thread_info *info)
1965 {
1966         int rc;
1967
1968         req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_DESTROY);
1969         rc = llog_origin_handle_destroy(mdt_info_req(info));
1970         return (rc < 0 ? err_serious(rc) : rc);
1971 }
1972
1973 static int mdt_llog_read_header(struct mdt_thread_info *info)
1974 {
1975         int rc;
1976
1977         req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_READ_HEADER);
1978         rc = llog_origin_handle_read_header(mdt_info_req(info));
1979         return (rc < 0 ? err_serious(rc) : rc);
1980 }
1981
1982 static int mdt_llog_next_block(struct mdt_thread_info *info)
1983 {
1984         int rc;
1985
1986         req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_NEXT_BLOCK);
1987         rc = llog_origin_handle_next_block(mdt_info_req(info));
1988         return (rc < 0 ? err_serious(rc) : rc);
1989 }
1990
1991 static int mdt_llog_prev_block(struct mdt_thread_info *info)
1992 {
1993         int rc;
1994
1995         req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_PREV_BLOCK);
1996         rc = llog_origin_handle_prev_block(mdt_info_req(info));
1997         return (rc < 0 ? err_serious(rc) : rc);
1998 }
1999
2000
2001 /*
2002  * DLM handlers.
2003  */
2004 static struct ldlm_callback_suite cbs = {
2005         .lcs_completion = ldlm_server_completion_ast,
2006         .lcs_blocking   = ldlm_server_blocking_ast,
2007         .lcs_glimpse    = NULL
2008 };
2009
2010 static int mdt_enqueue(struct mdt_thread_info *info)
2011 {
2012         struct ptlrpc_request *req;
2013         int rc;
2014
2015         /*
2016          * info->mti_dlm_req already contains swapped and (if necessary)
2017          * converted dlm request.
2018          */
2019         LASSERT(info->mti_dlm_req != NULL);
2020
2021         req = mdt_info_req(info);
2022         rc = ldlm_handle_enqueue0(info->mti_mdt->mdt_namespace,
2023                                   req, info->mti_dlm_req, &cbs);
2024         info->mti_fail_id = OBD_FAIL_LDLM_REPLY;
2025         return rc ? err_serious(rc) : req->rq_status;
2026 }
2027
2028 static int mdt_convert(struct mdt_thread_info *info)
2029 {
2030         int rc;
2031         struct ptlrpc_request *req;
2032
2033         LASSERT(info->mti_dlm_req);
2034         req = mdt_info_req(info);
2035         rc = ldlm_handle_convert0(req, info->mti_dlm_req);
2036         return rc ? err_serious(rc) : req->rq_status;
2037 }
2038
2039 static int mdt_bl_callback(struct mdt_thread_info *info)
2040 {
2041         CERROR("bl callbacks should not happen on MDS\n");
2042         LBUG();
2043         return err_serious(-EOPNOTSUPP);
2044 }
2045
2046 static int mdt_cp_callback(struct mdt_thread_info *info)
2047 {
2048         CERROR("cp callbacks should not happen on MDS\n");
2049         LBUG();
2050         return err_serious(-EOPNOTSUPP);
2051 }
2052
2053 /*
2054  * sec context handlers
2055  */
2056 static int mdt_sec_ctx_handle(struct mdt_thread_info *info)
2057 {
2058         int rc;
2059
2060         rc = mdt_handle_idmap(info);
2061
2062         if (unlikely(rc)) {
2063                 struct ptlrpc_request *req = mdt_info_req(info);
2064                 __u32                  opc;
2065
2066                 opc = lustre_msg_get_opc(req->rq_reqmsg);
2067                 if (opc == SEC_CTX_INIT || opc == SEC_CTX_INIT_CONT)
2068                         sptlrpc_svc_ctx_invalidate(req);
2069         }
2070
2071         CFS_FAIL_TIMEOUT(OBD_FAIL_SEC_CTX_HDL_PAUSE, cfs_fail_val);
2072
2073         return rc;
2074 }
2075
2076 static struct mdt_object *mdt_obj(struct lu_object *o)
2077 {
2078         LASSERT(lu_device_is_mdt(o->lo_dev));
2079         return container_of0(o, struct mdt_object, mot_obj.mo_lu);
2080 }
2081
2082 struct mdt_object *mdt_object_find(const struct lu_env *env,
2083                                    struct mdt_device *d,
2084                                    const struct lu_fid *f)
2085 {
2086         struct lu_object *o;
2087         struct mdt_object *m;
2088         ENTRY;
2089
2090         CDEBUG(D_INFO, "Find object for "DFID"\n", PFID(f));
2091         o = lu_object_find(env, &d->mdt_md_dev.md_lu_dev, f, NULL);
2092         if (unlikely(IS_ERR(o)))
2093                 m = (struct mdt_object *)o;
2094         else
2095                 m = mdt_obj(o);
2096         RETURN(m);
2097 }
2098
2099 /**
2100  * Asyncronous commit for mdt device.
2101  *
2102  * Pass asynchonous commit call down the MDS stack.
2103  *
2104  * \param env environment
2105  * \param mdt the mdt device
2106  */
2107 static void mdt_device_commit_async(const struct lu_env *env,
2108                                     struct mdt_device *mdt)
2109 {
2110         struct dt_device *dt = mdt->mdt_bottom;
2111         int rc;
2112
2113         rc = dt->dd_ops->dt_commit_async(env, dt);
2114         if (unlikely(rc != 0))
2115                 CWARN("async commit start failed with rc = %d", rc);
2116 }
2117
2118 /**
2119  * Mark the lock as "synchonous".
2120  *
2121  * Mark the lock to deffer transaction commit to the unlock time.
2122  *
2123  * \param lock the lock to mark as "synchonous"
2124  *
2125  * \see mdt_is_lock_sync
2126  * \see mdt_save_lock
2127  */
2128 static inline void mdt_set_lock_sync(struct ldlm_lock *lock)
2129 {
2130         lock->l_ast_data = (void*)1;
2131 }
2132
2133 /**
2134  * Check whehter the lock "synchonous" or not.
2135  *
2136  * \param lock the lock to check
2137  * \retval 1 the lock is "synchonous"
2138  * \retval 0 the lock isn't "synchronous"
2139  *
2140  * \see mdt_set_lock_sync
2141  * \see mdt_save_lock
2142  */
2143 static inline int mdt_is_lock_sync(struct ldlm_lock *lock)
2144 {
2145         return lock->l_ast_data != NULL;
2146 }
2147
2148 /**
2149  * Blocking AST for mdt locks.
2150  *
2151  * Starts transaction commit if in case of COS lock conflict or
2152  * deffers such a commit to the mdt_save_lock.
2153  *
2154  * \param lock the lock which blocks a request or cancelling lock
2155  * \param desc unused
2156  * \param data unused
2157  * \param flag indicates whether this cancelling or blocking callback
2158  * \retval 0
2159  * \see ldlm_blocking_ast_nocheck
2160  */
2161 int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
2162                      void *data, int flag)
2163 {
2164         struct obd_device *obd = ldlm_lock_to_ns(lock)->ns_obd;
2165         struct mdt_device *mdt = mdt_dev(obd->obd_lu_dev);
2166         int rc;
2167         ENTRY;
2168
2169         if (flag == LDLM_CB_CANCELING)
2170                 RETURN(0);
2171         lock_res_and_lock(lock);
2172         if (lock->l_blocking_ast != mdt_blocking_ast) {
2173                 unlock_res_and_lock(lock);
2174                 RETURN(0);
2175         }
2176         if (mdt_cos_is_enabled(mdt) &&
2177             lock->l_req_mode & (LCK_PW | LCK_EX) &&
2178             lock->l_blocking_lock != NULL &&
2179             lock->l_client_cookie != lock->l_blocking_lock->l_client_cookie) {
2180                 mdt_set_lock_sync(lock);
2181         }
2182         rc = ldlm_blocking_ast_nocheck(lock);
2183
2184         /* There is no lock conflict if l_blocking_lock == NULL,
2185          * it indicates a blocking ast sent from ldlm_lock_decref_internal
2186          * when the last reference to a local lock was released */
2187         if (lock->l_req_mode == LCK_COS && lock->l_blocking_lock != NULL) {
2188                 struct lu_env env;
2189
2190                 rc = lu_env_init(&env, LCT_LOCAL);
2191                 if (unlikely(rc != 0))
2192                         CWARN("lu_env initialization failed with rc = %d,"
2193                               "cannot start asynchronous commit\n", rc);
2194                 else
2195                         mdt_device_commit_async(&env, mdt);
2196                 lu_env_fini(&env);
2197         }
2198         RETURN(rc);
2199 }
2200
2201 int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o,
2202                     struct mdt_lock_handle *lh, __u64 ibits, int locality)
2203 {
2204         struct ldlm_namespace *ns = info->mti_mdt->mdt_namespace;
2205         ldlm_policy_data_t *policy = &info->mti_policy;
2206         struct ldlm_res_id *res_id = &info->mti_res_id;
2207         int rc;
2208         ENTRY;
2209
2210         LASSERT(!lustre_handle_is_used(&lh->mlh_reg_lh));
2211         LASSERT(!lustre_handle_is_used(&lh->mlh_pdo_lh));
2212         LASSERT(lh->mlh_reg_mode != LCK_MINMODE);
2213         LASSERT(lh->mlh_type != MDT_NUL_LOCK);
2214
2215         if (mdt_object_exists(o) < 0) {
2216                 if (locality == MDT_CROSS_LOCK) {
2217                         /* cross-ref object fix */
2218                         ibits &= ~MDS_INODELOCK_UPDATE;
2219                         ibits |= MDS_INODELOCK_LOOKUP;
2220                 } else {
2221                         LASSERT(!(ibits & MDS_INODELOCK_UPDATE));
2222                         LASSERT(ibits & MDS_INODELOCK_LOOKUP);
2223                 }
2224                 /* No PDO lock on remote object */
2225                 LASSERT(lh->mlh_type != MDT_PDO_LOCK);
2226         }
2227
2228         if (lh->mlh_type == MDT_PDO_LOCK) {
2229                 /* check for exists after object is locked */
2230                 if (mdt_object_exists(o) == 0) {
2231                         /* Non-existent object shouldn't have PDO lock */
2232                         RETURN(-ESTALE);
2233                 } else {
2234                         /* Non-dir object shouldn't have PDO lock */
2235                         if (!S_ISDIR(lu_object_attr(&o->mot_obj.mo_lu)))
2236                                 RETURN(-ENOTDIR);
2237                 }
2238         }
2239
2240         memset(policy, 0, sizeof(*policy));
2241         fid_build_reg_res_name(mdt_object_fid(o), res_id);
2242
2243         /*
2244          * Take PDO lock on whole directory and build correct @res_id for lock
2245          * on part of directory.
2246          */
2247         if (lh->mlh_pdo_hash != 0) {
2248                 LASSERT(lh->mlh_type == MDT_PDO_LOCK);
2249                 mdt_lock_pdo_mode(info, o, lh);
2250                 if (lh->mlh_pdo_mode != LCK_NL) {
2251                         /*
2252                          * Do not use LDLM_FL_LOCAL_ONLY for parallel lock, it
2253                          * is never going to be sent to client and we do not
2254                          * want it slowed down due to possible cancels.
2255                          */
2256                         policy->l_inodebits.bits = MDS_INODELOCK_UPDATE;
2257                         rc = mdt_fid_lock(ns, &lh->mlh_pdo_lh, lh->mlh_pdo_mode,
2258                                           policy, res_id, LDLM_FL_ATOMIC_CB,
2259                                           &info->mti_exp->exp_handle.h_cookie);
2260                         if (unlikely(rc))
2261                                 RETURN(rc);
2262                 }
2263
2264                 /*
2265                  * Finish res_id initializing by name hash marking part of
2266                  * directory which is taking modification.
2267                  */
2268                 res_id->name[LUSTRE_RES_ID_HSH_OFF] = lh->mlh_pdo_hash;
2269         }
2270
2271         policy->l_inodebits.bits = ibits;
2272
2273         /*
2274          * Use LDLM_FL_LOCAL_ONLY for this lock. We do not know yet if it is
2275          * going to be sent to client. If it is - mdt_intent_policy() path will
2276          * fix it up and turn FL_LOCAL flag off.
2277          */
2278         rc = mdt_fid_lock(ns, &lh->mlh_reg_lh, lh->mlh_reg_mode, policy,
2279                           res_id, LDLM_FL_LOCAL_ONLY | LDLM_FL_ATOMIC_CB,
2280                           &info->mti_exp->exp_handle.h_cookie);
2281         if (rc)
2282                 mdt_object_unlock(info, o, lh, 1);
2283         else if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_MDS_PDO_LOCK)) &&
2284                  lh->mlh_pdo_hash != 0 &&
2285                  (lh->mlh_reg_mode == LCK_PW || lh->mlh_reg_mode == LCK_EX)) {
2286                 OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PDO_LOCK, 15);
2287         }
2288
2289         RETURN(rc);
2290 }
2291
2292 /**
2293  * Save a lock within request object.
2294  *
2295  * Keep the lock referenced until whether client ACK or transaction
2296  * commit happens or release the lock immediately depending on input
2297  * parameters. If COS is ON, a write lock is converted to COS lock
2298  * before saving.
2299  *
2300  * \param info thead info object
2301  * \param h lock handle
2302  * \param mode lock mode
2303  * \param decref force immediate lock releasing
2304  */
2305 static
2306 void mdt_save_lock(struct mdt_thread_info *info, struct lustre_handle *h,
2307                    ldlm_mode_t mode, int decref)
2308 {
2309         ENTRY;
2310
2311         if (lustre_handle_is_used(h)) {
2312                 if (decref || !info->mti_has_trans ||
2313                     !(mode & (LCK_PW | LCK_EX))){
2314                         mdt_fid_unlock(h, mode);
2315                 } else {
2316                         struct mdt_device *mdt = info->mti_mdt;
2317                         struct ldlm_lock *lock = ldlm_handle2lock(h);
2318                         struct ptlrpc_request *req = mdt_info_req(info);
2319                         int no_ack = 0;
2320
2321                         LASSERTF(lock != NULL, "no lock for cookie "LPX64"\n",
2322                                  h->cookie);
2323                         CDEBUG(D_HA, "request = %p reply state = %p"
2324                                " transno = "LPD64"\n",
2325                                req, req->rq_reply_state, req->rq_transno);
2326                         if (mdt_cos_is_enabled(mdt)) {
2327                                 no_ack = 1;
2328                                 ldlm_lock_downgrade(lock, LCK_COS);
2329                                 mode = LCK_COS;
2330                         }
2331                         ptlrpc_save_lock(req, h, mode, no_ack);
2332                         if (mdt_is_lock_sync(lock)) {
2333                                 CDEBUG(D_HA, "found sync-lock,"
2334                                        " async commit started\n");
2335                                 mdt_device_commit_async(info->mti_env,
2336                                                         mdt);
2337                         }
2338                         LDLM_LOCK_PUT(lock);
2339                 }
2340                 h->cookie = 0ull;
2341         }
2342
2343         EXIT;
2344 }
2345
2346 /**
2347  * Unlock mdt object.
2348  *
2349  * Immeditely release the regular lock and the PDO lock or save the
2350  * lock in reqeuest and keep them referenced until client ACK or
2351  * transaction commit.
2352  *
2353  * \param info thread info object
2354  * \param o mdt object
2355  * \param lh mdt lock handle referencing regular and PDO locks
2356  * \param decref force immediate lock releasing
2357  */
2358 void mdt_object_unlock(struct mdt_thread_info *info, struct mdt_object *o,
2359                        struct mdt_lock_handle *lh, int decref)
2360 {
2361         ENTRY;
2362
2363         mdt_save_lock(info, &lh->mlh_pdo_lh, lh->mlh_pdo_mode, decref);
2364         mdt_save_lock(info, &lh->mlh_reg_lh, lh->mlh_reg_mode, decref);
2365
2366         EXIT;
2367 }
2368
2369 struct mdt_object *mdt_object_find_lock(struct mdt_thread_info *info,
2370                                         const struct lu_fid *f,
2371                                         struct mdt_lock_handle *lh,
2372                                         __u64 ibits)
2373 {
2374         struct mdt_object *o;
2375
2376         o = mdt_object_find(info->mti_env, info->mti_mdt, f);
2377         if (!IS_ERR(o)) {
2378                 int rc;
2379
2380                 rc = mdt_object_lock(info, o, lh, ibits,
2381                                      MDT_LOCAL_LOCK);
2382                 if (rc != 0) {
2383                         mdt_object_put(info->mti_env, o);
2384                         o = ERR_PTR(rc);
2385                 }
2386         }
2387         return o;
2388 }
2389
2390 void mdt_object_unlock_put(struct mdt_thread_info * info,
2391                            struct mdt_object * o,
2392                            struct mdt_lock_handle *lh,
2393                            int decref)
2394 {
2395         mdt_object_unlock(info, o, lh, decref);
2396         mdt_object_put(info->mti_env, o);
2397 }
2398
2399 static struct mdt_handler *mdt_handler_find(__u32 opc,
2400                                             struct mdt_opc_slice *supported)
2401 {
2402         struct mdt_opc_slice *s;
2403         struct mdt_handler   *h;
2404
2405         h = NULL;
2406         for (s = supported; s->mos_hs != NULL; s++) {
2407                 if (s->mos_opc_start <= opc && opc < s->mos_opc_end) {
2408                         h = s->mos_hs + (opc - s->mos_opc_start);
2409                         if (likely(h->mh_opc != 0))
2410                                 LASSERTF(h->mh_opc == opc,
2411                                          "opcode mismatch %d != %d\n",
2412                                          h->mh_opc, opc);
2413                         else
2414                                 h = NULL; /* unsupported opc */
2415                         break;
2416                 }
2417         }
2418         return h;
2419 }
2420
2421 static int mdt_lock_resname_compat(struct mdt_device *m,
2422                                    struct ldlm_request *req)
2423 {
2424         /* XXX something... later. */
2425         return 0;
2426 }
2427
2428 static int mdt_lock_reply_compat(struct mdt_device *m, struct ldlm_reply *rep)
2429 {
2430         /* XXX something... later. */
2431         return 0;
2432 }
2433
2434 /*
2435  * Generic code handling requests that have struct mdt_body passed in:
2436  *
2437  *  - extract mdt_body from request and save it in @info, if present;
2438  *
2439  *  - create lu_object, corresponding to the fid in mdt_body, and save it in
2440  *  @info;
2441  *
2442  *  - if HABEO_CORPUS flag is set for this request type check whether object
2443  *  actually exists on storage (lu_object_exists()).
2444  *
2445  */
2446 static int mdt_body_unpack(struct mdt_thread_info *info, __u32 flags)
2447 {
2448         const struct mdt_body    *body;
2449         struct mdt_object        *obj;
2450         const struct lu_env      *env;
2451         struct req_capsule       *pill;
2452         int                       rc;
2453         ENTRY;
2454
2455         env = info->mti_env;
2456         pill = info->mti_pill;
2457
2458         body = info->mti_body = req_capsule_client_get(pill, &RMF_MDT_BODY);
2459         if (body == NULL)
2460                 RETURN(-EFAULT);
2461
2462         if (!(body->valid & OBD_MD_FLID))
2463                 RETURN(0);
2464
2465         if (!fid_is_sane(&body->fid1)) {
2466                 CERROR("Invalid fid: "DFID"\n", PFID(&body->fid1));
2467                 RETURN(-EINVAL);
2468         }
2469
2470         /*
2471          * Do not get size or any capa fields before we check that request
2472          * contains capa actually. There are some requests which do not, for
2473          * instance MDS_IS_SUBDIR.
2474          */
2475         if (req_capsule_has_field(pill, &RMF_CAPA1, RCL_CLIENT) &&
2476             req_capsule_get_size(pill, &RMF_CAPA1, RCL_CLIENT))
2477                 mdt_set_capainfo(info, 0, &body->fid1,
2478                                  req_capsule_client_get(pill, &RMF_CAPA1));
2479
2480         obj = mdt_object_find(env, info->mti_mdt, &body->fid1);
2481         if (!IS_ERR(obj)) {
2482                 if ((flags & HABEO_CORPUS) &&
2483                     !mdt_object_exists(obj)) {
2484                         mdt_object_put(env, obj);
2485                         /* for capability renew ENOENT will be handled in
2486                          * mdt_renew_capa */
2487                         if (body->valid & OBD_MD_FLOSSCAPA)
2488                                 rc = 0;
2489                         else
2490                                 rc = -ENOENT;
2491                 } else {
2492                         info->mti_object = obj;
2493                         rc = 0;
2494                 }
2495         } else
2496                 rc = PTR_ERR(obj);
2497
2498         RETURN(rc);
2499 }
2500
2501 static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags)
2502 {
2503         struct req_capsule *pill = info->mti_pill;
2504         int rc;
2505         ENTRY;
2506
2507         if (req_capsule_has_field(pill, &RMF_MDT_BODY, RCL_CLIENT))
2508                 rc = mdt_body_unpack(info, flags);
2509         else
2510                 rc = 0;
2511
2512         if (rc == 0 && (flags & HABEO_REFERO)) {
2513                 /* Pack reply. */
2514                 if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER))
2515                         req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER,
2516                                              info->mti_body->eadatasize);
2517                 if (req_capsule_has_field(pill, &RMF_LOGCOOKIES, RCL_SERVER))
2518                         req_capsule_set_size(pill, &RMF_LOGCOOKIES, RCL_SERVER,
2519                                              info->mti_mdt->mdt_max_cookiesize);
2520
2521                 rc = req_capsule_server_pack(pill);
2522         }
2523         RETURN(rc);
2524 }
2525
2526 static int mdt_init_capa_ctxt(const struct lu_env *env, struct mdt_device *m)
2527 {
2528         struct md_device *next = m->mdt_child;
2529
2530         return next->md_ops->mdo_init_capa_ctxt(env, next,
2531                                                 m->mdt_opts.mo_mds_capa,
2532                                                 m->mdt_capa_timeout,
2533                                                 m->mdt_capa_alg,
2534                                                 m->mdt_capa_keys);
2535 }
2536
2537 /*
2538  * Invoke handler for this request opc. Also do necessary preprocessing
2539  * (according to handler ->mh_flags), and post-processing (setting of
2540  * ->last_{xid,committed}).
2541  */
2542 static int mdt_req_handle(struct mdt_thread_info *info,
2543                           struct mdt_handler *h, struct ptlrpc_request *req)
2544 {
2545         int   rc, serious = 0;
2546         __u32 flags;
2547
2548         ENTRY;
2549
2550         LASSERT(h->mh_act != NULL);
2551         LASSERT(h->mh_opc == lustre_msg_get_opc(req->rq_reqmsg));
2552         LASSERT(current->journal_info == NULL);
2553
2554         /*
2555          * Checking for various OBD_FAIL_$PREF_$OPC_NET codes. _Do_ not try
2556          * to put same checks into handlers like mdt_close(), mdt_reint(),
2557          * etc., without talking to mdt authors first. Checking same thing
2558          * there again is useless and returning 0 error without packing reply
2559          * is buggy! Handlers either pack reply or return error.
2560          *
2561          * We return 0 here and do not send any reply in order to emulate
2562          * network failure. Do not send any reply in case any of NET related
2563          * fail_id has occured.
2564          */
2565         if (OBD_FAIL_CHECK_ORSET(h->mh_fail_id, OBD_FAIL_ONCE))
2566                 RETURN(0);
2567
2568         rc = 0;
2569         flags = h->mh_flags;
2570         LASSERT(ergo(flags & (HABEO_CORPUS|HABEO_REFERO), h->mh_fmt != NULL));
2571
2572         if (h->mh_fmt != NULL) {
2573                 req_capsule_set(info->mti_pill, h->mh_fmt);
2574                 rc = mdt_unpack_req_pack_rep(info, flags);
2575         }
2576
2577         if (rc == 0 && flags & MUTABOR &&
2578             req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY)
2579                 /* should it be rq_status? */
2580                 rc = -EROFS;
2581
2582         if (rc == 0 && flags & HABEO_CLAVIS) {
2583                 struct ldlm_request *dlm_req;
2584
2585                 LASSERT(h->mh_fmt != NULL);
2586
2587                 dlm_req = req_capsule_client_get(info->mti_pill, &RMF_DLM_REQ);
2588                 if (dlm_req != NULL) {
2589                         if (unlikely(dlm_req->lock_desc.l_resource.lr_type ==
2590                                         LDLM_IBITS &&
2591                                      dlm_req->lock_desc.l_policy_data.\
2592                                         l_inodebits.bits == 0)) {
2593                                 /*
2594                                  * Lock without inodebits makes no sense and
2595                                  * will oops later in ldlm. If client miss to
2596                                  * set such bits, do not trigger ASSERTION.
2597                                  *
2598                                  * For liblustre flock case, it maybe zero.
2599                                  */
2600                                 rc = -EPROTO;
2601                         } else {
2602                                 if (info->mti_mdt->mdt_opts.mo_compat_resname)
2603                                         rc = mdt_lock_resname_compat(
2604                                                                 info->mti_mdt,
2605                                                                 dlm_req);
2606                                 info->mti_dlm_req = dlm_req;
2607                         }
2608                 } else {
2609                         rc = -EFAULT;
2610                 }
2611         }
2612
2613         /* capability setting changed via /proc, needs reinitialize ctxt */
2614         if (info->mti_mdt && info->mti_mdt->mdt_capa_conf) {
2615                 mdt_init_capa_ctxt(info->mti_env, info->mti_mdt);
2616                 info->mti_mdt->mdt_capa_conf = 0;
2617         }
2618
2619         if (likely(rc == 0)) {
2620                 /*
2621                  * Process request, there can be two types of rc:
2622                  * 1) errors with msg unpack/pack, other failures outside the
2623                  * operation itself. This is counted as serious errors;
2624                  * 2) errors during fs operation, should be placed in rq_status
2625                  * only
2626                  */
2627                 rc = h->mh_act(info);
2628                 if (rc == 0 &&
2629                     !req->rq_no_reply && req->rq_reply_state == NULL) {
2630                         DEBUG_REQ(D_ERROR, req, "MDT \"handler\" %s did not "
2631                                   "pack reply and returned 0 error\n",
2632                                   h->mh_name);
2633                         LBUG();
2634                 }
2635                 serious = is_serious(rc);
2636                 rc = clear_serious(rc);
2637         } else
2638                 serious = 1;
2639
2640         req->rq_status = rc;
2641
2642         /*
2643          * ELDLM_* codes which > 0 should be in rq_status only as well as
2644          * all non-serious errors.
2645          */
2646         if (rc > 0 || !serious)
2647                 rc = 0;
2648
2649         LASSERT(current->journal_info == NULL);
2650
2651         if (rc == 0 && (flags & HABEO_CLAVIS) &&
2652             info->mti_mdt->mdt_opts.mo_compat_resname) {
2653                 struct ldlm_reply *dlmrep;
2654
2655                 dlmrep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP);
2656                 if (dlmrep != NULL)
2657                         rc = mdt_lock_reply_compat(info->mti_mdt, dlmrep);
2658         }
2659
2660         /* If we're DISCONNECTing, the mdt_export_data is already freed */
2661         if (likely(rc == 0 && req->rq_export && h->mh_opc != MDS_DISCONNECT))
2662                 target_committed_to_req(req);
2663
2664         if (unlikely(req_is_replay(req) &&
2665                      lustre_msg_get_transno(req->rq_reqmsg) == 0)) {
2666                 DEBUG_REQ(D_ERROR, req, "transno is 0 during REPLAY");
2667                 LBUG();
2668         }
2669
2670         target_send_reply(req, rc, info->mti_fail_id);
2671         RETURN(0);
2672 }
2673
2674 void mdt_lock_handle_init(struct mdt_lock_handle *lh)
2675 {
2676         lh->mlh_type = MDT_NUL_LOCK;
2677         lh->mlh_reg_lh.cookie = 0ull;
2678         lh->mlh_reg_mode = LCK_MINMODE;
2679         lh->mlh_pdo_lh.cookie = 0ull;
2680         lh->mlh_pdo_mode = LCK_MINMODE;
2681 }
2682
2683 void mdt_lock_handle_fini(struct mdt_lock_handle *lh)
2684 {
2685         LASSERT(!lustre_handle_is_used(&lh->mlh_reg_lh));
2686         LASSERT(!lustre_handle_is_used(&lh->mlh_pdo_lh));
2687 }
2688
2689 /*
2690  * Initialize fields of struct mdt_thread_info. Other fields are left in
2691  * uninitialized state, because it's too expensive to zero out whole
2692  * mdt_thread_info (> 1K) on each request arrival.
2693  */
2694 static void mdt_thread_info_init(struct ptlrpc_request *req,
2695                                  struct mdt_thread_info *info)
2696 {
2697         int i;
2698         struct md_capainfo *ci;
2699
2700         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
2701         info->mti_pill = &req->rq_pill;
2702
2703         /* lock handle */
2704         for (i = 0; i < ARRAY_SIZE(info->mti_lh); i++)
2705                 mdt_lock_handle_init(&info->mti_lh[i]);
2706
2707         /* mdt device: it can be NULL while CONNECT */
2708         if (req->rq_export) {
2709                 info->mti_mdt = mdt_dev(req->rq_export->exp_obd->obd_lu_dev);
2710                 info->mti_exp = req->rq_export;
2711         } else
2712                 info->mti_mdt = NULL;
2713         info->mti_env = req->rq_svc_thread->t_env;
2714         ci = md_capainfo(info->mti_env);
2715         memset(ci, 0, sizeof *ci);
2716         if (req->rq_export) {
2717                 if (exp_connect_rmtclient(req->rq_export))
2718                         ci->mc_auth = LC_ID_CONVERT;
2719                 else if (req->rq_export->exp_connect_flags &
2720                          OBD_CONNECT_MDS_CAPA)
2721                         ci->mc_auth = LC_ID_PLAIN;
2722                 else
2723                         ci->mc_auth = LC_ID_NONE;
2724         }
2725
2726         info->mti_fail_id = OBD_FAIL_MDS_ALL_REPLY_NET;
2727         info->mti_transno = lustre_msg_get_transno(req->rq_reqmsg);
2728         info->mti_mos = NULL;
2729
2730         memset(&info->mti_attr, 0, sizeof(info->mti_attr));
2731         info->mti_body = NULL;
2732         info->mti_object = NULL;
2733         info->mti_dlm_req = NULL;
2734         info->mti_has_trans = 0;
2735         info->mti_no_need_trans = 0;
2736         info->mti_cross_ref = 0;
2737         info->mti_opdata = 0;
2738
2739         /* To not check for split by default. */
2740         info->mti_spec.sp_ck_split = 0;
2741         info->mti_spec.no_create = 0;
2742 }
2743
2744 static void mdt_thread_info_fini(struct mdt_thread_info *info)
2745 {
2746         int i;
2747
2748         req_capsule_fini(info->mti_pill);
2749         if (info->mti_object != NULL) {
2750                 /*
2751                  * freeing an object may lead to OSD level transaction, do not
2752                  * let it mess with MDT. bz19385.
2753                  */
2754                 info->mti_no_need_trans = 1;
2755                 mdt_object_put(info->mti_env, info->mti_object);
2756                 info->mti_object = NULL;
2757         }
2758         for (i = 0; i < ARRAY_SIZE(info->mti_lh); i++)
2759                 mdt_lock_handle_fini(&info->mti_lh[i]);
2760         info->mti_env = NULL;
2761 }
2762
2763 static int mdt_filter_recovery_request(struct ptlrpc_request *req,
2764                                        struct obd_device *obd, int *process)
2765 {
2766         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
2767         case MDS_CONNECT: /* This will never get here, but for completeness. */
2768         case OST_CONNECT: /* This will never get here, but for completeness. */
2769         case MDS_DISCONNECT:
2770         case OST_DISCONNECT:
2771                *process = 1;
2772                RETURN(0);
2773
2774         case MDS_CLOSE:
2775         case MDS_DONE_WRITING:
2776         case MDS_SYNC: /* used in unmounting */
2777         case OBD_PING:
2778         case MDS_REINT:
2779         case SEQ_QUERY:
2780         case FLD_QUERY:
2781         case LDLM_ENQUEUE:
2782                 *process = target_queue_recovery_request(req, obd);
2783                 RETURN(0);
2784
2785         default:
2786                 DEBUG_REQ(D_ERROR, req, "not permitted during recovery");
2787                 *process = -EAGAIN;
2788                 RETURN(0);
2789         }
2790 }
2791
2792 /*
2793  * Handle recovery. Return:
2794  *        +1: continue request processing;
2795  *       -ve: abort immediately with the given error code;
2796  *         0: send reply with error code in req->rq_status;
2797  */
2798 static int mdt_recovery(struct mdt_thread_info *info)
2799 {
2800         struct ptlrpc_request *req = mdt_info_req(info);
2801         struct obd_device *obd;
2802
2803         ENTRY;
2804
2805         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
2806         case MDS_CONNECT:
2807         case SEC_CTX_INIT:
2808         case SEC_CTX_INIT_CONT:
2809         case SEC_CTX_FINI:
2810                 {
2811 #if 0
2812                         int rc;
2813
2814                         rc = mdt_handle_idmap(info);
2815                         if (rc)
2816                                 RETURN(rc);
2817                         else
2818 #endif
2819                                 RETURN(+1);
2820                 }
2821         }
2822
2823         if (unlikely(!class_connected_export(req->rq_export))) {
2824                 CERROR("operation %d on unconnected MDS from %s\n",
2825                        lustre_msg_get_opc(req->rq_reqmsg),
2826                        libcfs_id2str(req->rq_peer));
2827                 /* FIXME: For CMD cleanup, when mds_B stop, the req from
2828                  * mds_A will get -ENOTCONN(especially for ping req),
2829                  * which will cause that mds_A deactive timeout, then when
2830                  * mds_A cleanup, the cleanup process will be suspended since
2831                  * deactive timeout is not zero.
2832                  */
2833                 req->rq_status = -ENOTCONN;
2834                 target_send_reply(req, -ENOTCONN, info->mti_fail_id);
2835                 RETURN(0);
2836         }
2837
2838         /* sanity check: if the xid matches, the request must be marked as a
2839          * resent or replayed */
2840         if (req_xid_is_last(req)) {
2841                 if (!(lustre_msg_get_flags(req->rq_reqmsg) &
2842                       (MSG_RESENT | MSG_REPLAY))) {
2843                         DEBUG_REQ(D_WARNING, req, "rq_xid "LPU64" matches last_xid, "
2844                                   "expected REPLAY or RESENT flag (%x)", req->rq_xid,
2845                                   lustre_msg_get_flags(req->rq_reqmsg));
2846                         LBUG();
2847                         req->rq_status = -ENOTCONN;
2848                         RETURN(-ENOTCONN);
2849                 }
2850         }
2851
2852         /* else: note the opposite is not always true; a RESENT req after a
2853          * failover will usually not match the last_xid, since it was likely
2854          * never committed. A REPLAYed request will almost never match the
2855          * last xid, however it could for a committed, but still retained,
2856          * open. */
2857
2858         obd = req->rq_export->exp_obd;
2859
2860         /* Check for aborted recovery... */
2861         if (unlikely(obd->obd_recovering)) {
2862                 int rc;
2863                 int should_process;
2864                 DEBUG_REQ(D_INFO, req, "Got new replay");
2865                 rc = mdt_filter_recovery_request(req, obd, &should_process);
2866                 if (rc != 0 || !should_process)
2867                         RETURN(rc);
2868                 else if (should_process < 0) {
2869                         req->rq_status = should_process;
2870                         rc = ptlrpc_error(req);
2871                         RETURN(rc);
2872                 }
2873         }
2874         RETURN(+1);
2875 }
2876
2877 static int mdt_msg_check_version(struct lustre_msg *msg)
2878 {
2879         int rc;
2880
2881         switch (lustre_msg_get_opc(msg)) {
2882         case MDS_CONNECT:
2883         case MDS_DISCONNECT:
2884         case OBD_PING:
2885         case SEC_CTX_INIT:
2886         case SEC_CTX_INIT_CONT:
2887         case SEC_CTX_FINI:
2888                 rc = lustre_msg_check_version(msg, LUSTRE_OBD_VERSION);
2889                 if (rc)
2890                         CERROR("bad opc %u version %08x, expecting %08x\n",
2891                                lustre_msg_get_opc(msg),
2892                                lustre_msg_get_version(msg),
2893                                LUSTRE_OBD_VERSION);
2894                 break;
2895         case MDS_GETSTATUS:
2896         case MDS_GETATTR:
2897         case MDS_GETATTR_NAME:
2898         case MDS_STATFS:
2899         case MDS_READPAGE:
2900         case MDS_WRITEPAGE:
2901         case MDS_IS_SUBDIR:
2902         case MDS_REINT:
2903         case MDS_CLOSE:
2904         case MDS_DONE_WRITING:
2905         case MDS_PIN:
2906         case MDS_SYNC:
2907         case MDS_GETXATTR:
2908         case MDS_SETXATTR:
2909         case MDS_SET_INFO:
2910         case MDS_GET_INFO:
2911         case MDS_QUOTACHECK:
2912         case MDS_QUOTACTL:
2913         case QUOTA_DQACQ:
2914         case QUOTA_DQREL:
2915         case SEQ_QUERY:
2916         case FLD_QUERY:
2917                 rc = lustre_msg_check_version(msg, LUSTRE_MDS_VERSION);
2918                 if (rc)
2919                         CERROR("bad opc %u version %08x, expecting %08x\n",
2920                                lustre_msg_get_opc(msg),
2921                                lustre_msg_get_version(msg),
2922                                LUSTRE_MDS_VERSION);
2923                 break;
2924         case LDLM_ENQUEUE:
2925         case LDLM_CONVERT:
2926         case LDLM_BL_CALLBACK:
2927         case LDLM_CP_CALLBACK:
2928                 rc = lustre_msg_check_version(msg, LUSTRE_DLM_VERSION);
2929                 if (rc)
2930                         CERROR("bad opc %u version %08x, expecting %08x\n",
2931                                lustre_msg_get_opc(msg),
2932                                lustre_msg_get_version(msg),
2933                                LUSTRE_DLM_VERSION);
2934                 break;
2935         case OBD_LOG_CANCEL:
2936         case LLOG_ORIGIN_HANDLE_CREATE:
2937         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
2938         case LLOG_ORIGIN_HANDLE_READ_HEADER:
2939         case LLOG_ORIGIN_HANDLE_CLOSE:
2940         case LLOG_ORIGIN_HANDLE_DESTROY:
2941         case LLOG_ORIGIN_HANDLE_PREV_BLOCK:
2942         case LLOG_CATINFO:
2943                 rc = lustre_msg_check_version(msg, LUSTRE_LOG_VERSION);
2944                 if (rc)
2945                         CERROR("bad opc %u version %08x, expecting %08x\n",
2946                                lustre_msg_get_opc(msg),
2947                                lustre_msg_get_version(msg),
2948                                LUSTRE_LOG_VERSION);
2949                 break;
2950         default:
2951                 CERROR("MDS unknown opcode %d\n", lustre_msg_get_opc(msg));
2952                 rc = -ENOTSUPP;
2953         }
2954         return rc;
2955 }
2956
2957 static int mdt_handle0(struct ptlrpc_request *req,
2958                        struct mdt_thread_info *info,
2959                        struct mdt_opc_slice *supported)
2960 {
2961         struct mdt_handler *h;
2962         struct lustre_msg  *msg;
2963         int                 rc;
2964
2965         ENTRY;
2966
2967         if (OBD_FAIL_CHECK_ORSET(OBD_FAIL_MDS_ALL_REQUEST_NET, OBD_FAIL_ONCE))
2968                 RETURN(0);
2969
2970         LASSERT(current->journal_info == NULL);
2971
2972         msg = req->rq_reqmsg;
2973         rc = mdt_msg_check_version(msg);
2974         if (likely(rc == 0)) {
2975                 rc = mdt_recovery(info);
2976                 if (likely(rc == +1)) {
2977                         h = mdt_handler_find(lustre_msg_get_opc(msg),
2978                                              supported);
2979                         if (likely(h != NULL)) {
2980                                 rc = mdt_req_handle(info, h, req);
2981                         } else {
2982                                 CERROR("The unsupported opc: 0x%x\n",
2983                                        lustre_msg_get_opc(msg) );
2984                                 req->rq_status = -ENOTSUPP;
2985                                 rc = ptlrpc_error(req);
2986                                 RETURN(rc);
2987                         }
2988                 }
2989         } else
2990                 CERROR(LUSTRE_MDT_NAME" drops mal-formed request\n");
2991         RETURN(rc);
2992 }
2993
2994 /*
2995  * MDT handler function called by ptlrpc service thread when request comes.
2996  *
2997  * XXX common "target" functionality should be factored into separate module
2998  * shared by mdt, ost and stand-alone services like fld.
2999  */
3000 static int mdt_handle_common(struct ptlrpc_request *req,
3001                              struct mdt_opc_slice *supported)
3002 {
3003         struct lu_env          *env;
3004         struct mdt_thread_info *info;
3005         int                     rc;
3006         ENTRY;
3007
3008         env = req->rq_svc_thread->t_env;
3009         LASSERT(env != NULL);
3010         LASSERT(env->le_ses != NULL);
3011         LASSERT(env->le_ctx.lc_thread == req->rq_svc_thread);
3012         info = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
3013         LASSERT(info != NULL);
3014
3015         mdt_thread_info_init(req, info);
3016
3017         rc = mdt_handle0(req, info, supported);
3018
3019         mdt_thread_info_fini(info);
3020         RETURN(rc);
3021 }
3022
3023 /*
3024  * This is called from recovery code as handler of _all_ RPC types, FLD and SEQ
3025  * as well.
3026  */
3027 int mdt_recovery_handle(struct ptlrpc_request *req)
3028 {
3029         int rc;
3030         ENTRY;
3031
3032         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
3033         case FLD_QUERY:
3034                 rc = mdt_handle_common(req, mdt_fld_handlers);
3035                 break;
3036         case SEQ_QUERY:
3037                 rc = mdt_handle_common(req, mdt_seq_handlers);
3038                 break;
3039         default:
3040                 rc = mdt_handle_common(req, mdt_regular_handlers);
3041                 break;
3042         }
3043
3044         RETURN(rc);
3045 }
3046
3047 static int mdt_regular_handle(struct ptlrpc_request *req)
3048 {
3049         return mdt_handle_common(req, mdt_regular_handlers);
3050 }
3051
3052 static int mdt_readpage_handle(struct ptlrpc_request *req)
3053 {
3054         return mdt_handle_common(req, mdt_readpage_handlers);
3055 }
3056
3057 static int mdt_xmds_handle(struct ptlrpc_request *req)
3058 {
3059         return mdt_handle_common(req, mdt_xmds_handlers);
3060 }
3061
3062 static int mdt_mdsc_handle(struct ptlrpc_request *req)
3063 {
3064         return mdt_handle_common(req, mdt_seq_handlers);
3065 }
3066
3067 static int mdt_mdss_handle(struct ptlrpc_request *req)
3068 {
3069         return mdt_handle_common(req, mdt_seq_handlers);
3070 }
3071
3072 static int mdt_dtss_handle(struct ptlrpc_request *req)
3073 {
3074         return mdt_handle_common(req, mdt_seq_handlers);
3075 }
3076
3077 static int mdt_fld_handle(struct ptlrpc_request *req)
3078 {
3079         return mdt_handle_common(req, mdt_fld_handlers);
3080 }
3081
3082 enum mdt_it_code {
3083         MDT_IT_OPEN,
3084         MDT_IT_OCREAT,
3085         MDT_IT_CREATE,
3086         MDT_IT_GETATTR,
3087         MDT_IT_READDIR,
3088         MDT_IT_LOOKUP,
3089         MDT_IT_UNLINK,
3090         MDT_IT_TRUNC,
3091         MDT_IT_GETXATTR,
3092         MDT_IT_LAYOUT,
3093         MDT_IT_NR
3094 };
3095
3096 static int mdt_intent_getattr(enum mdt_it_code opcode,
3097                               struct mdt_thread_info *info,
3098                               struct ldlm_lock **,
3099                               int);
3100 static int mdt_intent_reint(enum mdt_it_code opcode,
3101                             struct mdt_thread_info *info,
3102                             struct ldlm_lock **,
3103                             int);
3104
3105 static struct mdt_it_flavor {
3106         const struct req_format *it_fmt;
3107         __u32                    it_flags;
3108         int                    (*it_act)(enum mdt_it_code ,
3109                                          struct mdt_thread_info *,
3110                                          struct ldlm_lock **,
3111                                          int);
3112         long                     it_reint;
3113 } mdt_it_flavor[] = {
3114         [MDT_IT_OPEN]     = {
3115                 .it_fmt   = &RQF_LDLM_INTENT,
3116                 /*.it_flags = HABEO_REFERO,*/
3117                 .it_flags = 0,
3118                 .it_act   = mdt_intent_reint,
3119                 .it_reint = REINT_OPEN
3120         },
3121         [MDT_IT_OCREAT]   = {
3122                 .it_fmt   = &RQF_LDLM_INTENT,
3123                 .it_flags = MUTABOR,
3124                 .it_act   = mdt_intent_reint,
3125                 .it_reint = REINT_OPEN
3126         },
3127         [MDT_IT_CREATE]   = {
3128                 .it_fmt   = &RQF_LDLM_INTENT,
3129                 .it_flags = MUTABOR,
3130                 .it_act   = mdt_intent_reint,
3131                 .it_reint = REINT_CREATE
3132         },
3133         [MDT_IT_GETATTR]  = {
3134                 .it_fmt   = &RQF_LDLM_INTENT_GETATTR,
3135                 .it_flags = HABEO_REFERO,
3136                 .it_act   = mdt_intent_getattr
3137         },
3138         [MDT_IT_READDIR]  = {
3139                 .it_fmt   = NULL,
3140                 .it_flags = 0,
3141                 .it_act   = NULL
3142         },
3143         [MDT_IT_LOOKUP]   = {
3144                 .it_fmt   = &RQF_LDLM_INTENT_GETATTR,
3145                 .it_flags = HABEO_REFERO,
3146                 .it_act   = mdt_intent_getattr
3147         },
3148         [MDT_IT_UNLINK]   = {
3149                 .it_fmt   = &RQF_LDLM_INTENT_UNLINK,
3150                 .it_flags = MUTABOR,
3151                 .it_act   = NULL,
3152                 .it_reint = REINT_UNLINK
3153         },
3154         [MDT_IT_TRUNC]    = {
3155                 .it_fmt   = NULL,
3156                 .it_flags = MUTABOR,
3157                 .it_act   = NULL
3158         },
3159         [MDT_IT_GETXATTR] = {
3160                 .it_fmt   = NULL,
3161                 .it_flags = 0,
3162                 .it_act   = NULL
3163         },
3164         [MDT_IT_LAYOUT] = {
3165                 .it_fmt   = &RQF_LDLM_INTENT_GETATTR,
3166                 .it_flags = HABEO_REFERO,
3167                 .it_act   = mdt_intent_getattr
3168         }
3169 };
3170
3171 int mdt_intent_lock_replace(struct mdt_thread_info *info,
3172                             struct ldlm_lock **lockp,
3173                             struct ldlm_lock *new_lock,
3174                             struct mdt_lock_handle *lh,
3175                             int flags)
3176 {
3177         struct ptlrpc_request  *req = mdt_info_req(info);
3178         struct ldlm_lock       *lock = *lockp;
3179
3180         /*
3181          * Get new lock only for cases when possible resent did not find any
3182          * lock.
3183          */
3184         if (new_lock == NULL)
3185                 new_lock = ldlm_handle2lock_long(&lh->mlh_reg_lh, 0);
3186
3187         if (new_lock == NULL && (flags & LDLM_FL_INTENT_ONLY)) {
3188                 lh->mlh_reg_lh.cookie = 0;
3189                 RETURN(0);
3190         }
3191
3192         LASSERTF(new_lock != NULL,
3193                  "lockh "LPX64"\n", lh->mlh_reg_lh.cookie);
3194
3195         /*
3196          * If we've already given this lock to a client once, then we should
3197          * have no readers or writers.  Otherwise, we should have one reader
3198          * _or_ writer ref (which will be zeroed below) before returning the
3199          * lock to a client.
3200          */
3201         if (new_lock->l_export == req->rq_export) {
3202                 LASSERT(new_lock->l_readers + new_lock->l_writers == 0);
3203         } else {
3204                 LASSERT(new_lock->l_export == NULL);
3205                 LASSERT(new_lock->l_readers + new_lock->l_writers == 1);
3206         }
3207
3208         *lockp = new_lock;
3209
3210         if (new_lock->l_export == req->rq_export) {
3211                 /*
3212                  * Already gave this to the client, which means that we
3213                  * reconstructed a reply.
3214                  */
3215                 LASSERT(lustre_msg_get_flags(req->rq_reqmsg) &
3216                         MSG_RESENT);
3217                 lh->mlh_reg_lh.cookie = 0;
3218                 RETURN(ELDLM_LOCK_REPLACED);
3219         }
3220
3221         /*
3222          * Fixup the lock to be given to the client.
3223          */
3224         lock_res_and_lock(new_lock);
3225         /* Zero new_lock->l_readers and new_lock->l_writers without triggering
3226          * possible blocking AST. */
3227         while (new_lock->l_readers > 0) {
3228                 lu_ref_del(&new_lock->l_reference, "reader", new_lock);
3229                 lu_ref_del(&new_lock->l_reference, "user", new_lock);
3230                 new_lock->l_readers--;
3231         }
3232         while (new_lock->l_writers > 0) {
3233                 lu_ref_del(&new_lock->l_reference, "writer", new_lock);
3234                 lu_ref_del(&new_lock->l_reference, "user", new_lock);
3235                 new_lock->l_writers--;
3236         }
3237
3238         new_lock->l_export = class_export_lock_get(req->rq_export, new_lock);
3239         new_lock->l_blocking_ast = lock->l_blocking_ast;
3240         new_lock->l_completion_ast = lock->l_completion_ast;
3241         new_lock->l_remote_handle = lock->l_remote_handle;
3242         new_lock->l_flags &= ~LDLM_FL_LOCAL;
3243
3244         unlock_res_and_lock(new_lock);
3245
3246         cfs_hash_add(new_lock->l_export->exp_lock_hash,
3247                      &new_lock->l_remote_handle,
3248                      &new_lock->l_exp_hash);
3249
3250         LDLM_LOCK_RELEASE(new_lock);
3251         lh->mlh_reg_lh.cookie = 0;
3252
3253         RETURN(ELDLM_LOCK_REPLACED);
3254 }
3255
3256 static void mdt_intent_fixup_resent(struct mdt_thread_info *info,
3257                                     struct ldlm_lock *new_lock,
3258                                     struct ldlm_lock **old_lock,
3259                                     struct mdt_lock_handle *lh)
3260 {
3261         struct ptlrpc_request  *req = mdt_info_req(info);
3262         struct obd_export      *exp = req->rq_export;
3263         struct lustre_handle    remote_hdl;
3264         struct ldlm_request    *dlmreq;
3265         struct ldlm_lock       *lock;
3266
3267         if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT))
3268                 return;
3269
3270         dlmreq = req_capsule_client_get(info->mti_pill, &RMF_DLM_REQ);
3271         remote_hdl = dlmreq->lock_handle[0];
3272
3273         lock = cfs_hash_lookup(exp->exp_lock_hash, &remote_hdl);
3274         if (lock) {
3275                 if (lock != new_lock) {
3276                         lh->mlh_reg_lh.cookie = lock->l_handle.h_cookie;
3277                         lh->mlh_reg_mode = lock->l_granted_mode;
3278
3279                         LDLM_DEBUG(lock, "Restoring lock cookie");
3280                         DEBUG_REQ(D_DLMTRACE, req,
3281                                   "restoring lock cookie "LPX64,
3282                                   lh->mlh_reg_lh.cookie);
3283                         if (old_lock)
3284                                 *old_lock = LDLM_LOCK_GET(lock);
3285                         cfs_hash_put(exp->exp_lock_hash, &lock->l_exp_hash);
3286                         return;
3287                 }
3288
3289                 cfs_hash_put(exp->exp_lock_hash, &lock->l_exp_hash);
3290         }
3291
3292         /*
3293          * If the xid matches, then we know this is a resent request, and allow
3294          * it. (It's probably an OPEN, for which we don't send a lock.
3295          */
3296         if (req_xid_is_last(req))
3297                 return;
3298
3299         /*
3300          * This remote handle isn't enqueued, so we never received or processed
3301          * this request.  Clear MSG_RESENT, because it can be handled like any
3302          * normal request now.
3303          */
3304         lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
3305
3306         DEBUG_REQ(D_DLMTRACE, req, "no existing lock with rhandle "LPX64,
3307                   remote_hdl.cookie);
3308 }
3309
3310 static int mdt_intent_getattr(enum mdt_it_code opcode,
3311                               struct mdt_thread_info *info,
3312                               struct ldlm_lock **lockp,
3313                               int flags)
3314 {
3315         struct mdt_lock_handle *lhc = &info->mti_lh[MDT_LH_RMT];
3316         struct ldlm_lock       *new_lock = NULL;
3317         __u64                   child_bits;
3318         struct ldlm_reply      *ldlm_rep;
3319         struct ptlrpc_request  *req;
3320         struct mdt_body        *reqbody;
3321         struct mdt_body        *repbody;
3322         int                     rc, rc2;
3323         ENTRY;
3324
3325         reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY);
3326         LASSERT(reqbody);
3327
3328         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
3329         LASSERT(repbody);
3330
3331         info->mti_spec.sp_ck_split = !!(reqbody->valid & OBD_MD_FLCKSPLIT);
3332         info->mti_cross_ref = !!(reqbody->valid & OBD_MD_FLCROSSREF);
3333         repbody->eadatasize = 0;
3334         repbody->aclsize = 0;
3335
3336         switch (opcode) {
3337         case MDT_IT_LOOKUP:
3338                 child_bits = MDS_INODELOCK_LOOKUP;
3339                 break;
3340         case MDT_IT_GETATTR:
3341                 child_bits = MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE;
3342                 break;
3343         case MDT_IT_LAYOUT: {
3344                 static int printed = 0;
3345
3346                 if (!printed) {
3347                         CERROR("layout lock not supported by this version\n");
3348                         printed = 1;
3349                 }
3350                 GOTO(out_shrink, rc = -EINVAL);
3351                 break;
3352         }
3353         default:
3354                 CERROR("Unsupported intent (%d)\n", opcode);
3355                 GOTO(out_shrink, rc = -EINVAL);
3356         }
3357
3358         rc = mdt_init_ucred(info, reqbody);
3359         if (rc)
3360                 GOTO(out_shrink, rc);
3361
3362         req = info->mti_pill->rc_req;
3363         ldlm_rep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP);
3364         mdt_set_disposition(info, ldlm_rep, DISP_IT_EXECD);
3365
3366         /* Get lock from request for possible resent case. */
3367         mdt_intent_fixup_resent(info, *lockp, &new_lock, lhc);
3368
3369         ldlm_rep->lock_policy_res2 =
3370                 mdt_getattr_name_lock(info, lhc, child_bits, ldlm_rep);
3371
3372         if (mdt_get_disposition(ldlm_rep, DISP_LOOKUP_NEG))
3373                 ldlm_rep->lock_policy_res2 = 0;
3374         if (!mdt_get_disposition(ldlm_rep, DISP_LOOKUP_POS) ||
3375             ldlm_rep->lock_policy_res2) {
3376                 lhc->mlh_reg_lh.cookie = 0ull;
3377                 GOTO(out_ucred, rc = ELDLM_LOCK_ABORTED);
3378         }
3379
3380         rc = mdt_intent_lock_replace(info, lockp, new_lock, lhc, flags);
3381         EXIT;
3382 out_ucred:
3383         mdt_exit_ucred(info);
3384 out_shrink:
3385         mdt_client_compatibility(info);
3386         rc2 = mdt_fix_reply(info);
3387         if (rc == 0)
3388                 rc = rc2;
3389         return rc;
3390 }
3391
3392 static int mdt_intent_reint(enum mdt_it_code opcode,
3393                             struct mdt_thread_info *info,
3394                             struct ldlm_lock **lockp,
3395                             int flags)
3396 {
3397         struct mdt_lock_handle *lhc = &info->mti_lh[MDT_LH_RMT];
3398         struct ldlm_reply      *rep = NULL;
3399         long                    opc;
3400         int                     rc;
3401
3402         static const struct req_format *intent_fmts[REINT_MAX] = {
3403                 [REINT_CREATE]  = &RQF_LDLM_INTENT_CREATE,
3404                 [REINT_OPEN]    = &RQF_LDLM_INTENT_OPEN
3405         };
3406
3407         ENTRY;
3408
3409         opc = mdt_reint_opcode(info, intent_fmts);
3410         if (opc < 0)
3411                 RETURN(opc);
3412
3413         if (mdt_it_flavor[opcode].it_reint != opc) {
3414                 CERROR("Reint code %ld doesn't match intent: %d\n",
3415                        opc, opcode);
3416                 RETURN(err_serious(-EPROTO));
3417         }
3418
3419         /* Get lock from request for possible resent case. */
3420         mdt_intent_fixup_resent(info, *lockp, NULL, lhc);
3421
3422         rc = mdt_reint_internal(info, lhc, opc);
3423
3424         /* Check whether the reply has been packed successfully. */
3425         if (mdt_info_req(info)->rq_repmsg != NULL)
3426                 rep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP);
3427         if (rep == NULL)
3428                 RETURN(err_serious(-EFAULT));
3429
3430         /* MDC expects this in any case */
3431         if (rc != 0)
3432                 mdt_set_disposition(info, rep, DISP_LOOKUP_EXECD);
3433
3434         /* Cross-ref case, the lock should be returned to the client */
3435         if (rc == -EREMOTE) {
3436                 LASSERT(lustre_handle_is_used(&lhc->mlh_reg_lh));
3437                 rep->lock_policy_res2 = 0;
3438                 rc = mdt_intent_lock_replace(info, lockp, NULL, lhc, flags);
3439                 RETURN(rc);
3440         }
3441         rep->lock_policy_res2 = clear_serious(rc);
3442
3443         if (rep->lock_policy_res2 == -ENOENT &&
3444             mdt_get_disposition(rep, DISP_LOOKUP_NEG))
3445                 rep->lock_policy_res2 = 0;
3446
3447         if (rc == -ENOTCONN || rc == -ENODEV ||
3448             rc == -EOVERFLOW) { /**< if VBR failure then return error */
3449                 /*
3450                  * If it is the disconnect error (ENODEV & ENOCONN), the error
3451                  * will be returned by rq_status, and client at ptlrpc layer
3452                  * will detect this, then disconnect, reconnect the import
3453                  * immediately, instead of impacting the following the rpc.
3454                  */
3455                 lhc->mlh_reg_lh.cookie = 0ull;
3456                 RETURN(rc);
3457         } else {
3458                 /*
3459                  * For other cases, the error will be returned by intent.
3460                  * and client will retrieve the result from intent.
3461                  */
3462                  /*
3463                   * FIXME: when open lock is finished, that should be
3464                   * checked here.
3465                   */
3466                 if (lustre_handle_is_used(&lhc->mlh_reg_lh)) {
3467                         LASSERTF(rc == 0, "Error occurred but lock handle "
3468                                  "is still in use\n");
3469                         rep->lock_policy_res2 = 0;
3470                         rc = mdt_intent_lock_replace(info, lockp, NULL, lhc, flags);
3471                         RETURN(rc);
3472                 } else {
3473                         lhc->mlh_reg_lh.cookie = 0ull;
3474                         RETURN(ELDLM_LOCK_ABORTED);
3475                 }
3476         }
3477 }
3478
3479 static int mdt_intent_code(long itcode)
3480 {
3481         int rc;
3482
3483         switch(itcode) {
3484         case IT_OPEN:
3485                 rc = MDT_IT_OPEN;
3486                 break;
3487         case IT_OPEN|IT_CREAT:
3488                 rc = MDT_IT_OCREAT;
3489                 break;
3490         case IT_CREAT:
3491                 rc = MDT_IT_CREATE;
3492                 break;
3493         case IT_READDIR:
3494                 rc = MDT_IT_READDIR;
3495                 break;
3496         case IT_GETATTR:
3497                 rc = MDT_IT_GETATTR;
3498                 break;
3499         case IT_LOOKUP:
3500                 rc = MDT_IT_LOOKUP;
3501                 break;
3502         case IT_UNLINK:
3503                 rc = MDT_IT_UNLINK;
3504                 break;
3505         case IT_TRUNC:
3506                 rc = MDT_IT_TRUNC;
3507                 break;
3508         case IT_GETXATTR:
3509                 rc = MDT_IT_GETXATTR;
3510                 break;
3511         case IT_LAYOUT:
3512                 rc = MDT_IT_LAYOUT;
3513                 break;
3514         default:
3515                 CERROR("Unknown intent opcode: %ld\n", itcode);
3516                 rc = -EINVAL;
3517                 break;
3518         }
3519         return rc;
3520 }
3521
3522 static int mdt_intent_opc(long itopc, struct mdt_thread_info *info,
3523                           struct ldlm_lock **lockp, int flags)
3524 {
3525         struct req_capsule   *pill;
3526         struct mdt_it_flavor *flv;
3527         int opc;
3528         int rc;
3529         ENTRY;
3530
3531         opc = mdt_intent_code(itopc);
3532         if (opc < 0)
3533                 RETURN(-EINVAL);
3534
3535         pill = info->mti_pill;
3536         flv  = &mdt_it_flavor[opc];
3537
3538         if (flv->it_fmt != NULL)
3539                 req_capsule_extend(pill, flv->it_fmt);
3540
3541         rc = mdt_unpack_req_pack_rep(info, flv->it_flags);
3542         if (rc == 0) {
3543                 struct ptlrpc_request *req = mdt_info_req(info);
3544                 if (flv->it_flags & MUTABOR &&
3545                     req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY)
3546                         RETURN(-EROFS);
3547         }
3548         if (rc == 0 && flv->it_act != NULL) {
3549                 /* execute policy */
3550                 rc = flv->it_act(opc, info, lockp, flags);
3551         } else {
3552                 rc = -EOPNOTSUPP;
3553         }
3554         RETURN(rc);
3555 }
3556
3557 static int mdt_intent_policy(struct ldlm_namespace *ns,
3558                              struct ldlm_lock **lockp, void *req_cookie,
3559                              ldlm_mode_t mode, int flags, void *data)
3560 {
3561         struct mdt_thread_info *info;
3562         struct ptlrpc_request  *req  =  req_cookie;
3563         struct ldlm_intent     *it;
3564         struct req_capsule     *pill;
3565         int rc;
3566
3567         ENTRY;
3568
3569         LASSERT(req != NULL);
3570
3571         info = lu_context_key_get(&req->rq_svc_thread->t_env->le_ctx,
3572                                   &mdt_thread_key);
3573         LASSERT(info != NULL);
3574         pill = info->mti_pill;
3575         LASSERT(pill->rc_req == req);
3576
3577         if (req->rq_reqmsg->lm_bufcount > DLM_INTENT_IT_OFF) {
3578                 req_capsule_extend(pill, &RQF_LDLM_INTENT);
3579                 it = req_capsule_client_get(pill, &RMF_LDLM_INTENT);
3580                 if (it != NULL) {
3581                         rc = mdt_intent_opc(it->opc, info, lockp, flags);
3582                         if (rc == 0)
3583                                 rc = ELDLM_OK;
3584
3585                         /* Lock without inodebits makes no sense and will oops
3586                          * later in ldlm. Let's check it now to see if we have
3587                          * ibits corrupted somewhere in mdt_intent_opc().
3588                          * The case for client miss to set ibits has been
3589                          * processed by others. */
3590                         LASSERT(ergo(info->mti_dlm_req->lock_desc.l_resource.\
3591                                         lr_type == LDLM_IBITS,
3592                                      info->mti_dlm_req->lock_desc.\
3593                                         l_policy_data.l_inodebits.bits != 0));
3594                 } else
3595                         rc = err_serious(-EFAULT);
3596         } else {
3597                 /* No intent was provided */
3598                 LASSERT(pill->rc_fmt == &RQF_LDLM_ENQUEUE);
3599                 rc = req_capsule_server_pack(pill);
3600                 if (rc)
3601                         rc = err_serious(rc);
3602         }
3603         RETURN(rc);
3604 }
3605
3606 static int mdt_seq_fini(const struct lu_env *env,
3607                         struct mdt_device *m)
3608 {
3609         struct md_site *ms = mdt_md_site(m);
3610         ENTRY;
3611
3612         if (ms != NULL) {
3613                 if (ms->ms_server_seq) {
3614                         seq_server_fini(ms->ms_server_seq, env);
3615                         OBD_FREE_PTR(ms->ms_server_seq);
3616                         ms->ms_server_seq = NULL;
3617         }
3618
3619                 if (ms->ms_control_seq) {
3620                         seq_server_fini(ms->ms_control_seq, env);
3621                         OBD_FREE_PTR(ms->ms_control_seq);
3622                         ms->ms_control_seq = NULL;
3623         }
3624
3625                 if (ms->ms_client_seq) {
3626                         seq_client_fini(ms->ms_client_seq);
3627                         OBD_FREE_PTR(ms->ms_client_seq);
3628                         ms->ms_client_seq = NULL;
3629                 }
3630         }
3631
3632         RETURN(0);
3633 }
3634
3635 static int mdt_seq_init(const struct lu_env *env,
3636                         const char *uuid,
3637                         struct mdt_device *m)
3638 {
3639         struct md_site *ms;
3640         char *prefix;
3641         int rc;
3642         ENTRY;
3643
3644         ms = mdt_md_site(m);
3645
3646         /*
3647          * This is sequence-controller node. Init seq-controller server on local
3648          * MDT.
3649          */
3650         if (ms->ms_node_id == 0) {
3651                 LASSERT(ms->ms_control_seq == NULL);
3652
3653                 OBD_ALLOC_PTR(ms->ms_control_seq);
3654                 if (ms->ms_control_seq == NULL)
3655                         RETURN(-ENOMEM);
3656
3657                 rc = seq_server_init(ms->ms_control_seq,
3658                                      m->mdt_bottom, uuid,
3659                                      LUSTRE_SEQ_CONTROLLER,
3660                                      ms,
3661                                      env);
3662
3663                 if (rc)
3664                         GOTO(out_seq_fini, rc);
3665
3666                 OBD_ALLOC_PTR(ms->ms_client_seq);
3667                 if (ms->ms_client_seq == NULL)
3668                         GOTO(out_seq_fini, rc = -ENOMEM);
3669
3670                 OBD_ALLOC(prefix, MAX_OBD_NAME + 5);
3671                 if (prefix == NULL) {
3672                         OBD_FREE_PTR(ms->ms_client_seq);
3673                         GOTO(out_seq_fini, rc = -ENOMEM);
3674                 }
3675
3676                 snprintf(prefix, MAX_OBD_NAME + 5, "ctl-%s",
3677                          uuid);
3678
3679                 /*
3680                  * Init seq-controller client after seq-controller server is
3681                  * ready. Pass ms->ms_control_seq to it for direct talking.
3682                  */
3683                 rc = seq_client_init(ms->ms_client_seq, NULL,
3684                                      LUSTRE_SEQ_METADATA, prefix,
3685                                      ms->ms_control_seq);
3686                 OBD_FREE(prefix, MAX_OBD_NAME + 5);
3687
3688                 if (rc)
3689                         GOTO(out_seq_fini, rc);
3690         }
3691
3692         /* Init seq-server on local MDT */
3693         LASSERT(ms->ms_server_seq == NULL);
3694
3695         OBD_ALLOC_PTR(ms->ms_server_seq);
3696         if (ms->ms_server_seq == NULL)
3697                 GOTO(out_seq_fini, rc = -ENOMEM);
3698
3699         rc = seq_server_init(ms->ms_server_seq,
3700                              m->mdt_bottom, uuid,
3701                              LUSTRE_SEQ_SERVER,
3702                              ms,
3703                              env);
3704         if (rc)
3705                 GOTO(out_seq_fini, rc = -ENOMEM);
3706
3707         /* Assign seq-controller client to local seq-server. */
3708         if (ms->ms_node_id == 0) {
3709                 LASSERT(ms->ms_client_seq != NULL);
3710
3711                 rc = seq_server_set_cli(ms->ms_server_seq,
3712                                         ms->ms_client_seq,
3713                                         env);
3714         }
3715
3716         EXIT;
3717 out_seq_fini:
3718         if (rc)
3719                 mdt_seq_fini(env, m);
3720
3721         return rc;
3722 }
3723 /*
3724  * Init client sequence manager which is used by local MDS to talk to sequence
3725  * controller on remote node.
3726  */
3727 static int mdt_seq_init_cli(const struct lu_env *env,
3728                             struct mdt_device *m,
3729                             struct lustre_cfg *cfg)
3730 {
3731         struct md_site    *ms = mdt_md_site(m);
3732         struct obd_device *mdc;
3733         struct obd_uuid   *uuidp, *mdcuuidp;
3734         char              *uuid_str, *mdc_uuid_str;
3735         int                rc;
3736         int                index;
3737         struct mdt_thread_info *info;
3738         char *p, *index_string = lustre_cfg_string(cfg, 2);
3739         ENTRY;
3740
3741         info = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
3742         uuidp = &info->mti_u.uuid[0];
3743         mdcuuidp = &info->mti_u.uuid[1];
3744
3745         LASSERT(index_string);
3746
3747         index = simple_strtol(index_string, &p, 10);
3748         if (*p) {
3749                 CERROR("Invalid index in lustre_cgf, offset 2\n");
3750                 RETURN(-EINVAL);
3751         }
3752
3753         /* check if this is adding the first MDC and controller is not yet
3754          * initialized. */
3755         if (index != 0 || ms->ms_client_seq)
3756                 RETURN(0);
3757
3758         uuid_str = lustre_cfg_string(cfg, 1);
3759         mdc_uuid_str = lustre_cfg_string(cfg, 4);
3760         obd_str2uuid(uuidp, uuid_str);
3761         obd_str2uuid(mdcuuidp, mdc_uuid_str);
3762
3763         mdc = class_find_client_obd(uuidp, LUSTRE_MDC_NAME, mdcuuidp);
3764         if (!mdc) {
3765                 CERROR("can't find controller MDC by uuid %s\n",
3766                        uuid_str);
3767                 rc = -ENOENT;
3768         } else if (!mdc->obd_set_up) {
3769                 CERROR("target %s not set up\n", mdc->obd_name);
3770                 rc = -EINVAL;
3771         } else {
3772                 LASSERT(ms->ms_control_exp);
3773                 OBD_ALLOC_PTR(ms->ms_client_seq);
3774                 if (ms->ms_client_seq != NULL) {
3775                         char *prefix;
3776
3777                         OBD_ALLOC(prefix, MAX_OBD_NAME + 5);
3778                         if (!prefix)
3779                                 RETURN(-ENOMEM);
3780
3781                         snprintf(prefix, MAX_OBD_NAME + 5, "ctl-%s",
3782                                  mdc->obd_name);
3783
3784                         rc = seq_client_init(ms->ms_client_seq,
3785                                              ms->ms_control_exp,
3786                                              LUSTRE_SEQ_METADATA,
3787                                              prefix, NULL);
3788                         OBD_FREE(prefix, MAX_OBD_NAME + 5);
3789                 } else
3790                         rc = -ENOMEM;
3791
3792                 if (rc)
3793                         RETURN(rc);
3794
3795                 LASSERT(ms->ms_server_seq != NULL);
3796                 rc = seq_server_set_cli(ms->ms_server_seq, ms->ms_client_seq,
3797                                         env);
3798         }
3799
3800         RETURN(rc);
3801 }
3802
3803 static void mdt_seq_fini_cli(struct mdt_device *m)
3804 {
3805         struct md_site *ms;
3806
3807         ENTRY;
3808
3809         ms = mdt_md_site(m);
3810
3811         if (ms != NULL) {
3812                 if (ms->ms_server_seq)
3813                         seq_server_set_cli(ms->ms_server_seq,
3814                                    NULL, NULL);
3815
3816                 if (ms->ms_control_exp) {
3817                         class_export_put(ms->ms_control_exp);
3818                         ms->ms_control_exp = NULL;
3819                 }
3820         }
3821         EXIT;
3822 }
3823
3824 /*
3825  * FLD wrappers
3826  */
3827 static int mdt_fld_fini(const struct lu_env *env,
3828                         struct mdt_device *m)
3829 {
3830         struct md_site *ms = mdt_md_site(m);
3831         ENTRY;
3832
3833         if (ms && ms->ms_server_fld) {
3834                 fld_server_fini(ms->ms_server_fld, env);
3835                 OBD_FREE_PTR(ms->ms_server_fld);
3836                 ms->ms_server_fld = NULL;
3837         }
3838
3839         RETURN(0);
3840 }
3841
3842 static int mdt_fld_init(const struct lu_env *env,
3843                         const char *uuid,
3844                         struct mdt_device *m)
3845 {
3846         struct md_site *ms;
3847         int rc;
3848         ENTRY;
3849
3850         ms = mdt_md_site(m);
3851
3852         OBD_ALLOC_PTR(ms->ms_server_fld);
3853         if (ms->ms_server_fld == NULL)
3854                 RETURN(rc = -ENOMEM);
3855
3856         rc = fld_server_init(ms->ms_server_fld,
3857                              m->mdt_bottom, uuid,
3858                              env, ms->ms_node_id);
3859         if (rc) {
3860                 OBD_FREE_PTR(ms->ms_server_fld);
3861                 ms->ms_server_fld = NULL;
3862                 RETURN(rc);
3863         }
3864
3865         RETURN(0);
3866 }
3867
3868 /* device init/fini methods */
3869 static void mdt_stop_ptlrpc_service(struct mdt_device *m)
3870 {
3871         ENTRY;
3872         if (m->mdt_regular_service != NULL) {
3873                 ptlrpc_unregister_service(m->mdt_regular_service);
3874                 m->mdt_regular_service = NULL;
3875         }
3876         if (m->mdt_readpage_service != NULL) {
3877                 ptlrpc_unregister_service(m->mdt_readpage_service);
3878                 m->mdt_readpage_service = NULL;
3879         }
3880         if (m->mdt_xmds_service != NULL) {
3881                 ptlrpc_unregister_service(m->mdt_xmds_service);
3882                 m->mdt_xmds_service = NULL;
3883         }
3884         if (m->mdt_setattr_service != NULL) {
3885                 ptlrpc_unregister_service(m->mdt_setattr_service);
3886                 m->mdt_setattr_service = NULL;
3887         }
3888         if (m->mdt_mdsc_service != NULL) {
3889                 ptlrpc_unregister_service(m->mdt_mdsc_service);
3890                 m->mdt_mdsc_service = NULL;
3891         }
3892         if (m->mdt_mdss_service != NULL) {
3893                 ptlrpc_unregister_service(m->mdt_mdss_service);
3894                 m->mdt_mdss_service = NULL;
3895         }
3896         if (m->mdt_dtss_service != NULL) {
3897                 ptlrpc_unregister_service(m->mdt_dtss_service);
3898                 m->mdt_dtss_service = NULL;
3899         }
3900         if (m->mdt_fld_service != NULL) {
3901                 ptlrpc_unregister_service(m->mdt_fld_service);
3902                 m->mdt_fld_service = NULL;
3903         }
3904         EXIT;
3905 }
3906
3907 static int mdt_start_ptlrpc_service(struct mdt_device *m)
3908 {
3909         int rc;
3910         static struct ptlrpc_service_conf conf;
3911         cfs_proc_dir_entry_t *procfs_entry;
3912         ENTRY;
3913
3914         procfs_entry = m->mdt_md_dev.md_lu_dev.ld_obd->obd_proc_entry;
3915
3916         conf = (typeof(conf)) {
3917                 .psc_nbufs           = MDS_NBUFS,
3918                 .psc_bufsize         = MDS_BUFSIZE,
3919                 .psc_max_req_size    = MDS_MAXREQSIZE,
3920                 .psc_max_reply_size  = MDS_MAXREPSIZE,
3921                 .psc_req_portal      = MDS_REQUEST_PORTAL,
3922                 .psc_rep_portal      = MDC_REPLY_PORTAL,
3923                 .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR,
3924                 /*
3925                  * We'd like to have a mechanism to set this on a per-device
3926                  * basis, but alas...
3927                  */
3928                 .psc_min_threads     = mdt_min_threads,
3929                 .psc_max_threads     = mdt_max_threads,
3930                 .psc_ctx_tags        = LCT_MD_THREAD
3931         };
3932
3933         m->mdt_ldlm_client = &m->mdt_md_dev.md_lu_dev.ld_obd->obd_ldlm_client;
3934         ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL,
3935                            "mdt_ldlm_client", m->mdt_ldlm_client);
3936
3937         m->mdt_regular_service =
3938                 ptlrpc_init_svc_conf(&conf, mdt_regular_handle, LUSTRE_MDT_NAME,
3939                                      procfs_entry, target_print_req,
3940                                      LUSTRE_MDT_NAME);
3941         if (m->mdt_regular_service == NULL)
3942                 RETURN(-ENOMEM);
3943
3944         rc = ptlrpc_start_threads(m->mdt_regular_service);
3945         if (rc)
3946                 GOTO(err_mdt_svc, rc);
3947
3948         /*
3949          * readpage service configuration. Parameters have to be adjusted,
3950          * ideally.
3951          */
3952         conf = (typeof(conf)) {
3953                 .psc_nbufs           = MDS_NBUFS,
3954                 .psc_bufsize         = MDS_BUFSIZE,
3955                 .psc_max_req_size    = MDS_MAXREQSIZE,
3956                 .psc_max_reply_size  = MDS_MAXREPSIZE,
3957                 .psc_req_portal      = MDS_READPAGE_PORTAL,
3958                 .psc_rep_portal      = MDC_REPLY_PORTAL,
3959                 .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR,
3960                 .psc_min_threads     = mdt_min_threads,
3961                 .psc_max_threads     = mdt_max_threads,
3962                 .psc_ctx_tags        = LCT_MD_THREAD
3963         };
3964         m->mdt_readpage_service =
3965                 ptlrpc_init_svc_conf(&conf, mdt_readpage_handle,
3966                                      LUSTRE_MDT_NAME "_readpage",
3967                                      procfs_entry, target_print_req,"mdt_rdpg");
3968
3969         if (m->mdt_readpage_service == NULL) {
3970                 CERROR("failed to start readpage service\n");
3971                 GOTO(err_mdt_svc, rc = -ENOMEM);
3972         }
3973
3974         rc = ptlrpc_start_threads(m->mdt_readpage_service);
3975
3976         /*
3977          * setattr service configuration.
3978          *
3979          * XXX To keep the compatibility with old client(< 2.2), we need to
3980          * preserve this portal for a certain time, it should be removed
3981          * eventually. LU-617.
3982          */
3983         conf = (typeof(conf)) {
3984                 .psc_nbufs           = MDS_NBUFS,
3985                 .psc_bufsize         = MDS_BUFSIZE,
3986                 .psc_max_req_size    = MDS_MAXREQSIZE,
3987                 .psc_max_reply_size  = MDS_MAXREPSIZE,
3988                 .psc_req_portal      = MDS_SETATTR_PORTAL,
3989                 .psc_rep_portal      = MDC_REPLY_PORTAL,
3990                 .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR,
3991                 .psc_min_threads     = mdt_min_threads,
3992                 .psc_max_threads     = mdt_max_threads,
3993                 .psc_ctx_tags        = LCT_MD_THREAD
3994         };
3995
3996         m->mdt_setattr_service =
3997                 ptlrpc_init_svc_conf(&conf, mdt_regular_handle,
3998                                      LUSTRE_MDT_NAME "_setattr",
3999                                      procfs_entry, target_print_req,"mdt_attr");
4000
4001         if (!m->mdt_setattr_service) {
4002                 CERROR("failed to start setattr service\n");
4003                 GOTO(err_mdt_svc, rc = -ENOMEM);
4004         }
4005
4006         rc = ptlrpc_start_threads(m->mdt_setattr_service);
4007         if (rc)
4008                 GOTO(err_mdt_svc, rc);
4009
4010         /*
4011          * sequence controller service configuration
4012          */
4013         conf = (typeof(conf)) {
4014                 .psc_nbufs           = MDS_NBUFS,
4015                 .psc_bufsize         = MDS_BUFSIZE,
4016                 .psc_max_req_size    = SEQ_MAXREQSIZE,
4017                 .psc_max_reply_size  = SEQ_MAXREPSIZE,
4018                 .psc_req_portal      = SEQ_CONTROLLER_PORTAL,
4019                 .psc_rep_portal      = MDC_REPLY_PORTAL,
4020                 .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR,
4021                 .psc_min_threads     = mdt_min_threads,
4022                 .psc_max_threads     = mdt_max_threads,
4023                 .psc_ctx_tags        = LCT_MD_THREAD
4024         };
4025
4026         m->mdt_mdsc_service =
4027                 ptlrpc_init_svc_conf(&conf, mdt_mdsc_handle,
4028                                      LUSTRE_MDT_NAME"_mdsc",
4029                                      procfs_entry, target_print_req,"mdt_mdsc");
4030         if (!m->mdt_mdsc_service) {
4031                 CERROR("failed to start seq controller service\n");
4032                 GOTO(err_mdt_svc, rc = -ENOMEM);
4033         }
4034
4035         rc = ptlrpc_start_threads(m->mdt_mdsc_service);
4036         if (rc)
4037                 GOTO(err_mdt_svc, rc);
4038
4039         /*
4040          * metadata sequence server service configuration
4041          */
4042         conf = (typeof(conf)) {
4043                 .psc_nbufs           = MDS_NBUFS,
4044                 .psc_bufsize         = MDS_BUFSIZE,
4045                 .psc_max_req_size    = SEQ_MAXREQSIZE,
4046                 .psc_max_reply_size  = SEQ_MAXREPSIZE,
4047                 .psc_req_portal      = SEQ_METADATA_PORTAL,
4048                 .psc_rep_portal      = MDC_REPLY_PORTAL,
4049                 .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR,
4050                 .psc_min_threads     = mdt_min_threads,
4051                 .psc_max_threads     = mdt_max_threads,
4052                 .psc_ctx_tags        = LCT_MD_THREAD|LCT_DT_THREAD
4053         };
4054
4055         m->mdt_mdss_service =
4056                 ptlrpc_init_svc_conf(&conf, mdt_mdss_handle,
4057                                      LUSTRE_MDT_NAME"_mdss",
4058                                      procfs_entry, target_print_req,"mdt_mdss");
4059         if (!m->mdt_mdss_service) {
4060                 CERROR("failed to start metadata seq server service\n");
4061                 GOTO(err_mdt_svc, rc = -ENOMEM);
4062         }
4063
4064         rc = ptlrpc_start_threads(m->mdt_mdss_service);
4065         if (rc)
4066                 GOTO(err_mdt_svc, rc);
4067
4068
4069         /*
4070          * Data sequence server service configuration. We want to have really
4071          * cluster-wide sequences space. This is why we start only one sequence
4072          * controller which manages space.
4073          */
4074         conf = (typeof(conf)) {
4075                 .psc_nbufs           = MDS_NBUFS,
4076                 .psc_bufsize         = MDS_BUFSIZE,
4077                 .psc_max_req_size    = SEQ_MAXREQSIZE,
4078                 .psc_max_reply_size  = SEQ_MAXREPSIZE,
4079                 .psc_req_portal      = SEQ_DATA_PORTAL,
4080                 .psc_rep_portal      = OSC_REPLY_PORTAL,
4081                 .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR,
4082                 .psc_min_threads     = mdt_min_threads,
4083                 .psc_max_threads     = mdt_max_threads,
4084                 .psc_ctx_tags        = LCT_MD_THREAD|LCT_DT_THREAD
4085         };
4086
4087         m->mdt_dtss_service =
4088                 ptlrpc_init_svc_conf(&conf, mdt_dtss_handle,
4089                                      LUSTRE_MDT_NAME"_dtss",
4090                                      procfs_entry, target_print_req,"mdt_dtss");
4091         if (!m->mdt_dtss_service) {
4092                 CERROR("failed to start data seq server service\n");
4093                 GOTO(err_mdt_svc, rc = -ENOMEM);
4094         }
4095
4096         rc = ptlrpc_start_threads(m->mdt_dtss_service);
4097         if (rc)
4098                 GOTO(err_mdt_svc, rc);
4099
4100         /* FLD service start */
4101         conf = (typeof(conf)) {
4102                 .psc_nbufs           = MDS_NBUFS,
4103                 .psc_bufsize         = MDS_BUFSIZE,
4104                 .psc_max_req_size    = FLD_MAXREQSIZE,
4105                 .psc_max_reply_size  = FLD_MAXREPSIZE,
4106                 .psc_req_portal      = FLD_REQUEST_PORTAL,
4107                 .psc_rep_portal      = MDC_REPLY_PORTAL,
4108                 .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR,
4109                 .psc_min_threads     = mdt_min_threads,
4110                 .psc_max_threads     = mdt_max_threads,
4111                 .psc_ctx_tags        = LCT_DT_THREAD|LCT_MD_THREAD
4112         };
4113
4114         m->mdt_fld_service =
4115                 ptlrpc_init_svc_conf(&conf, mdt_fld_handle,
4116                                      LUSTRE_MDT_NAME"_fld",
4117                                      procfs_entry, target_print_req, "mdt_fld");
4118         if (!m->mdt_fld_service) {
4119                 CERROR("failed to start fld service\n");
4120                 GOTO(err_mdt_svc, rc = -ENOMEM);
4121         }
4122
4123         rc = ptlrpc_start_threads(m->mdt_fld_service);
4124         if (rc)
4125                 GOTO(err_mdt_svc, rc);
4126
4127         /*
4128          * mds-mds service configuration. Separate portal is used to allow
4129          * mds-mds requests be not blocked during recovery.
4130          */
4131         conf = (typeof(conf)) {
4132                 .psc_nbufs           = MDS_NBUFS,
4133                 .psc_bufsize         = MDS_BUFSIZE,
4134                 .psc_max_req_size    = MDS_MAXREQSIZE,
4135                 .psc_max_reply_size  = MDS_MAXREPSIZE,
4136                 .psc_req_portal      = MDS_MDS_PORTAL,
4137                 .psc_rep_portal      = MDC_REPLY_PORTAL,
4138                 .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR,
4139                 .psc_min_threads     = mdt_min_threads,
4140                 .psc_max_threads     = mdt_max_threads,
4141                 .psc_ctx_tags        = LCT_MD_THREAD
4142         };
4143         m->mdt_xmds_service =
4144                 ptlrpc_init_svc_conf(&conf, mdt_xmds_handle,
4145                                      LUSTRE_MDT_NAME "_mds",
4146                                      procfs_entry, target_print_req,"mdt_xmds");
4147
4148         if (m->mdt_xmds_service == NULL) {
4149                 CERROR("failed to start xmds service\n");
4150                 GOTO(err_mdt_svc, rc = -ENOMEM);
4151         }
4152
4153         rc = ptlrpc_start_threads(m->mdt_xmds_service);
4154         if (rc)
4155                 GOTO(err_mdt_svc, rc);
4156
4157         EXIT;
4158 err_mdt_svc:
4159         if (rc)
4160                 mdt_stop_ptlrpc_service(m);
4161
4162         return rc;
4163 }
4164
4165 static void mdt_stack_fini(const struct lu_env *env,
4166                            struct mdt_device *m, struct lu_device *top)
4167 {
4168         struct obd_device       *obd = mdt2obd_dev(m);
4169         struct lustre_cfg_bufs  *bufs;
4170         struct lustre_cfg       *lcfg;
4171         struct mdt_thread_info  *info;
4172         char flags[3]="";
4173         ENTRY;
4174
4175         info = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
4176         LASSERT(info != NULL);
4177
4178         bufs = &info->mti_u.bufs;
4179         /* process cleanup, pass mdt obd name to get obd umount flags */
4180         lustre_cfg_bufs_reset(bufs, obd->obd_name);
4181         if (obd->obd_force)
4182                 strcat(flags, "F");
4183         if (obd->obd_fail)
4184                 strcat(flags, "A");
4185         lustre_cfg_bufs_set_string(bufs, 1, flags);
4186         lcfg = lustre_cfg_new(LCFG_CLEANUP, bufs);
4187         if (!lcfg) {
4188                 CERROR("Cannot alloc lcfg!\n");
4189                 return;
4190         }
4191
4192         LASSERT(top);
4193         top->ld_ops->ldo_process_config(env, top, lcfg);
4194         lustre_cfg_free(lcfg);
4195
4196         lu_stack_fini(env, top);
4197         m->mdt_child = NULL;
4198         m->mdt_bottom = NULL;
4199 }
4200
4201 static struct lu_device *mdt_layer_setup(struct lu_env *env,
4202                                          const char *typename,
4203                                          struct lu_device *child,
4204                                          struct lustre_cfg *cfg)
4205 {
4206         const char            *dev = lustre_cfg_string(cfg, 0);
4207         struct obd_type       *type;
4208         struct lu_device_type *ldt;
4209         struct lu_device      *d;
4210         int rc;
4211         ENTRY;
4212
4213         /* find the type */
4214         type = class_get_type(typename);
4215         if (!type) {
4216                 CERROR("Unknown type: '%s'\n", typename);
4217                 GOTO(out, rc = -ENODEV);
4218         }
4219
4220         rc = lu_env_refill((struct lu_env *)env);
4221         if (rc != 0) {
4222                 CERROR("Failure to refill session: '%d'\n", rc);
4223                 GOTO(out_type, rc);
4224         }
4225
4226         ldt = type->typ_lu;
4227         if (ldt == NULL) {
4228                 CERROR("type: '%s'\n", typename);
4229                 GOTO(out_type, rc = -EINVAL);
4230         }
4231
4232         ldt->ldt_obd_type = type;
4233         d = ldt->ldt_ops->ldto_device_alloc(env, ldt, cfg);
4234         if (IS_ERR(d)) {
4235                 CERROR("Cannot allocate device: '%s'\n", typename);
4236                 GOTO(out_type, rc = -ENODEV);
4237         }
4238
4239         LASSERT(child->ld_site);
4240         d->ld_site = child->ld_site;
4241
4242         type->typ_refcnt++;
4243         rc = ldt->ldt_ops->ldto_device_init(env, d, dev, child);
4244         if (rc) {
4245                 CERROR("can't init device '%s', rc %d\n", typename, rc);
4246                 GOTO(out_alloc, rc);
4247         }
4248         lu_device_get(d);
4249         lu_ref_add(&d->ld_reference, "lu-stack", &lu_site_init);
4250
4251         cfs_spin_lock(&d->ld_site->ls_ld_lock);
4252         cfs_list_add_tail(&d->ld_linkage, &d->ld_site->ls_ld_linkage);
4253         cfs_spin_unlock(&d->ld_site->ls_ld_lock);
4254
4255         RETURN(d);
4256 out_alloc:
4257         ldt->ldt_ops->ldto_device_free(env, d);
4258         type->typ_refcnt--;
4259 out_type:
4260         class_put_type(type);
4261 out:
4262         return ERR_PTR(rc);
4263 }
4264
4265 static int mdt_stack_init(struct lu_env *env,
4266                           struct mdt_device *m,
4267                           struct lustre_cfg *cfg,
4268                           struct lustre_mount_info  *lmi)
4269 {
4270         struct lu_device  *d = &m->mdt_md_dev.md_lu_dev;
4271         struct lu_device  *tmp;
4272         struct md_device  *md;
4273         struct lu_device  *child_lu_dev;
4274         int rc;
4275         ENTRY;
4276
4277         /* init the stack */
4278         tmp = mdt_layer_setup(env, LUSTRE_OSD_NAME, d, cfg);
4279         if (IS_ERR(tmp)) {
4280                 RETURN(PTR_ERR(tmp));
4281         }
4282         m->mdt_bottom = lu2dt_dev(tmp);
4283         d = tmp;
4284         tmp = mdt_layer_setup(env, LUSTRE_MDD_NAME, d, cfg);
4285         if (IS_ERR(tmp)) {
4286                 GOTO(out, rc = PTR_ERR(tmp));
4287         }
4288         d = tmp;
4289         md = lu2md_dev(d);
4290
4291         tmp = mdt_layer_setup(env, LUSTRE_CMM_NAME, d, cfg);
4292         if (IS_ERR(tmp)) {
4293                 GOTO(out, rc = PTR_ERR(tmp));
4294         }
4295         d = tmp;
4296         /*set mdd upcall device*/
4297         md_upcall_dev_set(md, lu2md_dev(d));
4298
4299         md = lu2md_dev(d);
4300         /*set cmm upcall device*/
4301         md_upcall_dev_set(md, &m->mdt_md_dev);
4302
4303         m->mdt_child = lu2md_dev(d);
4304
4305         /* process setup config */
4306         tmp = &m->mdt_md_dev.md_lu_dev;
4307         rc = tmp->ld_ops->ldo_process_config(env, tmp, cfg);
4308         if (rc)
4309                 GOTO(out, rc);
4310
4311         /* initialize local objects */
4312         child_lu_dev = &m->mdt_child->md_lu_dev;
4313
4314         rc = child_lu_dev->ld_ops->ldo_prepare(env,
4315                                                &m->mdt_md_dev.md_lu_dev,
4316                                                child_lu_dev);
4317 out:
4318         /* fini from last known good lu_device */
4319         if (rc)
4320                 mdt_stack_fini(env, m, d);
4321
4322         return rc;
4323 }
4324
4325 /**
4326  * setup CONFIG_ORIG context, used to access local config log.
4327  * this may need to be rewrite as part of llog rewrite for lu-api.
4328  */
4329 static int mdt_obd_llog_setup(struct obd_device *obd,
4330                               struct lustre_sb_info *lsi)
4331 {
4332         int     rc;
4333
4334         LASSERT(obd->obd_fsops == NULL);
4335
4336         obd->obd_fsops = fsfilt_get_ops(MT_STR(lsi->lsi_ldd));
4337         if (IS_ERR(obd->obd_fsops))
4338                 return PTR_ERR(obd->obd_fsops);
4339
4340         rc = fsfilt_setup(obd, lsi->lsi_srv_mnt->mnt_sb);
4341         if (rc) {
4342                 fsfilt_put_ops(obd->obd_fsops);
4343                 return rc;
4344         }
4345
4346         OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
4347         obd->obd_lvfs_ctxt.pwdmnt = lsi->lsi_srv_mnt;
4348         obd->obd_lvfs_ctxt.pwd = lsi->lsi_srv_mnt->mnt_root;
4349         obd->obd_lvfs_ctxt.fs = get_ds();
4350
4351         rc = llog_setup(obd, &obd->obd_olg, LLOG_CONFIG_ORIG_CTXT, obd,
4352                         0, NULL, &llog_lvfs_ops);
4353         if (rc) {
4354                 CERROR("llog_setup() failed: %d\n", rc);
4355                 fsfilt_put_ops(obd->obd_fsops);
4356         }
4357
4358         return rc;
4359 }
4360
4361 static void mdt_obd_llog_cleanup(struct obd_device *obd)
4362 {
4363         struct llog_ctxt *ctxt;
4364
4365         ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
4366         if (ctxt)
4367                 llog_cleanup(ctxt);
4368
4369         if (obd->obd_fsops) {
4370                 fsfilt_put_ops(obd->obd_fsops);
4371                 obd->obd_fsops = NULL;
4372         }
4373 }
4374
4375 static void mdt_fini(const struct lu_env *env, struct mdt_device *m)
4376 {
4377         struct md_device  *next = m->mdt_child;
4378         struct lu_device  *d    = &m->mdt_md_dev.md_lu_dev;
4379         struct lu_site    *ls   = d->ld_site;
4380         struct obd_device *obd = mdt2obd_dev(m);
4381         ENTRY;
4382
4383         target_recovery_fini(obd);
4384
4385         ping_evictor_stop();
4386
4387         mdt_stop_ptlrpc_service(m);
4388         mdt_llog_ctxt_unclone(env, m, LLOG_CHANGELOG_ORIG_CTXT);
4389         mdt_obd_llog_cleanup(obd);
4390         obd_exports_barrier(obd);
4391         obd_zombie_barrier();
4392
4393         mdt_procfs_fini(m);
4394
4395 #ifdef HAVE_QUOTA_SUPPORT
4396         next->md_ops->mdo_quota.mqo_cleanup(env, next);
4397 #endif
4398         lut_fini(env, &m->mdt_lut);
4399         mdt_fs_cleanup(env, m);
4400         upcall_cache_cleanup(m->mdt_identity_cache);
4401         m->mdt_identity_cache = NULL;
4402
4403         if (m->mdt_namespace != NULL) {
4404                 ldlm_namespace_free(m->mdt_namespace, NULL,
4405                                     d->ld_obd->obd_force);
4406                 d->ld_obd->obd_namespace = m->mdt_namespace = NULL;
4407         }
4408
4409         cfs_free_nidlist(&m->mdt_nosquash_nids);
4410         if (m->mdt_nosquash_str) {
4411                 OBD_FREE(m->mdt_nosquash_str, m->mdt_nosquash_strlen);
4412                 m->mdt_nosquash_str = NULL;
4413                 m->mdt_nosquash_strlen = 0;
4414         }
4415
4416         mdt_seq_fini(env, m);
4417         mdt_seq_fini_cli(m);
4418         mdt_fld_fini(env, m);
4419         sptlrpc_rule_set_free(&m->mdt_sptlrpc_rset);
4420
4421         next->md_ops->mdo_init_capa_ctxt(env, next, 0, 0, 0, NULL);
4422         cfs_timer_disarm(&m->mdt_ck_timer);
4423         mdt_ck_thread_stop(m);
4424
4425         /*
4426          * Finish the stack
4427          */
4428         mdt_stack_fini(env, m, md2lu_dev(m->mdt_child));
4429
4430         if (ls) {
4431                 struct md_site *mite;
4432
4433                 lu_site_fini(ls);
4434                 mite = lu_site2md(ls);
4435                 OBD_FREE_PTR(mite);
4436                 d->ld_site = NULL;
4437         }
4438         LASSERT(cfs_atomic_read(&d->ld_ref) == 0);
4439
4440         EXIT;
4441 }
4442
4443 static int mdt_adapt_sptlrpc_conf(struct obd_device *obd, int initial)
4444 {
4445         struct mdt_device       *m = mdt_dev(obd->obd_lu_dev);
4446         struct sptlrpc_rule_set  tmp_rset;
4447         int                      rc;
4448
4449         sptlrpc_rule_set_init(&tmp_rset);
4450         rc = sptlrpc_conf_target_get_rules(obd, &tmp_rset, initial);
4451         if (rc) {
4452                 CERROR("mdt %s: failed get sptlrpc rules: %d\n",
4453                        obd->obd_name, rc);
4454                 return rc;
4455         }
4456
4457         sptlrpc_target_update_exp_flavor(obd, &tmp_rset);
4458
4459         cfs_write_lock(&m->mdt_sptlrpc_lock);
4460         sptlrpc_rule_set_free(&m->mdt_sptlrpc_rset);
4461         m->mdt_sptlrpc_rset = tmp_rset;
4462         cfs_write_unlock(&m->mdt_sptlrpc_lock);
4463
4464         return 0;
4465 }
4466
4467 int mdt_postrecov(const struct lu_env *, struct mdt_device *);
4468
4469 static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
4470                      struct lu_device_type *ldt, struct lustre_cfg *cfg)
4471 {
4472         struct mdt_thread_info    *info;
4473         struct obd_device         *obd;
4474         const char                *dev = lustre_cfg_string(cfg, 0);
4475         const char                *num = lustre_cfg_string(cfg, 2);
4476         struct lustre_mount_info  *lmi = NULL;
4477         struct lustre_sb_info     *lsi;
4478         struct lustre_disk_data   *ldd;
4479         struct lu_site            *s;
4480         struct md_site            *mite;
4481         const char                *identity_upcall = "NONE";
4482         struct md_device          *next;
4483         int                        rc;
4484         int                        node_id;
4485         mntopt_t                   mntopts;
4486         ENTRY;
4487
4488         md_device_init(&m->mdt_md_dev, ldt);
4489         /*
4490          * Environment (env) might be missing mdt_thread_key values at that
4491          * point, if device is allocated when mdt_thread_key is in QUIESCENT
4492          * mode.
4493          *
4494          * Usually device allocation path doesn't use module key values, but
4495          * mdt has to do a lot of work here, so allocate key value.
4496          */
4497         rc = lu_env_refill((struct lu_env *)env);
4498         if (rc != 0)
4499                 RETURN(rc);
4500
4501         info = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
4502         LASSERT(info != NULL);
4503
4504         obd = class_name2obd(dev);
4505         LASSERT(obd != NULL);
4506
4507         m->mdt_max_mdsize = MAX_MD_SIZE; /* 4 stripes */
4508         m->mdt_max_cookiesize = sizeof(struct llog_cookie);
4509
4510         m->mdt_som_conf = 0;
4511
4512         m->mdt_opts.mo_cos = MDT_COS_DEFAULT;
4513         lmi = server_get_mount_2(dev);
4514         if (lmi == NULL) {
4515                 CERROR("Cannot get mount info for %s!\n", dev);
4516                 RETURN(-EFAULT);
4517         } else {
4518                 lsi = s2lsi(lmi->lmi_sb);
4519                 /* CMD is supported only in IAM mode */
4520                 ldd = lsi->lsi_ldd;
4521                 LASSERT(num);
4522                 node_id = simple_strtol(num, NULL, 10);
4523                 if (!(ldd->ldd_flags & LDD_F_IAM_DIR) && node_id) {
4524                         CERROR("CMD Operation not allowed in IOP mode\n");
4525                         GOTO(err_lmi, rc = -EINVAL);
4526                 }
4527
4528                 obd->u.obt.obt_magic = OBT_MAGIC;
4529         }
4530
4531         cfs_rwlock_init(&m->mdt_sptlrpc_lock);
4532         sptlrpc_rule_set_init(&m->mdt_sptlrpc_rset);
4533
4534         cfs_spin_lock_init(&m->mdt_ioepoch_lock);
4535         m->mdt_opts.mo_compat_resname = 0;
4536         m->mdt_opts.mo_mds_capa = 1;
4537         m->mdt_opts.mo_oss_capa = 1;
4538         m->mdt_capa_timeout = CAPA_TIMEOUT;
4539         m->mdt_capa_alg = CAPA_HMAC_ALG_SHA1;
4540         m->mdt_ck_timeout = CAPA_KEY_TIMEOUT;
4541         m->mdt_squash_uid = 0;
4542         m->mdt_squash_gid = 0;
4543         CFS_INIT_LIST_HEAD(&m->mdt_nosquash_nids);
4544         m->mdt_nosquash_str = NULL;
4545         m->mdt_nosquash_strlen = 0;
4546         cfs_init_rwsem(&m->mdt_squash_sem);
4547
4548         OBD_ALLOC_PTR(mite);
4549         if (mite == NULL)
4550                 GOTO(err_lmi, rc = -ENOMEM);
4551
4552         s = &mite->ms_lu;
4553
4554         m->mdt_md_dev.md_lu_dev.ld_ops = &mdt_lu_ops;
4555         m->mdt_md_dev.md_lu_dev.ld_obd = obd;
4556         /* set this lu_device to obd, because error handling need it */
4557         obd->obd_lu_dev = &m->mdt_md_dev.md_lu_dev;
4558
4559         rc = lu_site_init(s, &m->mdt_md_dev.md_lu_dev);
4560         if (rc) {
4561                 CERROR("Can't init lu_site, rc %d\n", rc);
4562                 GOTO(err_free_site, rc);
4563         }
4564
4565         /* set server index */
4566         lu_site2md(s)->ms_node_id = node_id;
4567
4568         /* failover is the default
4569          * FIXME: we do not failout mds0/mgs, which may cause some problems.
4570          * assumed whose ms_node_id == 0 XXX
4571          * */
4572         obd->obd_replayable = 1;
4573         /* No connection accepted until configurations will finish */
4574         obd->obd_no_conn = 1;
4575
4576         if (cfg->lcfg_bufcount > 4 && LUSTRE_CFG_BUFLEN(cfg, 4) > 0) {
4577                 char *str = lustre_cfg_string(cfg, 4);
4578                 if (strchr(str, 'n')) {
4579                         CWARN("%s: recovery disabled\n", obd->obd_name);
4580                         obd->obd_replayable = 0;
4581                 }
4582         }
4583
4584         /* init the stack */
4585         rc = mdt_stack_init((struct lu_env *)env, m, cfg, lmi);
4586         if (rc) {
4587                 CERROR("Can't init device stack, rc %d\n", rc);
4588                 GOTO(err_lu_site, rc);
4589         }
4590
4591         rc = lut_init(env, &m->mdt_lut, obd, m->mdt_bottom);
4592         if (rc)
4593                 GOTO(err_fini_stack, rc);
4594
4595         rc = mdt_fld_init(env, obd->obd_name, m);
4596         if (rc)
4597                 GOTO(err_lut, rc);
4598
4599         rc = mdt_seq_init(env, obd->obd_name, m);
4600         if (rc)
4601                 GOTO(err_fini_fld, rc);
4602
4603         snprintf(info->mti_u.ns_name, sizeof info->mti_u.ns_name,
4604                  LUSTRE_MDT_NAME"-%p", m);
4605         m->mdt_namespace = ldlm_namespace_new(obd, info->mti_u.ns_name,
4606                                               LDLM_NAMESPACE_SERVER,
4607                                               LDLM_NAMESPACE_GREEDY,
4608                                               LDLM_NS_TYPE_MDT);
4609         if (m->mdt_namespace == NULL)
4610                 GOTO(err_fini_seq, rc = -ENOMEM);
4611
4612         ldlm_register_intent(m->mdt_namespace, mdt_intent_policy);
4613         /* set obd_namespace for compatibility with old code */
4614         obd->obd_namespace = m->mdt_namespace;
4615
4616         cfs_timer_init(&m->mdt_ck_timer, mdt_ck_timer_callback, m);
4617
4618         rc = mdt_ck_thread_start(m);
4619         if (rc)
4620                 GOTO(err_free_ns, rc);
4621
4622         rc = mdt_fs_setup(env, m, obd, lsi);
4623         if (rc)
4624                 GOTO(err_capa, rc);
4625
4626         rc = mdt_obd_llog_setup(obd, lsi);
4627         if (rc)
4628                 GOTO(err_fs_cleanup, rc);
4629
4630         rc = mdt_llog_ctxt_clone(env, m, LLOG_CHANGELOG_ORIG_CTXT);
4631         if (rc)
4632                 GOTO(err_llog_cleanup, rc);
4633
4634         mdt_adapt_sptlrpc_conf(obd, 1);
4635
4636         next = m->mdt_child;
4637 #ifdef HAVE_QUOTA_SUPPORT
4638         rc = next->md_ops->mdo_quota.mqo_setup(env, next, lmi->lmi_mnt);
4639         if (rc)
4640                 GOTO(err_llog_cleanup, rc);
4641 #endif
4642
4643         server_put_mount_2(dev, lmi->lmi_mnt);
4644         lmi = NULL;
4645
4646         rc = next->md_ops->mdo_iocontrol(env, next, OBD_IOC_GET_MNTOPT, 0,
4647                                          &mntopts);
4648         if (rc)
4649                 GOTO(err_quota, rc);
4650
4651         if (mntopts & MNTOPT_USERXATTR)
4652                 m->mdt_opts.mo_user_xattr = 1;
4653         else
4654                 m->mdt_opts.mo_user_xattr = 0;
4655
4656         if (mntopts & MNTOPT_ACL)
4657                 m->mdt_opts.mo_acl = 1;
4658         else
4659                 m->mdt_opts.mo_acl = 0;
4660
4661         /* XXX: to support suppgid for ACL, we enable identity_upcall
4662          * by default, otherwise, maybe got unexpected -EACCESS. */
4663         if (m->mdt_opts.mo_acl)
4664                 identity_upcall = MDT_IDENTITY_UPCALL_PATH;
4665
4666         m->mdt_identity_cache = upcall_cache_init(obd->obd_name,identity_upcall,
4667                                                 &mdt_identity_upcall_cache_ops);
4668         if (IS_ERR(m->mdt_identity_cache)) {
4669                 rc = PTR_ERR(m->mdt_identity_cache);
4670                 m->mdt_identity_cache = NULL;
4671                 GOTO(err_quota, rc);
4672         }
4673
4674         target_recovery_init(&m->mdt_lut, mdt_recovery_handle);
4675
4676         rc = mdt_procfs_init(m, dev);
4677         if (rc) {
4678                 CERROR("Can't init MDT lprocfs, rc %d\n", rc);
4679                 GOTO(err_recovery, rc);
4680         }
4681
4682         rc = mdt_start_ptlrpc_service(m);
4683         if (rc)
4684                 GOTO(err_procfs, rc);
4685
4686         ping_evictor_start();
4687
4688         rc = lu_site_init_finish(s);
4689         if (rc)
4690                 GOTO(err_stop_service, rc);
4691
4692         if (obd->obd_recovering == 0)
4693                 mdt_postrecov(env, m);
4694
4695         mdt_init_capa_ctxt(env, m);
4696
4697         /* Reduce the initial timeout on an MDS because it doesn't need such
4698          * a long timeout as an OST does. Adaptive timeouts will adjust this
4699          * value appropriately. */
4700         if (ldlm_timeout == LDLM_TIMEOUT_DEFAULT)
4701                 ldlm_timeout = MDS_LDLM_TIMEOUT_DEFAULT;
4702
4703         RETURN(0);
4704
4705 err_stop_service:
4706         ping_evictor_stop();
4707         mdt_stop_ptlrpc_service(m);
4708 err_procfs:
4709         mdt_procfs_fini(m);
4710 err_recovery:
4711         target_recovery_fini(obd);
4712         upcall_cache_cleanup(m->mdt_identity_cache);
4713         m->mdt_identity_cache = NULL;
4714 err_quota:
4715 #ifdef HAVE_QUOTA_SUPPORT
4716         next->md_ops->mdo_quota.mqo_cleanup(env, next);
4717 #endif
4718 err_llog_cleanup:
4719         mdt_llog_ctxt_unclone(env, m, LLOG_CHANGELOG_ORIG_CTXT);
4720         mdt_obd_llog_cleanup(obd);
4721 err_fs_cleanup:
4722         mdt_fs_cleanup(env, m);
4723 err_capa:
4724         cfs_timer_disarm(&m->mdt_ck_timer);
4725         mdt_ck_thread_stop(m);
4726 err_free_ns:
4727         ldlm_namespace_free(m->mdt_namespace, NULL, 0);
4728         obd->obd_namespace = m->mdt_namespace = NULL;
4729 err_fini_seq:
4730         mdt_seq_fini(env, m);
4731 err_fini_fld:
4732         mdt_fld_fini(env, m);
4733 err_lut:
4734         lut_fini(env, &m->mdt_lut);
4735 err_fini_stack:
4736         mdt_stack_fini(env, m, md2lu_dev(m->mdt_child));
4737 err_lu_site:
4738         lu_site_fini(s);
4739 err_free_site:
4740         OBD_FREE_PTR(mite);
4741 err_lmi:
4742         if (lmi)
4743                 server_put_mount_2(dev, lmi->lmi_mnt);
4744         return (rc);
4745 }
4746
4747 /* used by MGS to process specific configurations */
4748 static int mdt_process_config(const struct lu_env *env,
4749                               struct lu_device *d, struct lustre_cfg *cfg)
4750 {
4751         struct mdt_device *m = mdt_dev(d);
4752         struct md_device *md_next = m->mdt_child;
4753         struct lu_device *next = md2lu_dev(md_next);
4754         int rc = 0;
4755         ENTRY;
4756
4757         switch (cfg->lcfg_command) {
4758         case LCFG_PARAM: {
4759                 struct lprocfs_static_vars lvars;
4760                 struct obd_device *obd = d->ld_obd;
4761
4762                 /*
4763                  * For interoperability between 1.8 and 2.0,
4764                  */
4765                 {
4766                         /* Skip old "mdt.group_upcall" param. */
4767                         char *param = lustre_cfg_string(cfg, 1);
4768                         if (param && !strncmp("mdt.group_upcall", param, 16)) {
4769                                 CWARN("For 1.8 interoperability, skip this"
4770                                        " mdt.group_upcall. It is obsolete\n");
4771                                 break;
4772                         }
4773                         /* Rename old "mdt.quota_type" to "mdd.quota_type. */
4774                         if (param && !strncmp("mdt.quota_type", param, 14)) {
4775                                 CWARN("Found old param mdt.quota_type, changed"
4776                                       " it to mdd.quota_type.\n");
4777                                 param[2] = 'd';
4778                         }
4779                 }
4780
4781                 lprocfs_mdt_init_vars(&lvars);
4782                 rc = class_process_proc_param(PARAM_MDT, lvars.obd_vars,
4783                                               cfg, obd);
4784                 if (rc > 0 || rc == -ENOSYS)
4785                         /* we don't understand; pass it on */
4786                         rc = next->ld_ops->ldo_process_config(env, next, cfg);
4787                 break;
4788         }
4789         case LCFG_ADD_MDC:
4790                 /*
4791                  * Add mdc hook to get first MDT uuid and connect it to
4792                  * ls->controller to use for seq manager.
4793                  */
4794                 rc = next->ld_ops->ldo_process_config(env, next, cfg);
4795                 if (rc)
4796                         CERROR("Can't add mdc, rc %d\n", rc);
4797                 else
4798                         rc = mdt_seq_init_cli(env, mdt_dev(d), cfg);
4799                 break;
4800         default:
4801                 /* others are passed further */
4802                 rc = next->ld_ops->ldo_process_config(env, next, cfg);
4803                 break;
4804         }
4805         RETURN(rc);
4806 }
4807
4808 static struct lu_object *mdt_object_alloc(const struct lu_env *env,
4809                                           const struct lu_object_header *hdr,
4810                                           struct lu_device *d)
4811 {
4812         struct mdt_object *mo;
4813
4814         ENTRY;
4815
4816         OBD_ALLOC_PTR(mo);
4817         if (mo != NULL) {
4818                 struct lu_object *o;
4819                 struct lu_object_header *h;
4820
4821                 o = &mo->mot_obj.mo_lu;
4822                 h = &mo->mot_header;
4823                 lu_object_header_init(h);
4824                 lu_object_init(o, h, d);
4825                 lu_object_add_top(h, o);
4826                 o->lo_ops = &mdt_obj_ops;
4827                 cfs_mutex_init(&mo->mot_ioepoch_mutex);
4828                 cfs_mutex_init(&mo->mot_lov_mutex);
4829                 RETURN(o);
4830         } else
4831                 RETURN(NULL);
4832 }
4833
4834 static int mdt_object_init(const struct lu_env *env, struct lu_object *o,
4835                            const struct lu_object_conf *unused)
4836 {
4837         struct mdt_device *d = mdt_dev(o->lo_dev);
4838         struct lu_device  *under;
4839         struct lu_object  *below;
4840         int                rc = 0;
4841         ENTRY;
4842
4843         CDEBUG(D_INFO, "object init, fid = "DFID"\n",
4844                PFID(lu_object_fid(o)));
4845
4846         under = &d->mdt_child->md_lu_dev;
4847         below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
4848         if (below != NULL) {
4849                 lu_object_add(o, below);
4850         } else
4851                 rc = -ENOMEM;
4852
4853         RETURN(rc);
4854 }
4855
4856 static void mdt_object_free(const struct lu_env *env, struct lu_object *o)
4857 {
4858         struct mdt_object *mo = mdt_obj(o);
4859         struct lu_object_header *h;
4860         ENTRY;
4861
4862         h = o->lo_header;
4863         CDEBUG(D_INFO, "object free, fid = "DFID"\n",
4864                PFID(lu_object_fid(o)));
4865
4866         lu_object_fini(o);
4867         lu_object_header_fini(h);
4868         OBD_FREE_PTR(mo);
4869         EXIT;
4870 }
4871
4872 static int mdt_object_print(const struct lu_env *env, void *cookie,
4873                             lu_printer_t p, const struct lu_object *o)
4874 {
4875         struct mdt_object *mdto = mdt_obj((struct lu_object *)o);
4876         return (*p)(env, cookie, LUSTRE_MDT_NAME"-object@%p(ioepoch="LPU64" "
4877                     "flags="LPX64", epochcount=%d, writecount=%d)",
4878                     mdto, mdto->mot_ioepoch, mdto->mot_flags,
4879                     mdto->mot_ioepoch_count, mdto->mot_writecount);
4880 }
4881
4882 static const struct lu_device_operations mdt_lu_ops = {
4883         .ldo_object_alloc   = mdt_object_alloc,
4884         .ldo_process_config = mdt_process_config,
4885 };
4886
4887 static const struct lu_object_operations mdt_obj_ops = {
4888         .loo_object_init    = mdt_object_init,
4889         .loo_object_free    = mdt_object_free,
4890         .loo_object_print   = mdt_object_print
4891 };
4892
4893 static int mdt_obd_set_info_async(const struct lu_env *env,
4894                                   struct obd_export *exp,
4895                                   __u32 keylen, void *key,
4896                                   __u32 vallen, void *val,
4897                                   struct ptlrpc_request_set *set)
4898 {
4899         struct obd_device     *obd = exp->exp_obd;
4900         int                    rc;
4901         ENTRY;
4902
4903         LASSERT(obd);
4904
4905         if (KEY_IS(KEY_SPTLRPC_CONF)) {
4906                 rc = mdt_adapt_sptlrpc_conf(obd, 0);
4907                 RETURN(rc);
4908         }
4909
4910         RETURN(0);
4911 }
4912
4913 /* mds_connect_internal */
4914 static int mdt_connect_internal(struct obd_export *exp,
4915                                 struct mdt_device *mdt,
4916                                 struct obd_connect_data *data)
4917 {
4918         if (data != NULL) {
4919                 data->ocd_connect_flags &= MDT_CONNECT_SUPPORTED;
4920                 data->ocd_ibits_known &= MDS_INODELOCK_FULL;
4921
4922                 /* If no known bits (which should not happen, probably,
4923                    as everybody should support LOOKUP and UPDATE bits at least)
4924                    revert to compat mode with plain locks. */
4925                 if (!data->ocd_ibits_known &&
4926                     data->ocd_connect_flags & OBD_CONNECT_IBITS)
4927                         data->ocd_connect_flags &= ~OBD_CONNECT_IBITS;
4928
4929                 if (!mdt->mdt_opts.mo_acl)
4930                         data->ocd_connect_flags &= ~OBD_CONNECT_ACL;
4931
4932                 if (!mdt->mdt_opts.mo_user_xattr)
4933                         data->ocd_connect_flags &= ~OBD_CONNECT_XATTR;
4934
4935                 if (!mdt->mdt_som_conf)
4936                         data->ocd_connect_flags &= ~OBD_CONNECT_SOM;
4937
4938                 if (data->ocd_connect_flags & OBD_CONNECT_BRW_SIZE) {
4939                         data->ocd_brw_size = min(data->ocd_brw_size,
4940                                (__u32)(PTLRPC_MAX_BRW_PAGES << CFS_PAGE_SHIFT));
4941                         if (data->ocd_brw_size == 0) {
4942                                 CERROR("%s: cli %s/%p ocd_connect_flags: "LPX64
4943                                        " ocd_version: %x ocd_grant: %d "
4944                                        "ocd_index: %u ocd_brw_size is "
4945                                        "unexpectedly zero, network data "
4946                                        "corruption? Refusing connection of this"
4947                                        " client\n",
4948                                        exp->exp_obd->obd_name,
4949                                        exp->exp_client_uuid.uuid,
4950                                        exp, data->ocd_connect_flags, data->ocd_version,
4951                                        data->ocd_grant, data->ocd_index);
4952                                 return -EPROTO;
4953                         }
4954                 }
4955
4956                 cfs_spin_lock(&exp->exp_lock);
4957                 exp->exp_connect_flags = data->ocd_connect_flags;
4958                 cfs_spin_unlock(&exp->exp_lock);
4959                 data->ocd_version = LUSTRE_VERSION_CODE;
4960                 exp->exp_mdt_data.med_ibits_known = data->ocd_ibits_known;
4961         }
4962
4963 #if 0
4964         if (mdt->mdt_opts.mo_acl &&
4965             ((exp->exp_connect_flags & OBD_CONNECT_ACL) == 0)) {
4966                 CWARN("%s: MDS requires ACL support but client does not\n",
4967                       mdt->mdt_md_dev.md_lu_dev.ld_obd->obd_name);
4968                 return -EBADE;
4969         }
4970 #endif
4971
4972         if ((exp->exp_connect_flags & OBD_CONNECT_FID) == 0) {
4973                 CWARN("%s: MDS requires FID support, but client not\n",
4974                       mdt->mdt_md_dev.md_lu_dev.ld_obd->obd_name);
4975                 return -EBADE;
4976         }
4977
4978         if (mdt->mdt_som_conf && !exp_connect_som(exp) &&
4979             !(exp->exp_connect_flags & OBD_CONNECT_MDS_MDS)) {
4980                 CWARN("%s: MDS has SOM enabled, but client does not support "
4981                       "it\n", mdt->mdt_md_dev.md_lu_dev.ld_obd->obd_name);
4982                 return -EBADE;
4983         }
4984
4985         return 0;
4986 }
4987
4988 static int mdt_connect_check_sptlrpc(struct mdt_device *mdt,
4989                                      struct obd_export *exp,
4990                                      struct ptlrpc_request *req)
4991 {
4992         struct sptlrpc_flavor   flvr;
4993         int                     rc = 0;
4994
4995         if (exp->exp_flvr.sf_rpc == SPTLRPC_FLVR_INVALID) {
4996                 cfs_read_lock(&mdt->mdt_sptlrpc_lock);
4997                 sptlrpc_target_choose_flavor(&mdt->mdt_sptlrpc_rset,
4998                                              req->rq_sp_from,
4999                                              req->rq_peer.nid,
5000                                              &flvr);
5001                 cfs_read_unlock(&mdt->mdt_sptlrpc_lock);
5002
5003                 cfs_spin_lock(&exp->exp_lock);
5004
5005                 exp->exp_sp_peer = req->rq_sp_from;
5006                 exp->exp_flvr = flvr;
5007
5008                 if (exp->exp_flvr.sf_rpc != SPTLRPC_FLVR_ANY &&
5009                     exp->exp_flvr.sf_rpc != req->rq_flvr.sf_rpc) {
5010                         CERROR("unauthorized rpc flavor %x from %s, "
5011                                "expect %x\n", req->rq_flvr.sf_rpc,
5012                                libcfs_nid2str(req->rq_peer.nid),
5013                                exp->exp_flvr.sf_rpc);
5014                         rc = -EACCES;
5015                 }
5016
5017                 cfs_spin_unlock(&exp->exp_lock);
5018         } else {
5019                 if (exp->exp_sp_peer != req->rq_sp_from) {
5020                         CERROR("RPC source %s doesn't match %s\n",
5021                                sptlrpc_part2name(req->rq_sp_from),
5022                                sptlrpc_part2name(exp->exp_sp_peer));
5023                         rc = -EACCES;
5024                 } else {
5025                         rc = sptlrpc_target_export_check(exp, req);
5026                 }
5027         }
5028
5029         return rc;
5030 }
5031
5032 /* mds_connect copy */
5033 static int mdt_obd_connect(const struct lu_env *env,
5034                            struct obd_export **exp, struct obd_device *obd,
5035                            struct obd_uuid *cluuid,
5036                            struct obd_connect_data *data,
5037                            void *localdata)
5038 {
5039         struct mdt_thread_info *info;
5040         struct obd_export      *lexp;
5041         struct lustre_handle    conn = { 0 };
5042         struct mdt_device      *mdt;
5043         struct ptlrpc_request  *req;
5044         int                     rc;
5045         ENTRY;
5046
5047         LASSERT(env != NULL);
5048         if (!exp || !obd || !cluuid)
5049                 RETURN(-EINVAL);
5050
5051         info = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
5052         req = info->mti_pill->rc_req;
5053         mdt = mdt_dev(obd->obd_lu_dev);
5054
5055         rc = class_connect(&conn, obd, cluuid);
5056         if (rc)
5057                 RETURN(rc);
5058
5059         lexp = class_conn2export(&conn);
5060         LASSERT(lexp != NULL);
5061
5062         rc = mdt_connect_check_sptlrpc(mdt, lexp, req);
5063         if (rc)
5064                 GOTO(out, rc);
5065
5066         if (OBD_FAIL_CHECK(OBD_FAIL_TGT_RCVG_FLAG))
5067                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_RECOVERING);
5068
5069         rc = mdt_connect_internal(lexp, mdt, data);
5070         if (rc == 0) {
5071                 struct lsd_client_data *lcd = lexp->exp_target_data.ted_lcd;
5072
5073                 LASSERT(lcd);
5074                 info->mti_exp = lexp;
5075                 memcpy(lcd->lcd_uuid, cluuid, sizeof lcd->lcd_uuid);
5076                 rc = lut_client_new(env, lexp);
5077                 if (rc == 0)
5078                         mdt_export_stats_init(obd, lexp, localdata);
5079         }
5080
5081 out:
5082         if (rc != 0) {
5083                 class_disconnect(lexp);
5084                 *exp = NULL;
5085         } else {
5086                 *exp = lexp;
5087         }
5088
5089         RETURN(rc);
5090 }
5091
5092 static int mdt_obd_reconnect(const struct lu_env *env,
5093                              struct obd_export *exp, struct obd_device *obd,
5094                              struct obd_uuid *cluuid,
5095                              struct obd_connect_data *data,
5096                              void *localdata)
5097 {
5098         struct mdt_thread_info *info;
5099         struct mdt_device      *mdt;
5100         struct ptlrpc_request  *req;
5101         int                     rc;
5102         ENTRY;
5103
5104         if (exp == NULL || obd == NULL || cluuid == NULL)
5105                 RETURN(-EINVAL);
5106
5107         info = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
5108         req = info->mti_pill->rc_req;
5109         mdt = mdt_dev(obd->obd_lu_dev);
5110
5111         rc = mdt_connect_check_sptlrpc(mdt, exp, req);
5112         if (rc)
5113                 RETURN(rc);
5114
5115         rc = mdt_connect_internal(exp, mdt_dev(obd->obd_lu_dev), data);
5116         if (rc == 0)
5117                 mdt_export_stats_init(obd, exp, localdata);
5118
5119         RETURN(rc);
5120 }
5121
5122 static int mdt_export_cleanup(struct obd_export *exp)
5123 {
5124         struct mdt_export_data *med = &exp->exp_mdt_data;
5125         struct obd_device      *obd = exp->exp_obd;
5126         struct mdt_device      *mdt;
5127         struct mdt_thread_info *info;
5128         struct lu_env           env;
5129         CFS_LIST_HEAD(closing_list);
5130         struct mdt_file_data *mfd, *n;
5131         int rc = 0;
5132         ENTRY;
5133
5134         cfs_spin_lock(&med->med_open_lock);
5135         while (!cfs_list_empty(&med->med_open_head)) {
5136                 cfs_list_t *tmp = med->med_open_head.next;
5137                 mfd = cfs_list_entry(tmp, struct mdt_file_data, mfd_list);
5138
5139                 /* Remove mfd handle so it can't be found again.
5140                  * We are consuming the mfd_list reference here. */
5141                 class_handle_unhash(&mfd->mfd_handle);
5142                 cfs_list_move_tail(&mfd->mfd_list, &closing_list);
5143         }
5144         cfs_spin_unlock(&med->med_open_lock);
5145         mdt = mdt_dev(obd->obd_lu_dev);
5146         LASSERT(mdt != NULL);
5147
5148         rc = lu_env_init(&env, LCT_MD_THREAD);
5149         if (rc)
5150                 RETURN(rc);
5151
5152         info = lu_context_key_get(&env.le_ctx, &mdt_thread_key);
5153         LASSERT(info != NULL);
5154         memset(info, 0, sizeof *info);
5155         info->mti_env = &env;
5156         info->mti_mdt = mdt;
5157         info->mti_exp = exp;
5158
5159         if (!cfs_list_empty(&closing_list)) {
5160                 struct md_attr *ma = &info->mti_attr;
5161                 int lmm_size;
5162                 int cookie_size;
5163
5164                 lmm_size = mdt->mdt_max_mdsize;
5165                 OBD_ALLOC_LARGE(ma->ma_lmm, lmm_size);
5166                 if (ma->ma_lmm == NULL)
5167                         GOTO(out_lmm, rc = -ENOMEM);
5168
5169                 cookie_size = mdt->mdt_max_cookiesize;
5170                 OBD_ALLOC_LARGE(ma->ma_cookie, cookie_size);
5171                 if (ma->ma_cookie == NULL)
5172                         GOTO(out_cookie, rc = -ENOMEM);
5173
5174                 /* Close any open files (which may also cause orphan unlinking). */
5175                 cfs_list_for_each_entry_safe(mfd, n, &closing_list, mfd_list) {
5176                         cfs_list_del_init(&mfd->mfd_list);
5177                         memset(&ma->ma_attr, 0, sizeof(ma->ma_attr));
5178                         ma->ma_lmm_size = lmm_size;
5179                         ma->ma_cookie_size = cookie_size;
5180                         ma->ma_need = 0;
5181                         /* It is not for setattr, just tell MDD to send
5182                          * DESTROY RPC to OSS if needed */
5183                         ma->ma_valid = MA_FLAGS;
5184                         ma->ma_attr_flags = MDS_CLOSE_CLEANUP;
5185                         /* Don't unlink orphan on failover umount, LU-184 */
5186                         if (exp->exp_flags & OBD_OPT_FAILOVER)
5187                                 ma->ma_attr_flags |= MDS_KEEP_ORPHAN;
5188                         mdt_mfd_close(info, mfd);
5189                 }
5190                 OBD_FREE_LARGE(ma->ma_cookie, cookie_size);
5191                 ma->ma_cookie = NULL;
5192 out_cookie:
5193                 OBD_FREE_LARGE(ma->ma_lmm, lmm_size);
5194                 ma->ma_lmm = NULL;
5195         }
5196 out_lmm:
5197         info->mti_mdt = NULL;
5198         /* cleanup client slot early */
5199         /* Do not erase record for recoverable client. */
5200         if (!(exp->exp_flags & OBD_OPT_FAILOVER) || exp->exp_failed)
5201                 lut_client_del(&env, exp);
5202         lu_env_fini(&env);
5203
5204         RETURN(rc);
5205 }
5206
5207 static int mdt_obd_disconnect(struct obd_export *exp)
5208 {
5209         int rc;
5210         ENTRY;
5211
5212         LASSERT(exp);
5213         class_export_get(exp);
5214
5215         rc = server_disconnect_export(exp);
5216         if (rc != 0)
5217                 CDEBUG(D_IOCTL, "server disconnect error: %d\n", rc);
5218
5219         rc = mdt_export_cleanup(exp);
5220         class_export_put(exp);
5221         RETURN(rc);
5222 }
5223
5224 /* FIXME: Can we avoid using these two interfaces? */
5225 static int mdt_init_export(struct obd_export *exp)
5226 {
5227         struct mdt_export_data *med = &exp->exp_mdt_data;
5228         int                     rc;
5229         ENTRY;
5230
5231         CFS_INIT_LIST_HEAD(&med->med_open_head);
5232         cfs_spin_lock_init(&med->med_open_lock);
5233         cfs_mutex_init(&med->med_idmap_mutex);
5234         med->med_idmap = NULL;
5235         cfs_spin_lock(&exp->exp_lock);
5236         exp->exp_connecting = 1;
5237         cfs_spin_unlock(&exp->exp_lock);
5238
5239         /* self-export doesn't need client data and ldlm initialization */
5240         if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid,
5241                                      &exp->exp_client_uuid)))
5242                 RETURN(0);
5243
5244         rc = lut_client_alloc(exp);
5245         if (rc == 0)
5246                 rc = ldlm_init_export(exp);
5247         if (rc)
5248                 CERROR("%s: Error %d while initializing export\n",
5249                        exp->exp_obd->obd_name, rc);
5250         RETURN(rc);
5251 }
5252
5253 static int mdt_destroy_export(struct obd_export *exp)
5254 {
5255         ENTRY;
5256
5257         if (exp_connect_rmtclient(exp))
5258                 mdt_cleanup_idmap(&exp->exp_mdt_data);
5259
5260         target_destroy_export(exp);
5261         /* destroy can be called from failed obd_setup, so
5262          * checking uuid is safer than obd_self_export */
5263         if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid,
5264                                      &exp->exp_client_uuid)))
5265                 RETURN(0);
5266
5267         ldlm_destroy_export(exp);
5268         lut_client_free(exp);
5269
5270         LASSERT(cfs_list_empty(&exp->exp_outstanding_replies));
5271         LASSERT(cfs_list_empty(&exp->exp_mdt_data.med_open_head));
5272
5273         RETURN(0);
5274 }
5275
5276 static void mdt_allow_cli(struct mdt_device *m, unsigned int flag)
5277 {
5278         if (flag & CONFIG_LOG)
5279                 cfs_set_bit(MDT_FL_CFGLOG, &m->mdt_state);
5280
5281         /* also notify active event */
5282         if (flag & CONFIG_SYNC)
5283                 cfs_set_bit(MDT_FL_SYNCED, &m->mdt_state);
5284
5285         if (cfs_test_bit(MDT_FL_CFGLOG, &m->mdt_state) &&
5286             cfs_test_bit(MDT_FL_SYNCED, &m->mdt_state)) {
5287                 struct obd_device *obd = m->mdt_md_dev.md_lu_dev.ld_obd;
5288
5289                 /* Open for clients */
5290                 if (obd->obd_no_conn) {
5291                         cfs_spin_lock(&obd->obd_dev_lock);
5292                         obd->obd_no_conn = 0;
5293                         cfs_spin_unlock(&obd->obd_dev_lock);
5294                 }
5295         }
5296 }
5297
5298 static int mdt_upcall(const struct lu_env *env, struct md_device *md,
5299                       enum md_upcall_event ev, void *data)
5300 {
5301         struct mdt_device *m = mdt_dev(&md->md_lu_dev);
5302         struct md_device  *next  = m->mdt_child;
5303         struct mdt_thread_info *mti;
5304         int rc = 0;
5305         ENTRY;
5306
5307         switch (ev) {
5308                 case MD_LOV_SYNC:
5309                         rc = next->md_ops->mdo_maxsize_get(env, next,
5310                                         &m->mdt_max_mdsize,
5311                                         &m->mdt_max_cookiesize);
5312                         CDEBUG(D_INFO, "get max mdsize %d max cookiesize %d\n",
5313                                      m->mdt_max_mdsize, m->mdt_max_cookiesize);
5314                         mdt_allow_cli(m, CONFIG_SYNC);
5315                         if (data)
5316                                 (*(__u64 *)data) =
5317                                       m->mdt_lut.lut_obd->u.obt.obt_mount_count;
5318                         break;
5319                 case MD_NO_TRANS:
5320                         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
5321                         mti->mti_no_need_trans = 1;
5322                         CDEBUG(D_INFO, "disable mdt trans for this thread\n");
5323                         break;
5324                 case MD_LOV_CONFIG:
5325                         /* Check that MDT is not yet configured */
5326                         LASSERT(!cfs_test_bit(MDT_FL_CFGLOG, &m->mdt_state));
5327                         break;
5328 #ifdef HAVE_QUOTA_SUPPORT
5329                 case MD_LOV_QUOTA:
5330                         if (md->md_lu_dev.ld_obd->obd_recovering == 0 &&
5331                             likely(md->md_lu_dev.ld_obd->obd_stopping == 0))
5332                                 next->md_ops->mdo_quota.mqo_recovery(env, next);
5333                         break;
5334 #endif
5335                 default:
5336                         CERROR("invalid event\n");
5337                         rc = -EINVAL;
5338                         break;
5339         }
5340         RETURN(rc);
5341 }
5342
5343 static int mdt_obd_notify(struct obd_device *obd,
5344                           struct obd_device *watched,
5345                           enum obd_notify_event ev, void *data)
5346 {
5347         struct mdt_device *mdt = mdt_dev(obd->obd_lu_dev);
5348 #ifdef HAVE_QUOTA_SUPPORT
5349         struct md_device *next = mdt->mdt_child;
5350 #endif
5351         ENTRY;
5352
5353         switch (ev) {
5354         case OBD_NOTIFY_CONFIG:
5355                 mdt_allow_cli(mdt, (unsigned long)data);
5356
5357 #ifdef HAVE_QUOTA_SUPPORT
5358                /* quota_type has been processed, we can now handle
5359                 * incoming quota requests */
5360                 next->md_ops->mdo_quota.mqo_notify(NULL, next);
5361 #endif
5362                 break;
5363         default:
5364                 CDEBUG(D_INFO, "Unhandled notification %#x\n", ev);
5365         }
5366         RETURN(0);
5367 }
5368
5369 static int mdt_rpc_fid2path(struct mdt_thread_info *info, void *key,
5370                             void *val, int vallen)
5371 {
5372         struct mdt_device *mdt = mdt_dev(info->mti_exp->exp_obd->obd_lu_dev);
5373         struct getinfo_fid2path *fpout, *fpin;
5374         int rc = 0;
5375
5376         fpin = key + cfs_size_round(sizeof(KEY_FID2PATH));
5377         fpout = val;
5378
5379         if (ptlrpc_req_need_swab(info->mti_pill->rc_req))
5380                 lustre_swab_fid2path(fpin);
5381
5382         memcpy(fpout, fpin, sizeof(*fpin));
5383         if (fpout->gf_pathlen != vallen - sizeof(*fpin))
5384                 RETURN(-EINVAL);
5385
5386         rc = mdt_fid2path(info->mti_env, mdt, fpout);
5387         RETURN(rc);
5388 }
5389
5390 static int mdt_fid2path(const struct lu_env *env, struct mdt_device *mdt,
5391                         struct getinfo_fid2path *fp)
5392 {
5393         struct mdt_object *obj;
5394         int    rc;
5395         ENTRY;
5396
5397         CDEBUG(D_IOCTL, "path get "DFID" from "LPU64" #%d\n",
5398                PFID(&fp->gf_fid), fp->gf_recno, fp->gf_linkno);
5399
5400         if (!fid_is_sane(&fp->gf_fid))
5401                 RETURN(-EINVAL);
5402
5403         obj = mdt_object_find(env, mdt, &fp->gf_fid);
5404         if (obj == NULL || IS_ERR(obj)) {
5405                 CDEBUG(D_IOCTL, "no object "DFID": %ld\n", PFID(&fp->gf_fid),
5406                        PTR_ERR(obj));
5407                 RETURN(-EINVAL);
5408         }
5409
5410         rc = lu_object_exists(&obj->mot_obj.mo_lu);
5411         if (rc <= 0) {
5412                 if (rc == -1)
5413                         rc = -EREMOTE;
5414                 else
5415                         rc = -ENOENT;
5416                 mdt_object_put(env, obj);
5417                 CDEBUG(D_IOCTL, "nonlocal object "DFID": %d\n",
5418                        PFID(&fp->gf_fid), rc);
5419                 RETURN(rc);
5420         }
5421
5422         rc = mo_path(env, md_object_next(&obj->mot_obj), fp->gf_path,
5423                      fp->gf_pathlen, &fp->gf_recno, &fp->gf_linkno);
5424         mdt_object_put(env, obj);
5425
5426         RETURN(rc);
5427 }
5428
5429 static int mdt_get_info(struct mdt_thread_info *info)
5430 {
5431         struct ptlrpc_request *req = mdt_info_req(info);
5432         char *key;
5433         int keylen;
5434         __u32 *vallen;
5435         void *valout;
5436         int rc;
5437         ENTRY;
5438
5439         key = req_capsule_client_get(info->mti_pill, &RMF_GETINFO_KEY);
5440         if (key == NULL) {
5441                 CDEBUG(D_IOCTL, "No GETINFO key");
5442                 RETURN(-EFAULT);
5443         }
5444         keylen = req_capsule_get_size(info->mti_pill, &RMF_GETINFO_KEY,
5445                                       RCL_CLIENT);
5446
5447         vallen = req_capsule_client_get(info->mti_pill, &RMF_GETINFO_VALLEN);
5448         if (vallen == NULL) {
5449                 CDEBUG(D_IOCTL, "Unable to get RMF_GETINFO_VALLEN buffer");
5450                 RETURN(-EFAULT);
5451         }
5452
5453         req_capsule_set_size(info->mti_pill, &RMF_GETINFO_VAL, RCL_SERVER,
5454                              *vallen);
5455         rc = req_capsule_server_pack(info->mti_pill);
5456         valout = req_capsule_server_get(info->mti_pill, &RMF_GETINFO_VAL);
5457         if (valout == NULL) {
5458                 CDEBUG(D_IOCTL, "Unable to get get-info RPC out buffer");
5459                 RETURN(-EFAULT);
5460         }
5461
5462         if (KEY_IS(KEY_FID2PATH))
5463                 rc = mdt_rpc_fid2path(info, key, valout, *vallen);
5464         else
5465                 rc = -EINVAL;
5466
5467         lustre_msg_set_status(req->rq_repmsg, rc);
5468
5469         RETURN(rc);
5470 }
5471
5472 /* Pass the ioc down */
5473 static int mdt_ioc_child(struct lu_env *env, struct mdt_device *mdt,
5474                          unsigned int cmd, int len, void *data)
5475 {
5476         struct lu_context ioctl_session;
5477         struct md_device *next = mdt->mdt_child;
5478         int rc;
5479         ENTRY;
5480
5481         rc = lu_context_init(&ioctl_session, LCT_SESSION);
5482         if (rc)
5483                 RETURN(rc);
5484         ioctl_session.lc_thread = (struct ptlrpc_thread *)cfs_current();
5485         lu_context_enter(&ioctl_session);
5486         env->le_ses = &ioctl_session;
5487
5488         LASSERT(next->md_ops->mdo_iocontrol);
5489         rc = next->md_ops->mdo_iocontrol(env, next, cmd, len, data);
5490
5491         lu_context_exit(&ioctl_session);
5492         lu_context_fini(&ioctl_session);
5493         RETURN(rc);
5494 }
5495
5496 static int mdt_ioc_version_get(struct mdt_thread_info *mti, void *karg)
5497 {
5498         struct obd_ioctl_data *data = karg;
5499         struct lu_fid *fid = (struct lu_fid *)data->ioc_inlbuf1;
5500         __u64 version;
5501         struct mdt_object *obj;
5502         struct mdt_lock_handle  *lh;
5503         int rc;
5504         ENTRY;
5505
5506         CDEBUG(D_IOCTL, "getting version for "DFID"\n", PFID(fid));
5507         if (!fid_is_sane(fid))
5508                 RETURN(-EINVAL);
5509
5510         lh = &mti->mti_lh[MDT_LH_PARENT];
5511         mdt_lock_reg_init(lh, LCK_CR);
5512
5513         obj = mdt_object_find_lock(mti, fid, lh, MDS_INODELOCK_UPDATE);
5514         if (IS_ERR(obj))
5515                 RETURN(PTR_ERR(obj));
5516
5517         rc = mdt_object_exists(obj);
5518         if (rc < 0) {
5519                 rc = -EREMOTE;
5520                 /**
5521                  * before calling version get the correct MDS should be
5522                  * fid, this is error to find remote object here
5523                  */
5524                 CERROR("nonlocal object "DFID"\n", PFID(fid));
5525         } else if (rc == 0) {
5526                 *(__u64 *)data->ioc_inlbuf2 = ENOENT_VERSION;
5527                 rc = -ENOENT;
5528         } else {
5529                 version = dt_version_get(mti->mti_env, mdt_obj2dt(obj));
5530                *(__u64 *)data->ioc_inlbuf2 = version;
5531                 rc = 0;
5532         }
5533         mdt_object_unlock_put(mti, obj, lh, 1);
5534         RETURN(rc);
5535 }
5536
5537 /* ioctls on obd dev */
5538 static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
5539                          void *karg, void *uarg)
5540 {
5541         struct lu_env      env;
5542         struct obd_device *obd = exp->exp_obd;
5543         struct mdt_device *mdt = mdt_dev(obd->obd_lu_dev);
5544         struct dt_device  *dt = mdt->mdt_bottom;
5545         int rc;
5546
5547         ENTRY;
5548         CDEBUG(D_IOCTL, "handling ioctl cmd %#x\n", cmd);
5549         rc = lu_env_init(&env, LCT_MD_THREAD);
5550         if (rc)
5551                 RETURN(rc);
5552
5553         switch (cmd) {
5554         case OBD_IOC_SYNC:
5555                 rc = mdt_device_sync(&env, mdt);
5556                 break;
5557         case OBD_IOC_SET_READONLY:
5558                 rc = dt->dd_ops->dt_ro(&env, dt);
5559                 break;
5560         case OBD_IOC_ABORT_RECOVERY:
5561                 CERROR("Aborting recovery for device %s\n", obd->obd_name);
5562                 target_stop_recovery_thread(obd);
5563                 rc = 0;
5564                 break;
5565         case OBD_IOC_CHANGELOG_REG:
5566         case OBD_IOC_CHANGELOG_DEREG:
5567         case OBD_IOC_CHANGELOG_CLEAR:
5568                 rc = mdt_ioc_child(&env, mdt, cmd, len, karg);
5569                 break;
5570         case OBD_IOC_GET_OBJ_VERSION: {
5571                 struct mdt_thread_info *mti;
5572                 mti = lu_context_key_get(&env.le_ctx, &mdt_thread_key);
5573                 memset(mti, 0, sizeof *mti);
5574                 mti->mti_env = &env;
5575                 mti->mti_mdt = mdt;
5576                 mti->mti_exp = exp;
5577
5578                 rc = mdt_ioc_version_get(mti, karg);
5579                 break;
5580         }
5581         default:
5582                 CERROR("Not supported cmd = %d for device %s\n",
5583                        cmd, obd->obd_name);
5584                 rc = -EOPNOTSUPP;
5585         }
5586
5587         lu_env_fini(&env);
5588         RETURN(rc);
5589 }
5590
5591 int mdt_postrecov(const struct lu_env *env, struct mdt_device *mdt)
5592 {
5593         struct lu_device *ld = md2lu_dev(mdt->mdt_child);
5594 #ifdef HAVE_QUOTA_SUPPORT
5595         struct obd_device *obd = mdt2obd_dev(mdt);
5596         struct md_device *next = mdt->mdt_child;
5597 #endif
5598         int rc;
5599         ENTRY;
5600
5601         rc = ld->ld_ops->ldo_recovery_complete(env, ld);
5602 #ifdef HAVE_QUOTA_SUPPORT
5603         if (likely(obd->obd_stopping == 0))
5604                 next->md_ops->mdo_quota.mqo_recovery(env, next);
5605 #endif
5606         RETURN(rc);
5607 }
5608
5609 int mdt_obd_postrecov(struct obd_device *obd)
5610 {
5611         struct lu_env env;
5612         int rc;
5613
5614         rc = lu_env_init(&env, LCT_MD_THREAD);
5615         if (rc)
5616                 RETURN(rc);
5617         rc = mdt_postrecov(&env, mdt_dev(obd->obd_lu_dev));
5618         lu_env_fini(&env);
5619         return rc;
5620 }
5621
5622 /**
5623  * Send a copytool req to a client
5624  * Note this sends a request RPC from a server (MDT) to a client (MDC),
5625  * backwards of normal comms.
5626  */
5627 int mdt_hsm_copytool_send(struct obd_export *exp)
5628 {
5629         struct kuc_hdr *lh;
5630         struct hsm_action_list *hal;
5631         struct hsm_action_item *hai;
5632         int rc, len;
5633         ENTRY;
5634
5635         CWARN("%s: writing to mdc at %s\n", exp->exp_obd->obd_name,
5636               libcfs_nid2str(exp->exp_connection->c_peer.nid));
5637
5638         len = sizeof(*lh) + sizeof(*hal) + MTI_NAME_MAXLEN +
5639                 /* for mockup below */ 2 * cfs_size_round(sizeof(*hai));
5640         OBD_ALLOC(lh, len);
5641         if (lh == NULL)
5642                 RETURN(-ENOMEM);
5643
5644         lh->kuc_magic = KUC_MAGIC;
5645         lh->kuc_transport = KUC_TRANSPORT_HSM;
5646         lh->kuc_msgtype = HMT_ACTION_LIST;
5647         lh->kuc_msglen = len;
5648
5649         hal = (struct hsm_action_list *)(lh + 1);
5650         hal->hal_version = HAL_VERSION;
5651         hal->hal_archive_num = 1;
5652         obd_uuid2fsname(hal->hal_fsname, exp->exp_obd->obd_name,
5653                         MTI_NAME_MAXLEN);
5654
5655         /* mock up an action list */
5656         hal->hal_count = 2;
5657         hai = hai_zero(hal);
5658         hai->hai_action = HSMA_ARCHIVE;
5659         hai->hai_fid.f_oid = 0xA00A;
5660         hai->hai_len = sizeof(*hai);
5661         hai = hai_next(hai);
5662         hai->hai_action = HSMA_RESTORE;
5663         hai->hai_fid.f_oid = 0xB00B;
5664         hai->hai_len = sizeof(*hai);
5665
5666         /* Uses the ldlm reverse import; this rpc will be seen by
5667           the ldlm_callback_handler */
5668         rc = do_set_info_async(exp->exp_imp_reverse,
5669                                LDLM_SET_INFO, LUSTRE_OBD_VERSION,
5670                                sizeof(KEY_HSM_COPYTOOL_SEND),
5671                                KEY_HSM_COPYTOOL_SEND,
5672                                len, lh, NULL);
5673
5674         OBD_FREE(lh, len);
5675
5676         RETURN(rc);
5677 }
5678
5679 static struct obd_ops mdt_obd_device_ops = {
5680         .o_owner          = THIS_MODULE,
5681         .o_set_info_async = mdt_obd_set_info_async,
5682         .o_connect        = mdt_obd_connect,
5683         .o_reconnect      = mdt_obd_reconnect,
5684         .o_disconnect     = mdt_obd_disconnect,
5685         .o_init_export    = mdt_init_export,
5686         .o_destroy_export = mdt_destroy_export,
5687         .o_iocontrol      = mdt_iocontrol,
5688         .o_postrecov      = mdt_obd_postrecov,
5689         .o_notify         = mdt_obd_notify
5690 };
5691
5692 static struct lu_device* mdt_device_fini(const struct lu_env *env,
5693                                          struct lu_device *d)
5694 {
5695         struct mdt_device *m = mdt_dev(d);
5696         ENTRY;
5697
5698         mdt_fini(env, m);
5699         RETURN(NULL);
5700 }
5701
5702 static struct lu_device *mdt_device_free(const struct lu_env *env,
5703                                          struct lu_device *d)
5704 {
5705         struct mdt_device *m = mdt_dev(d);
5706         ENTRY;
5707
5708         md_device_fini(&m->mdt_md_dev);
5709         OBD_FREE_PTR(m);
5710         RETURN(NULL);
5711 }
5712
5713 static struct lu_device *mdt_device_alloc(const struct lu_env *env,
5714                                           struct lu_device_type *t,
5715                                           struct lustre_cfg *cfg)
5716 {
5717         struct lu_device  *l;
5718         struct mdt_device *m;
5719
5720         OBD_ALLOC_PTR(m);
5721         if (m != NULL) {
5722                 int rc;
5723
5724                 l = &m->mdt_md_dev.md_lu_dev;
5725                 rc = mdt_init0(env, m, t, cfg);
5726                 if (rc != 0) {
5727                         mdt_device_free(env, l);
5728                         l = ERR_PTR(rc);
5729                         return l;
5730                 }
5731                 md_upcall_init(&m->mdt_md_dev, mdt_upcall);
5732         } else
5733                 l = ERR_PTR(-ENOMEM);
5734         return l;
5735 }
5736
5737 /* context key constructor/destructor: mdt_key_init, mdt_key_fini */
5738 LU_KEY_INIT_FINI(mdt, struct mdt_thread_info);
5739
5740 /* context key: mdt_thread_key */
5741 LU_CONTEXT_KEY_DEFINE(mdt, LCT_MD_THREAD);
5742
5743 struct md_ucred *mdt_ucred(const struct mdt_thread_info *info)
5744 {
5745         return md_ucred(info->mti_env);
5746 }
5747
5748 /**
5749  * Enable/disable COS (Commit On Sharing).
5750  *
5751  * Set/Clear the COS flag in mdt options.
5752  *
5753  * \param mdt mdt device
5754  * \param val 0 disables COS, other values enable COS
5755  */
5756 void mdt_enable_cos(struct mdt_device *mdt, int val)
5757 {
5758         struct lu_env env;
5759         int rc;
5760
5761         mdt->mdt_opts.mo_cos = !!val;
5762         rc = lu_env_init(&env, LCT_LOCAL);
5763         if (unlikely(rc != 0)) {
5764                 CWARN("lu_env initialization failed with rc = %d,"
5765                       "cannot sync\n", rc);
5766                 return;
5767         }
5768         mdt_device_sync(&env, mdt);
5769         lu_env_fini(&env);
5770 }
5771
5772 /**
5773  * Check COS (Commit On Sharing) status.
5774  *
5775  * Return COS flag status.
5776  *
5777  * \param mdt mdt device
5778  */
5779 int mdt_cos_is_enabled(struct mdt_device *mdt)
5780 {
5781         return mdt->mdt_opts.mo_cos != 0;
5782 }
5783
5784 /* type constructor/destructor: mdt_type_init, mdt_type_fini */
5785 LU_TYPE_INIT_FINI(mdt, &mdt_thread_key);
5786
5787 static struct lu_device_type_operations mdt_device_type_ops = {
5788         .ldto_init = mdt_type_init,
5789         .ldto_fini = mdt_type_fini,
5790
5791         .ldto_start = mdt_type_start,
5792         .ldto_stop  = mdt_type_stop,
5793
5794         .ldto_device_alloc = mdt_device_alloc,
5795         .ldto_device_free  = mdt_device_free,
5796         .ldto_device_fini  = mdt_device_fini
5797 };
5798
5799 static struct lu_device_type mdt_device_type = {
5800         .ldt_tags     = LU_DEVICE_MD,
5801         .ldt_name     = LUSTRE_MDT_NAME,
5802         .ldt_ops      = &mdt_device_type_ops,
5803         .ldt_ctx_tags = LCT_MD_THREAD
5804 };
5805
5806 static int __init mdt_mod_init(void)
5807 {
5808         struct lprocfs_static_vars lvars;
5809         int rc;
5810
5811         if (mdt_num_threads > 0) {
5812                 if (mdt_num_threads > MDT_MAX_THREADS)
5813                         mdt_num_threads = MDT_MAX_THREADS;
5814                 if (mdt_num_threads < MDT_MIN_THREADS)
5815                         mdt_num_threads = MDT_MIN_THREADS;
5816                 mdt_max_threads = mdt_min_threads = mdt_num_threads;
5817         } else {
5818                 mdt_max_threads = MDT_MAX_THREADS;
5819                 mdt_min_threads = MDT_MIN_THREADS;
5820         }
5821
5822         lprocfs_mdt_init_vars(&lvars);
5823         rc = class_register_type(&mdt_obd_device_ops, NULL,
5824                                  lvars.module_vars, LUSTRE_MDT_NAME,
5825                                  &mdt_device_type);
5826
5827         return rc;
5828 }
5829
5830 static void __exit mdt_mod_exit(void)
5831 {
5832         class_unregister_type(LUSTRE_MDT_NAME);
5833 }
5834
5835
5836 #define DEF_HNDL(prefix, base, suffix, flags, opc, fn, fmt)             \
5837 [prefix ## _ ## opc - prefix ## _ ## base] = {                          \
5838         .mh_name    = #opc,                                             \
5839         .mh_fail_id = OBD_FAIL_ ## prefix ## _  ## opc ## suffix,       \
5840         .mh_opc     = prefix ## _  ## opc,                              \
5841         .mh_flags   = flags,                                            \
5842         .mh_act     = fn,                                               \
5843         .mh_fmt     = fmt                                               \
5844 }
5845
5846 #define DEF_MDT_HNDL(flags, name, fn, fmt)                                  \
5847         DEF_HNDL(MDS, GETATTR, _NET, flags, name, fn, fmt)
5848
5849 #define DEF_SEQ_HNDL(flags, name, fn, fmt)                      \
5850         DEF_HNDL(SEQ, QUERY, _NET, flags, name, fn, fmt)
5851
5852 #define DEF_FLD_HNDL(flags, name, fn, fmt)                      \
5853         DEF_HNDL(FLD, QUERY, _NET, flags, name, fn, fmt)
5854 /*
5855  * Request with a format known in advance
5856  */
5857 #define DEF_MDT_HNDL_F(flags, name, fn)                                 \
5858         DEF_HNDL(MDS, GETATTR, _NET, flags, name, fn, &RQF_MDS_ ## name)
5859
5860 #define DEF_SEQ_HNDL_F(flags, name, fn)                                 \
5861         DEF_HNDL(SEQ, QUERY, _NET, flags, name, fn, &RQF_SEQ_ ## name)
5862
5863 #define DEF_FLD_HNDL_F(flags, name, fn)                                 \
5864         DEF_HNDL(FLD, QUERY, _NET, flags, name, fn, &RQF_FLD_ ## name)
5865 /*
5866  * Request with a format we do not yet know
5867  */
5868 #define DEF_MDT_HNDL_0(flags, name, fn)                                 \
5869         DEF_HNDL(MDS, GETATTR, _NET, flags, name, fn, NULL)
5870
5871 static struct mdt_handler mdt_mds_ops[] = {
5872 DEF_MDT_HNDL_F(0,                         CONNECT,      mdt_connect),
5873 DEF_MDT_HNDL_F(0,                         DISCONNECT,   mdt_disconnect),
5874 DEF_MDT_HNDL  (0,                         SET_INFO,     mdt_set_info,
5875                                                              &RQF_OBD_SET_INFO),
5876 DEF_MDT_HNDL_F(0,                         GET_INFO,     mdt_get_info),
5877 DEF_MDT_HNDL_F(0           |HABEO_REFERO, GETSTATUS,    mdt_getstatus),
5878 DEF_MDT_HNDL_F(HABEO_CORPUS,              GETATTR,      mdt_getattr),
5879 DEF_MDT_HNDL_F(HABEO_CORPUS|HABEO_REFERO, GETATTR_NAME, mdt_getattr_name),
5880 DEF_MDT_HNDL_F(HABEO_CORPUS,              GETXATTR,     mdt_getxattr),
5881 DEF_MDT_HNDL_F(0           |HABEO_REFERO, STATFS,       mdt_statfs),
5882 DEF_MDT_HNDL_F(0           |MUTABOR,      REINT,        mdt_reint),
5883 DEF_MDT_HNDL_F(HABEO_CORPUS,              CLOSE,        mdt_close),
5884 DEF_MDT_HNDL_F(HABEO_CORPUS,              DONE_WRITING, mdt_done_writing),
5885 DEF_MDT_HNDL_F(0           |HABEO_REFERO, PIN,          mdt_pin),
5886 DEF_MDT_HNDL_0(0,                         SYNC,         mdt_sync),
5887 DEF_MDT_HNDL_F(HABEO_CORPUS|HABEO_REFERO, IS_SUBDIR,    mdt_is_subdir),
5888 #ifdef HAVE_QUOTA_SUPPORT
5889 DEF_MDT_HNDL_F(0,                         QUOTACHECK,   mdt_quotacheck_handle),
5890 DEF_MDT_HNDL_F(0,                         QUOTACTL,     mdt_quotactl_handle)
5891 #endif
5892 };
5893
5894 #define DEF_OBD_HNDL(flags, name, fn)                   \
5895         DEF_HNDL(OBD, PING, _NET, flags, name, fn, NULL)
5896
5897
5898 static struct mdt_handler mdt_obd_ops[] = {
5899         DEF_OBD_HNDL(0, PING,           mdt_obd_ping),
5900         DEF_OBD_HNDL(0, LOG_CANCEL,     mdt_obd_log_cancel),
5901         DEF_OBD_HNDL(0, QC_CALLBACK,    mdt_obd_qc_callback)
5902 };
5903
5904 #define DEF_DLM_HNDL_0(flags, name, fn)                   \
5905         DEF_HNDL(LDLM, ENQUEUE, , flags, name, fn, NULL)
5906 #define DEF_DLM_HNDL_F(flags, name, fn)                   \
5907         DEF_HNDL(LDLM, ENQUEUE, , flags, name, fn, &RQF_LDLM_ ## name)
5908
5909 static struct mdt_handler mdt_dlm_ops[] = {
5910         DEF_DLM_HNDL_F(HABEO_CLAVIS, ENQUEUE,        mdt_enqueue),
5911         DEF_DLM_HNDL_0(HABEO_CLAVIS, CONVERT,        mdt_convert),
5912         DEF_DLM_HNDL_0(0,            BL_CALLBACK,    mdt_bl_callback),
5913         DEF_DLM_HNDL_0(0,            CP_CALLBACK,    mdt_cp_callback)
5914 };
5915
5916 #define DEF_LLOG_HNDL(flags, name, fn)                   \
5917         DEF_HNDL(LLOG, ORIGIN_HANDLE_CREATE, _NET, flags, name, fn, NULL)
5918
5919 static struct mdt_handler mdt_llog_ops[] = {
5920         DEF_LLOG_HNDL(0, ORIGIN_HANDLE_CREATE,      mdt_llog_create),
5921         DEF_LLOG_HNDL(0, ORIGIN_HANDLE_NEXT_BLOCK,  mdt_llog_next_block),
5922         DEF_LLOG_HNDL(0, ORIGIN_HANDLE_READ_HEADER, mdt_llog_read_header),
5923         DEF_LLOG_HNDL(0, ORIGIN_HANDLE_WRITE_REC,   NULL),
5924         DEF_LLOG_HNDL(0, ORIGIN_HANDLE_CLOSE,       NULL),
5925         DEF_LLOG_HNDL(0, ORIGIN_CONNECT,            NULL),
5926         DEF_LLOG_HNDL(0, CATINFO,                   NULL),
5927         DEF_LLOG_HNDL(0, ORIGIN_HANDLE_PREV_BLOCK,  mdt_llog_prev_block),
5928         DEF_LLOG_HNDL(0, ORIGIN_HANDLE_DESTROY,     mdt_llog_destroy),
5929 };
5930
5931 #define DEF_SEC_CTX_HNDL(name, fn)                      \
5932         DEF_HNDL(SEC_CTX, INIT, _NET, 0, name, fn, NULL)
5933
5934 static struct mdt_handler mdt_sec_ctx_ops[] = {
5935         DEF_SEC_CTX_HNDL(INIT,          mdt_sec_ctx_handle),
5936         DEF_SEC_CTX_HNDL(INIT_CONT,     mdt_sec_ctx_handle),
5937         DEF_SEC_CTX_HNDL(FINI,          mdt_sec_ctx_handle)
5938 };
5939
5940 static struct mdt_opc_slice mdt_regular_handlers[] = {
5941         {
5942                 .mos_opc_start = MDS_GETATTR,
5943                 .mos_opc_end   = MDS_LAST_OPC,
5944                 .mos_hs        = mdt_mds_ops
5945         },
5946         {
5947                 .mos_opc_start = OBD_PING,
5948                 .mos_opc_end   = OBD_LAST_OPC,
5949                 .mos_hs        = mdt_obd_ops
5950         },
5951         {
5952                 .mos_opc_start = LDLM_ENQUEUE,
5953                 .mos_opc_end   = LDLM_LAST_OPC,
5954                 .mos_hs        = mdt_dlm_ops
5955         },
5956         {
5957                 .mos_opc_start = LLOG_ORIGIN_HANDLE_CREATE,
5958                 .mos_opc_end   = LLOG_LAST_OPC,
5959                 .mos_hs        = mdt_llog_ops
5960         },
5961         {
5962                 .mos_opc_start = SEC_CTX_INIT,
5963                 .mos_opc_end   = SEC_LAST_OPC,
5964                 .mos_hs        = mdt_sec_ctx_ops
5965         },
5966         {
5967                 .mos_hs        = NULL
5968         }
5969 };
5970
5971 static struct mdt_handler mdt_readpage_ops[] = {
5972         DEF_MDT_HNDL_F(0,                         CONNECT,  mdt_connect),
5973         DEF_MDT_HNDL_F(HABEO_CORPUS|HABEO_REFERO, READPAGE, mdt_readpage),
5974 #ifdef HAVE_SPLIT_SUPPORT
5975         DEF_MDT_HNDL_F(HABEO_CORPUS|HABEO_REFERO, WRITEPAGE, mdt_writepage),
5976 #endif
5977
5978         /*
5979          * XXX: this is ugly and should be fixed one day, see mdc_close() for
5980          * detailed comments. --umka
5981          */
5982         DEF_MDT_HNDL_F(HABEO_CORPUS,              CLOSE,    mdt_close),
5983         DEF_MDT_HNDL_F(HABEO_CORPUS,              DONE_WRITING,    mdt_done_writing),
5984 };
5985
5986 static struct mdt_opc_slice mdt_readpage_handlers[] = {
5987         {
5988                 .mos_opc_start = MDS_GETATTR,
5989                 .mos_opc_end   = MDS_LAST_OPC,
5990                 .mos_hs        = mdt_readpage_ops
5991         },
5992         {
5993                 .mos_hs        = NULL
5994         }
5995 };
5996
5997 static struct mdt_handler mdt_xmds_ops[] = {
5998         DEF_MDT_HNDL_F(0,                         CONNECT,      mdt_connect),
5999         DEF_MDT_HNDL_F(HABEO_CORPUS             , GETATTR,      mdt_getattr),
6000         DEF_MDT_HNDL_F(0 | MUTABOR              , REINT,        mdt_reint),
6001         DEF_MDT_HNDL_F(HABEO_CORPUS|HABEO_REFERO, IS_SUBDIR,    mdt_is_subdir),
6002 };
6003
6004 static struct mdt_opc_slice mdt_xmds_handlers[] = {
6005         {
6006                 .mos_opc_start = MDS_GETATTR,
6007                 .mos_opc_end   = MDS_LAST_OPC,
6008                 .mos_hs        = mdt_xmds_ops
6009         },
6010         {
6011                 .mos_opc_start = OBD_PING,
6012                 .mos_opc_end   = OBD_LAST_OPC,
6013                 .mos_hs        = mdt_obd_ops
6014         },
6015         {
6016                 .mos_opc_start = SEC_CTX_INIT,
6017                 .mos_opc_end   = SEC_LAST_OPC,
6018                 .mos_hs        = mdt_sec_ctx_ops
6019         },
6020         {
6021                 .mos_hs        = NULL
6022         }
6023 };
6024
6025 static struct mdt_handler mdt_seq_ops[] = {
6026         DEF_SEQ_HNDL_F(0, QUERY, (int (*)(struct mdt_thread_info *))seq_query)
6027 };
6028
6029 static struct mdt_opc_slice mdt_seq_handlers[] = {
6030         {
6031                 .mos_opc_start = SEQ_QUERY,
6032                 .mos_opc_end   = SEQ_LAST_OPC,
6033                 .mos_hs        = mdt_seq_ops
6034         },
6035         {
6036                 .mos_hs        = NULL
6037         }
6038 };
6039
6040 static struct mdt_handler mdt_fld_ops[] = {
6041         DEF_FLD_HNDL_F(0, QUERY, (int (*)(struct mdt_thread_info *))fld_query)
6042 };
6043
6044 static struct mdt_opc_slice mdt_fld_handlers[] = {
6045         {
6046                 .mos_opc_start = FLD_QUERY,
6047                 .mos_opc_end   = FLD_LAST_OPC,
6048                 .mos_hs        = mdt_fld_ops
6049         },
6050         {
6051                 .mos_hs        = NULL
6052         }
6053 };
6054
6055 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
6056 MODULE_DESCRIPTION("Lustre Meta-data Target ("LUSTRE_MDT_NAME")");
6057 MODULE_LICENSE("GPL");
6058
6059 CFS_MODULE_PARM(mdt_num_threads, "ul", ulong, 0444,
6060                 "number of mdt service threads to start");
6061
6062 cfs_module(mdt, "0.2.0", mdt_mod_init, mdt_mod_exit);