lustre/mdt/mdt_open.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  linux/mdt/mdt_open.c
   5  *  Lustre Metadata Target (mdt) open/close file handling
   6  *
   7  *  Copyright (C) 2002-2006 Cluster File Systems, Inc.
   8  *   Author: Huang Hua <huanghua@clusterfs.com>
   9  *
  10  *   This file is part of the Lustre file system, http://www.lustre.org
  11  *   Lustre is a trademark of Cluster File Systems, Inc.
  12  *
  13  *   You may have signed or agreed to another license before downloading
  14  *   this software.  If so, you are bound by the terms and conditions
  15  *   of that agreement, and the following does not apply to you.  See the
  16  *   LICENSE file included with this distribution for more information.
  17  *
  18  *   If you did not agree to a different license, then this copy of Lustre
  19  *   is open source software; you can redistribute it and/or modify it
  20  *   under the terms of version 2 of the GNU General Public License as
  21  *   published by the Free Software Foundation.
  22  *
  23  *   In either case, Lustre is distributed in the hope that it will be
  24  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  25  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  26  *   license text for more details.
  27  */
  28
  29 #ifndef EXPORT_SYMTAB
  30 # define EXPORT_SYMTAB
  31 #endif
  32 #define DEBUG_SUBSYSTEM S_MDS
  33
  34 #include <linux/lustre_acl.h>
  35 #include <lustre_mds.h>
  36 #include "mdt_internal.h"
  37
  38 /* we do nothing because we do not have refcount now */
  39 static void mdt_mfd_get(void *mfdp)
  40 {
  41 }
  42
  43 /* Create a new mdt_file_data struct, initialize it,
  44  * and insert it to global hash table */
  45 struct mdt_file_data *mdt_mfd_new(void)
  46 {
  47         struct mdt_file_data *mfd;
  48         ENTRY;
  49
  50         OBD_ALLOC_PTR(mfd);
  51         if (mfd != NULL) {
  52                 INIT_LIST_HEAD(&mfd->mfd_handle.h_link);
  53                 INIT_LIST_HEAD(&mfd->mfd_list);
  54                 class_handle_hash(&mfd->mfd_handle, mdt_mfd_get);
  55         }
  56         RETURN(mfd);
  57 }
  58
  59 /* Find the mfd pointed to by handle in global hash table. */
  60 struct mdt_file_data *mdt_handle2mfd(const struct lustre_handle *handle)
  61 {
  62         ENTRY;
  63         LASSERT(handle != NULL);
  64         RETURN(class_handle2object(handle->cookie));
  65 }
  66
  67 /* free mfd */
  68 void mdt_mfd_free(struct mdt_file_data *mfd)
  69 {
  70         LASSERT(list_empty(&mfd->mfd_handle.h_link));
  71         LASSERT(list_empty(&mfd->mfd_list));
  72         OBD_FREE_PTR(mfd);
  73 }
  74
  75 static int mdt_create_data(struct mdt_thread_info *info,
  76                            struct mdt_object *p, struct mdt_object *o)
  77 {
  78         struct md_create_spec *spec = &info->mti_spec;
  79         struct md_attr        *ma = &info->mti_attr;
  80         int rc;
  81         ENTRY;
  82
  83         if ((spec->sp_cr_flags & MDS_OPEN_DELAY_CREATE) ||
  84             !(spec->sp_cr_flags & FMODE_WRITE))
  85                 RETURN(0);
  86
  87         ma->ma_need = MA_INODE | MA_LOV;
  88         ma->ma_valid = 0;
  89         rc = mdo_create_data(info->mti_env,
  90                              p ? mdt_object_child(p) : NULL,
  91                              mdt_object_child(o), spec, ma);
  92         RETURN(rc);
  93 }
  94
  95 static int mdt_epoch_opened(struct mdt_object *mo)
  96 {
  97         return mo->mot_epochcount;
  98 }
  99
 100 int mdt_sizeonmds_enabled(struct mdt_object *mo)
 101 {
 102         return !mo->mot_ioepoch;
 103 }
 104
 105 /* Re-enable Size-on-MDS. */
 106 void mdt_sizeonmds_enable(struct mdt_thread_info *info,
 107                           struct mdt_object *mo)
 108 {
 109        spin_lock(&info->mti_mdt->mdt_ioepoch_lock);
 110        if (info->mti_epoch->ioepoch == mo->mot_ioepoch) {
 111                 mo->mot_ioepoch = 0;
 112                 mo->mot_flags = 0;
 113        }
 114        spin_unlock(&info->mti_mdt->mdt_ioepoch_lock);
 115 }
 116
 117 /* Open the epoch. Epoch open is allowed if @writecount is not negative.
 118  * The epoch and writecount handling is performed under the mdt_ioepoch_lock. */
 119 int mdt_epoch_open(struct mdt_thread_info *info, struct mdt_object *o)
 120 {
 121         struct mdt_device *mdt = info->mti_mdt;
 122         int cancel = 0;
 123         int rc = 0;
 124         ENTRY;
 125
 126         if (!S_ISREG(lu_object_attr(&o->mot_obj.mo_lu)))
 127                 RETURN(0);
 128
 129         spin_lock(&mdt->mdt_ioepoch_lock);
 130         if (mdt_epoch_opened(o)) {
 131                 /* Epoch continues even if there is no writers yet. */
 132                 CDEBUG(D_INODE, "continue epoch "LPU64" for "DFID"\n",
 133                        o->mot_ioepoch, PFID(mdt_object_fid(o)));
 134         } else {
 135                 if (info->mti_replayepoch > mdt->mdt_ioepoch)
 136                         mdt->mdt_ioepoch = info->mti_replayepoch;
 137                 else
 138                         mdt->mdt_ioepoch++;
 139                 o->mot_ioepoch = info->mti_replayepoch ?
 140                         info->mti_replayepoch : mdt->mdt_ioepoch;
 141                 CDEBUG(D_INODE, "starting epoch "LPU64" for "DFID"\n",
 142                        mdt->mdt_ioepoch, PFID(mdt_object_fid(o)));
 143                 cancel = 1;
 144         }
 145         o->mot_epochcount++;
 146         spin_unlock(&mdt->mdt_ioepoch_lock);
 147
 148         /* Cancel Size-on-MDS attributes on clients if not truncate.
 149          * In the later case, mdt_reint_setattr will do it. */
 150         if (cancel && (info->mti_rr.rr_fid1 != NULL)) {
 151                 struct mdt_lock_handle  *lh = &info->mti_lh[MDT_LH_CHILD];
 152                 mdt_lock_reg_init(lh, LCK_EX);
 153                 rc = mdt_object_lock(info, o, lh, MDS_INODELOCK_UPDATE,
 154                                      MDT_LOCAL_LOCK);
 155                 if (rc == 0)
 156                         mdt_object_unlock(info, o, lh, 1);
 157         }
 158         RETURN(rc);
 159 }
 160
 161 /* Update the on-disk attributes if needed and re-enable Size-on-MDS caching. */
 162 static int mdt_sizeonmds_update(struct mdt_thread_info *info,
 163                                 struct mdt_object *o)
 164 {
 165         ENTRY;
 166
 167         CDEBUG(D_INODE, "Closing epoch "LPU64" on "DFID". Count %d\n",
 168                o->mot_ioepoch, PFID(mdt_object_fid(o)), o->mot_epochcount);
 169
 170         if (info->mti_attr.ma_attr.la_valid & LA_SIZE)
 171                 /* Do Size-on-MDS attribute update.
 172                  * Size-on-MDS is re-enabled inside. */
 173                 RETURN(mdt_attr_set(info, o, 0));
 174         else
 175                 mdt_sizeonmds_enable(info, o);
 176         RETURN(0);
 177 }
 178
 179 /* Epoch closes.
 180  * Returns 1 if epoch does not close.
 181  * Returns 0 if epoch closes.
 182  * Returns -EAGAIN if epoch closes but an Size-on-MDS Update is still needed
 183  * from the client. */
 184 static int mdt_epoch_close(struct mdt_thread_info *info, struct mdt_object *o)
 185 {
 186         int eviction = (mdt_info_req(info) == NULL ? 1 : 0);
 187         struct lu_attr *la = &info->mti_attr.ma_attr;
 188         int achange = 0;
 189         int opened;
 190         int rc = 1;
 191         ENTRY;
 192
 193         if (!S_ISREG(lu_object_attr(&o->mot_obj.mo_lu)))
 194                 RETURN(0);
 195
 196         spin_lock(&info->mti_mdt->mdt_ioepoch_lock);
 197
 198         /* Epoch closes only if client tells about it or eviction occures. */
 199         if (eviction || (info->mti_epoch->flags & MF_EPOCH_CLOSE)) {
 200                 LASSERT(o->mot_epochcount);
 201                 o->mot_epochcount--;
 202
 203                 CDEBUG(D_INODE, "Closing epoch "LPU64" on "DFID". Count %d\n",
 204                        o->mot_ioepoch, PFID(mdt_object_fid(o)),
 205                        o->mot_epochcount);
 206
 207                 if (!eviction)
 208                         achange = (info->mti_epoch->flags & MF_SOM_CHANGE);
 209
 210                 rc = 0;
 211                 if (!eviction && !mdt_epoch_opened(o)) {
 212                         /* Epoch ends. Is an Size-on-MDS update needed? */
 213                         if (o->mot_flags & MF_SOM_CHANGE) {
 214                                 /* Some previous writer changed the attribute.
 215                                  * Do not believe to the current Size-on-MDS
 216                                  * update, re-ask client. */
 217                                 rc = -EAGAIN;
 218                         } else if (!(la->la_valid & LA_SIZE) && achange) {
 219                                 /* Attributes were changed by the last writer
 220                                  * only but no Size-on-MDS update is received.*/
 221                                 rc = -EAGAIN;
 222                         }
 223                 }
 224
 225                 if (achange || eviction)
 226                         o->mot_flags |= MF_SOM_CHANGE;
 227         }
 228
 229         opened = mdt_epoch_opened(o);
 230         spin_unlock(&info->mti_mdt->mdt_ioepoch_lock);
 231
 232         /* XXX: if eviction occured, do nothing yet. */
 233         if ((rc == 0) && !opened && !eviction) {
 234                 /* Epoch ends and wanted Size-on-MDS update is obtained. */
 235                 rc = mdt_sizeonmds_update(info, o);
 236         }
 237         RETURN(rc);
 238 }
 239
 240 int mdt_write_get(struct mdt_device *mdt, struct mdt_object *o)
 241 {
 242         int rc = 0;
 243         ENTRY;
 244         spin_lock(&mdt->mdt_ioepoch_lock);
 245         if (o->mot_writecount < 0)
 246                 rc = -ETXTBSY;
 247         else
 248                 o->mot_writecount++;
 249         spin_unlock(&mdt->mdt_ioepoch_lock);
 250         RETURN(rc);
 251 }
 252
 253 static void mdt_write_put(struct mdt_device *mdt, struct mdt_object *o)
 254 {
 255         ENTRY;
 256         spin_lock(&mdt->mdt_ioepoch_lock);
 257         o->mot_writecount--;
 258         spin_unlock(&mdt->mdt_ioepoch_lock);
 259         EXIT;
 260 }
 261
 262 static int mdt_write_deny(struct mdt_device *mdt, struct mdt_object *o)
 263 {
 264         int rc = 0;
 265         ENTRY;
 266         spin_lock(&mdt->mdt_ioepoch_lock);
 267         if (o->mot_writecount > 0)
 268                 rc = -ETXTBSY;
 269         else
 270                 o->mot_writecount--;
 271         spin_unlock(&mdt->mdt_ioepoch_lock);
 272         RETURN(rc);
 273 }
 274
 275 static void mdt_write_allow(struct mdt_device *mdt, struct mdt_object *o)
 276 {
 277         ENTRY;
 278         spin_lock(&mdt->mdt_ioepoch_lock);
 279         o->mot_writecount++;
 280         spin_unlock(&mdt->mdt_ioepoch_lock);
 281         EXIT;
 282 }
 283
 284 /* there can be no real transaction so prepare the fake one */
 285 static void mdt_open_transno(struct mdt_thread_info* info)
 286 {
 287         struct mdt_device *mdt = info->mti_mdt;
 288         struct ptlrpc_request *req = mdt_info_req(info);
 289
 290         spin_lock(&mdt->mdt_transno_lock);
 291         if (info->mti_transno == 0) {
 292                 info->mti_transno = ++ mdt->mdt_last_transno;
 293         } else {
 294                 /* should be replay */
 295                 if (info->mti_transno > mdt->mdt_last_transno)
 296                         mdt->mdt_last_transno = info->mti_transno;
 297         }
 298         spin_unlock(&mdt->mdt_transno_lock);
 299
 300         CDEBUG(D_INODE, "open only: transno = %llu, last_committed = %llu\n",
 301                         info->mti_transno,
 302                         req->rq_export->exp_obd->obd_last_committed);
 303
 304         req->rq_transno = info->mti_transno;
 305         lustre_msg_set_transno(req->rq_repmsg, info->mti_transno);
 306         lustre_msg_set_last_xid(req->rq_repmsg, req->rq_xid);
 307 }
 308
 309 static int mdt_mfd_open(struct mdt_thread_info *info,
 310                         struct mdt_object *p,
 311                         struct mdt_object *o,
 312                         int flags,
 313                         int created,
 314                         struct ldlm_reply *rep)
 315 {
 316         struct ptlrpc_request  *req = mdt_info_req(info);
 317         struct mdt_export_data *med = &req->rq_export->exp_mdt_data;
 318         struct mdt_device      *mdt = info->mti_mdt;
 319         struct md_attr         *ma  = &info->mti_attr;
 320         struct lu_attr         *la  = &ma->ma_attr;
 321         struct mdt_file_data   *mfd;
 322         struct mdt_body        *repbody;
 323         int                     rc = 0;
 324         int                     isreg, isdir, islnk;
 325         struct list_head        *t;
 326         ENTRY;
 327
 328         LASSERT(ma->ma_valid & MA_INODE);
 329
 330         repbody = req_capsule_server_get(&info->mti_pill, &RMF_MDT_BODY);
 331
 332         isreg = S_ISREG(la->la_mode);
 333         isdir = S_ISDIR(la->la_mode);
 334         islnk = S_ISLNK(la->la_mode);
 335         mdt_pack_attr2body(info, repbody, la, mdt_object_fid(o));
 336
 337         if (med->med_rmtclient) {
 338                 void *buf = req_capsule_server_get(&info->mti_pill, &RMF_ACL);
 339
 340                 rc = mdt_pack_remote_perm(info, o, buf);
 341                 if (rc) {
 342                         repbody->valid &= ~OBD_MD_FLRMTPERM;
 343                         repbody->aclsize = 0;
 344                 } else {
 345                         repbody->valid |= OBD_MD_FLRMTPERM;
 346                         repbody->aclsize = sizeof(struct mdt_remote_perm);
 347                 }
 348         }
 349 #ifdef CONFIG_FS_POSIX_ACL
 350         else if (req->rq_export->exp_connect_flags & OBD_CONNECT_ACL) {
 351                 const struct lu_env *env = info->mti_env;
 352                 struct md_object *next = mdt_object_child(o);
 353                 struct lu_buf *buf = &info->mti_buf;
 354
 355                 buf->lb_buf = req_capsule_server_get(&info->mti_pill, &RMF_ACL);
 356                 buf->lb_len = req_capsule_get_size(&info->mti_pill, &RMF_ACL,
 357                                                    RCL_SERVER);
 358                 if (buf->lb_len > 0) {
 359                         rc = mo_xattr_get(env, next, buf,
 360                                           XATTR_NAME_ACL_ACCESS);
 361                         if (rc < 0) {
 362                                 if (rc == -ENODATA) {
 363                                         repbody->aclsize = 0;
 364                                         repbody->valid |= OBD_MD_FLACL;
 365                                         rc = 0;
 366                                 } else if (rc == -EOPNOTSUPP) {
 367                                         rc = 0;
 368                                 } else {
 369                                         CERROR("got acl size: %d\n", rc);
 370                                 }
 371                         } else {
 372                                 repbody->aclsize = rc;
 373                                 repbody->valid |= OBD_MD_FLACL;
 374                                 rc = 0;
 375                         }
 376                 }
 377         }
 378 #endif
 379
 380         if (mdt->mdt_opts.mo_mds_capa) {
 381                 struct lustre_capa *capa;
 382
 383                 capa = req_capsule_server_get(&info->mti_pill, &RMF_CAPA1);
 384                 LASSERT(capa);
 385                 capa->lc_opc = CAPA_OPC_MDS_DEFAULT;
 386                 rc = mo_capa_get(info->mti_env, mdt_object_child(o), capa, 0);
 387                 if (rc)
 388                         RETURN(rc);
 389                 repbody->valid |= OBD_MD_FLMDSCAPA;
 390         }
 391         if (mdt->mdt_opts.mo_oss_capa &&
 392             S_ISREG(lu_object_attr(&o->mot_obj.mo_lu))) {
 393                 struct lustre_capa *capa;
 394
 395                 capa = req_capsule_server_get(&info->mti_pill, &RMF_CAPA2);
 396                 LASSERT(capa);
 397                 capa->lc_opc = CAPA_OPC_OSS_DEFAULT | capa_open_opc(flags);
 398                 rc = mo_capa_get(info->mti_env, mdt_object_child(o), capa, 0);
 399                 if (rc)
 400                         RETURN(rc);
 401                 repbody->valid |= OBD_MD_FLOSSCAPA;
 402         }
 403
 404         /* if we are following a symlink, don't open; and
 405          * do not return open handle for special nodes as client required
 406          */
 407         if (islnk || (!isreg && !isdir &&
 408             (req->rq_export->exp_connect_flags & OBD_CONNECT_NODEVOH))) {
 409                 lustre_msg_set_transno(req->rq_repmsg, 0);
 410                 RETURN(0);
 411         }
 412
 413         mdt_set_disposition(info, rep, DISP_OPEN_OPEN);
 414         /* we need to return the existing object's fid back, so it is done
 415          * here, after preparing the reply */
 416         if (!created && (flags & MDS_OPEN_EXCL) && (flags & MDS_OPEN_CREAT))
 417                 RETURN(-EEXIST);
 418
 419         /* This can't be done earlier, we need to return reply body */
 420         if (isdir) {
 421                 if (flags & (MDS_OPEN_CREAT | FMODE_WRITE)) {
 422                         /* we are trying to create or
 423                          * write an existing dir. */
 424                         RETURN(-EISDIR);
 425                 }
 426         } else if (flags & MDS_OPEN_DIRECTORY)
 427                 RETURN(-ENOTDIR);
 428
 429         if (MDT_FAIL_CHECK(OBD_FAIL_MDS_OPEN_CREATE)) {
 430                 obd_fail_loc = OBD_FAIL_LDLM_REPLY | OBD_FAIL_ONCE;
 431                 RETURN(-EAGAIN);
 432         }
 433
 434         if (isreg && !(ma->ma_valid & MA_LOV)) {
 435                 /*
 436                  * No EA, check whether it is will set regEA and dirEA since in
 437                  * above attr get, these size might be zero, so reset it, to
 438                  * retrieve the MD after create obj.
 439                  */
 440                 ma->ma_lmm_size = req_capsule_get_size(&info->mti_pill,
 441                                                        &RMF_MDT_MD,
 442                                                        RCL_SERVER);
 443                 /* in replay case, p == NULL */
 444                 rc = mdt_create_data(info, p, o);
 445                 if (rc)
 446                         RETURN(rc);
 447         }
 448
 449         CDEBUG(D_INODE, "after open, ma_valid bit = "LPX64" lmm_size = %d\n",
 450                ma->ma_valid, ma->ma_lmm_size);
 451
 452         if (ma->ma_valid & MA_LOV) {
 453                 LASSERT(ma->ma_lmm_size != 0);
 454                 repbody->eadatasize = ma->ma_lmm_size;
 455                 if (isdir)
 456                         repbody->valid |= OBD_MD_FLDIREA;
 457                 else
 458                         repbody->valid |= OBD_MD_FLEASIZE;
 459         }
 460
 461         mfd = NULL;
 462         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT) {
 463                 spin_lock(&med->med_open_lock);
 464                 list_for_each(t, &med->med_open_head) {
 465                         mfd = list_entry(t, struct mdt_file_data, mfd_list);
 466                         if (mfd->mfd_xid == req->rq_xid) {
 467                                 break;
 468                         }
 469                         mfd = NULL;
 470                 }
 471                 spin_unlock(&med->med_open_lock);
 472
 473                 if (mfd != NULL) {
 474                         repbody->handle.cookie = mfd->mfd_handle.h_cookie;
 475                         RETURN(0);
 476                 }
 477         }
 478
 479         if (flags & FMODE_WRITE) {
 480                 rc = mdt_write_get(info->mti_mdt, o);
 481                 if (rc == 0) {
 482                         mdt_epoch_open(info, o);
 483                         repbody->ioepoch = o->mot_ioepoch;
 484                 }
 485         } else if (flags & MDS_FMODE_EXEC) {
 486                 rc = mdt_write_deny(info->mti_mdt, o);
 487         }
 488         if (rc)
 489                 RETURN(rc);
 490
 491         rc = mo_open(info->mti_env, mdt_object_child(o),
 492                      created ? flags | MDS_OPEN_CREATED : flags);
 493         if (rc)
 494                 RETURN(rc);
 495
 496         mfd = mdt_mfd_new();
 497         if (mfd != NULL) {
 498
 499                 /* keep a reference on this object for this open,
 500                 * and is released by mdt_mfd_close() */
 501                 mdt_object_get(info->mti_env, o);
 502                 /* open handling */
 503
 504                 mfd->mfd_mode = flags;
 505                 mfd->mfd_object = o;
 506                 mfd->mfd_xid = req->rq_xid;
 507
 508                 spin_lock(&med->med_open_lock);
 509                 list_add(&mfd->mfd_list, &med->med_open_head);
 510                 spin_unlock(&med->med_open_lock);
 511
 512                 repbody->handle.cookie = mfd->mfd_handle.h_cookie;
 513                 mdt_open_transno(info);
 514         } else
 515                 rc = -ENOMEM;
 516
 517         RETURN(rc);
 518 }
 519
 520 extern void mdt_req_from_mcd(struct ptlrpc_request *req,
 521                              struct mdt_client_data *mcd);
 522
 523 void mdt_reconstruct_open(struct mdt_thread_info *info,
 524                           struct mdt_lock_handle *lhc)
 525 {
 526         const struct lu_env *env = info->mti_env;
 527         struct mdt_device       *mdt  = info->mti_mdt;
 528         struct req_capsule      *pill = &info->mti_pill;
 529         struct ptlrpc_request   *req  = mdt_info_req(info);
 530         struct mdt_export_data  *med  = &req->rq_export->exp_mdt_data;
 531         struct mdt_client_data  *mcd  = med->med_mcd;
 532         struct md_attr          *ma   = &info->mti_attr;
 533         struct mdt_reint_record *rr   = &info->mti_rr;
 534         __u32                   flags = info->mti_spec.sp_cr_flags;
 535         struct ldlm_reply       *ldlm_rep;
 536         struct mdt_object       *parent;
 537         struct mdt_object       *child;
 538         struct mdt_body         *repbody;
 539         int                      rc;
 540         ENTRY;
 541
 542         LASSERT(pill->rc_fmt == &RQF_LDLM_INTENT_OPEN);
 543         ldlm_rep = req_capsule_server_get(&info->mti_pill, &RMF_DLM_REP);
 544         repbody = req_capsule_server_get(&info->mti_pill, &RMF_MDT_BODY);
 545
 546         ma->ma_lmm = req_capsule_server_get(&info->mti_pill, &RMF_MDT_MD);
 547         ma->ma_lmm_size = req_capsule_get_size(&info->mti_pill, &RMF_MDT_MD,
 548                                                RCL_SERVER);
 549         ma->ma_need = MA_INODE | MA_LOV;
 550         ma->ma_valid = 0;
 551
 552         mdt_req_from_mcd(req, med->med_mcd);
 553         mdt_set_disposition(info, ldlm_rep, mcd->mcd_last_data);
 554
 555         CERROR("This is reconstruct open: disp="LPX64", result=%d\n",
 556                 ldlm_rep->lock_policy_res1, req->rq_status);
 557
 558         if (mdt_get_disposition(ldlm_rep, DISP_OPEN_CREATE) &&
 559             req->rq_status != 0) {
 560                 /* We did not create successfully, return error to client. */
 561                 mdt_shrink_reply(info, DLM_REPLY_REC_OFF + 1, 1, 1);
 562                 GOTO(out, rc = req->rq_status);
 563         }
 564
 565         if (mdt_get_disposition(ldlm_rep, DISP_OPEN_CREATE)) {
 566                 /*
 567                  * We failed after creation, but we do not know in which step
 568                  * we failed. So try to check the child object.
 569                  */
 570                 parent = mdt_object_find(env, mdt, rr->rr_fid1);
 571                 LASSERT(!IS_ERR(parent));
 572
 573                 child = mdt_object_find(env, mdt, rr->rr_fid2);
 574                 LASSERT(!IS_ERR(child));
 575
 576                 rc = mdt_object_exists(child);
 577                 if (rc > 0) {
 578                         struct md_object *next;
 579                         next = mdt_object_child(child);
 580                         rc = mo_attr_get(env, next, ma);
 581                         if (rc == 0)
 582                               rc = mdt_mfd_open(info, parent, child,
 583                                                 flags, 1, ldlm_rep);
 584                 } else if (rc < 0) {
 585                         /* the child object was created on remote server */
 586                         repbody->fid1 = *rr->rr_fid2;
 587                         repbody->valid |= (OBD_MD_FLID | OBD_MD_MDS);
 588                         rc = 0;
 589                 } else if (rc == 0) {
 590                         /* the child does not exist, we should do regular open */
 591                         mdt_object_put(env, parent);
 592                         mdt_object_put(env, child);
 593                         GOTO(regular_open, 0);
 594                 }
 595                 mdt_object_put(env, parent);
 596                 mdt_object_put(env, child);
 597                 mdt_shrink_reply(info, DLM_REPLY_REC_OFF + 1, 1, 1);
 598                 GOTO(out, rc);
 599         } else {
 600 regular_open:
 601                 /* We did not try to create, so we are a pure open */
 602                 rc = mdt_reint_open(info, lhc);
 603         }
 604
 605         EXIT;
 606 out:
 607         req->rq_status = rc;
 608         lustre_msg_set_status(req->rq_repmsg, req->rq_status);
 609 }
 610
 611 static int mdt_open_by_fid(struct mdt_thread_info* info,
 612                            struct ldlm_reply *rep)
 613 {
 614         const struct lu_env     *env = info->mti_env;
 615         struct mdt_device       *mdt = info->mti_mdt;
 616         struct md_device        *next = mdt->mdt_child;
 617         __u32                    flags = info->mti_spec.sp_cr_flags;
 618         struct mdt_reint_record *rr = &info->mti_rr;
 619         struct md_attr          *ma = &info->mti_attr;
 620         struct mdt_object       *o;
 621         int                     rc;
 622         ENTRY;
 623
 624         o = mdt_object_find(info->mti_env, info->mti_mdt, rr->rr_fid2);
 625         if (IS_ERR(o))
 626                 RETURN(rc = PTR_ERR(o));
 627
 628         /* FIXME: capability in replay open request might have expired, disable
 629          * capability check for this kind of request. security hole?
 630          */
 631         if (mdt->mdt_opts.mo_mds_capa)
 632                 mdt->mdt_child->md_ops->mdo_init_capa_ctxt(env, next, 0, 0, 0,
 633                                                            0);
 634
 635         rc = mdt_object_exists(o);
 636         if (rc > 0) {
 637                 mdt_set_disposition(info, rep, (DISP_IT_EXECD |
 638                                                 DISP_LOOKUP_EXECD |
 639                                                 DISP_LOOKUP_POS));
 640
 641                 rc = mo_attr_get(env, mdt_object_child(o), ma);
 642                 if (rc == 0)
 643                         rc = mdt_mfd_open(info, NULL, o, flags, 0, rep);
 644         } else if (rc == 0) {
 645                 rc = -ENOENT;
 646         } else  {
 647                 /* the child object was created on remote server */
 648                 struct mdt_body *repbody;
 649                 repbody = req_capsule_server_get(&info->mti_pill, &RMF_MDT_BODY);
 650                 repbody->fid1 = *rr->rr_fid2;
 651                 repbody->valid |= (OBD_MD_FLID | OBD_MD_MDS);
 652                 rc = 0;
 653         }
 654
 655         if (mdt->mdt_opts.mo_mds_capa)
 656                 mdt->mdt_child->md_ops->mdo_init_capa_ctxt(env, next,
 657                                                 mdt->mdt_opts.mo_mds_capa,
 658                                                 mdt->mdt_capa_timeout,
 659                                                 mdt->mdt_capa_alg,
 660                                                 mdt->mdt_capa_keys);
 661
 662         mdt_object_put(info->mti_env, o);
 663         RETURN(rc);
 664 }
 665
 666 int mdt_pin(struct mdt_thread_info* info)
 667 {
 668         ENTRY;
 669         RETURN(err_serious(-EOPNOTSUPP));
 670 }
 671
 672 /* Cross-ref request. Currently it can only be a pure open (w/o create) */
 673 static int mdt_cross_open(struct mdt_thread_info* info,
 674                           const struct lu_fid *fid,
 675                           struct ldlm_reply *rep, __u32 flags)
 676 {
 677         struct md_attr    *ma = &info->mti_attr;
 678         struct mdt_object *o;
 679         int                rc;
 680         ENTRY;
 681
 682         o = mdt_object_find(info->mti_env, info->mti_mdt, fid);
 683         if (IS_ERR(o))
 684                 RETURN(rc = PTR_ERR(o));
 685
 686         rc = mdt_object_exists(o);
 687         if (rc > 0) {
 688                 mdt_set_capainfo(info, 0, fid, BYPASS_CAPA);
 689                 rc = mo_attr_get(info->mti_env, mdt_object_child(o), ma);
 690                 if (rc == 0)
 691                         rc = mdt_mfd_open(info, NULL, o, flags, 0, rep);
 692         } else if (rc == 0) {
 693                 /*
 694                  * FIXME: something wrong here, lookup was positive but there is
 695                  * no object!
 696                  */
 697                 CERROR("Cross-ref object doesn't exist!\n");
 698                 rc = -EFAULT;
 699         } else  {
 700                 /* FIXME: something wrong here, the object is on another MDS! */
 701                 CERROR("The object isn't on this server! FLD error?\n");
 702                 rc = -EFAULT;
 703         }
 704         mdt_object_put(info->mti_env, o);
 705         RETURN(rc);
 706 }
 707
 708 int mdt_reint_open(struct mdt_thread_info *info, struct mdt_lock_handle *lhc)
 709 {
 710         struct mdt_device      *mdt = info->mti_mdt;
 711         struct ptlrpc_request  *req = mdt_info_req(info);
 712         struct mdt_object      *parent;
 713         struct mdt_object      *child;
 714         struct mdt_lock_handle *lh;
 715         struct ldlm_reply      *ldlm_rep;
 716         struct mdt_body        *repbody;
 717         struct lu_fid          *child_fid = &info->mti_tmp_fid1;
 718         struct md_attr         *ma = &info->mti_attr;
 719         struct lu_attr         *la = &ma->ma_attr;
 720         __u32                   create_flags = info->mti_spec.sp_cr_flags;
 721         struct mdt_reint_record *rr = &info->mti_rr;
 722         int                     result;
 723         int                     created = 0;
 724         ENTRY;
 725
 726         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PAUSE_OPEN | OBD_FAIL_ONCE,
 727                          (obd_timeout + 1) / 4);
 728
 729         repbody = req_capsule_server_get(&info->mti_pill, &RMF_MDT_BODY);
 730
 731         ma->ma_lmm = req_capsule_server_get(&info->mti_pill, &RMF_MDT_MD);
 732         ma->ma_lmm_size = req_capsule_get_size(&info->mti_pill, &RMF_MDT_MD,
 733                                                RCL_SERVER);
 734         ma->ma_need = MA_INODE | MA_LOV;
 735         ma->ma_valid = 0;
 736
 737         LASSERT(info->mti_pill.rc_fmt == &RQF_LDLM_INTENT_OPEN);
 738         ldlm_rep = req_capsule_server_get(&info->mti_pill, &RMF_DLM_REP);
 739
 740         /* TODO: JOIN file */
 741         if (create_flags & MDS_OPEN_JOIN_FILE) {
 742                 CERROR("JOIN file will be supported soon\n");
 743                 GOTO(out, result = err_serious(-EOPNOTSUPP));
 744         }
 745
 746         CDEBUG(D_INODE, "I am going to open "DFID"/(%s->"DFID") "
 747                "cr_flag=0%o mode=0%06o msg_flag=0x%x\n",
 748                PFID(rr->rr_fid1), rr->rr_name,
 749                PFID(rr->rr_fid2), create_flags,
 750                la->la_mode, lustre_msg_get_flags(req->rq_reqmsg));
 751
 752         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) {
 753                 /* This is a replay request. */
 754                 result = mdt_open_by_fid(info, ldlm_rep);
 755
 756                 if (result != -ENOENT)
 757                         GOTO(out, result);
 758
 759                 /*
 760                  * We didn't find the correct object, so we need to re-create it
 761                  * via a regular replay.
 762                  */
 763                 if (!(create_flags & MDS_OPEN_CREAT)) {
 764                         DEBUG_REQ(D_ERROR, req,"OPEN & CREAT not in open replay.");
 765                         GOTO(out, result = -EFAULT);
 766                 }
 767                 CDEBUG(D_INFO, "Open replay did find object, continue as "
 768                        "regular open\n");
 769         }
 770
 771         if (MDT_FAIL_CHECK(OBD_FAIL_MDS_OPEN_PACK))
 772                 GOTO(out, result = err_serious(-ENOMEM));
 773
 774         mdt_set_disposition(info, ldlm_rep,
 775                             (DISP_IT_EXECD | DISP_LOOKUP_EXECD));
 776
 777         if (rr->rr_name[0] == 0) {
 778                 /* this is cross-ref open */
 779                 mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS);
 780                 result = mdt_cross_open(info, rr->rr_fid1, ldlm_rep,
 781                                         create_flags);
 782                 GOTO(out, result);
 783         }
 784
 785         lh = &info->mti_lh[MDT_LH_PARENT];
 786         if (!(create_flags & MDS_OPEN_CREAT)) {
 787                 mdt_lock_pdo_init(lh, LCK_PR, rr->rr_name,
 788                                   rr->rr_namelen);
 789         } else {
 790                 mdt_lock_pdo_init(lh, LCK_PW, rr->rr_name,
 791                                   rr->rr_namelen);
 792         }
 793         parent = mdt_object_find_lock(info, rr->rr_fid1, lh,
 794                                       MDS_INODELOCK_UPDATE);
 795         if (IS_ERR(parent))
 796                 GOTO(out, result = PTR_ERR(parent));
 797
 798         result = mdo_lookup(info->mti_env, mdt_object_child(parent),
 799                             rr->rr_name, child_fid);
 800         if (result != 0 && result != -ENOENT && result != -ESTALE)
 801                 GOTO(out_parent, result);
 802
 803         if (result == -ENOENT || result == -ESTALE) {
 804                 mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG);
 805                 if (result == -ESTALE) {
 806                         /*
 807                          * ESTALE means the parent is a dead(unlinked) dir, so
 808                          * it should return -ENOENT to in accordance with the
 809                          * original mds implementaion.
 810                          */
 811                         GOTO(out_parent, result = -ENOENT);
 812                 }
 813                 if (!(create_flags & MDS_OPEN_CREAT))
 814                         GOTO(out_parent, result);
 815                 *child_fid = *info->mti_rr.rr_fid2;
 816                 /* new object will be created. see the following */
 817         } else {
 818                 /*
 819                  * Check for O_EXCL is moved to the mdt_mfd_open, we need to
 820                  * return FID back in that case.
 821                  */
 822                 mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS);
 823         }
 824
 825         LASSERT(fid_is_sane(child_fid));
 826         child = mdt_object_find(info->mti_env, mdt, child_fid);
 827         if (IS_ERR(child))
 828                 GOTO(out_parent, result = PTR_ERR(child));
 829
 830         mdt_set_capainfo(info, 1, child_fid, BYPASS_CAPA);
 831         if (result == -ENOENT) {
 832                 /* Not found and with MDS_OPEN_CREAT: let's create it. */
 833                 mdt_set_disposition(info, ldlm_rep, DISP_OPEN_CREATE);
 834
 835                 /* Let lower layers know what is lock mode on directory. */
 836                 info->mti_spec.sp_cr_mode =
 837                         mdt_dlm_mode2mdl_mode(lh->mlh_pdo_mode);
 838                 result = mdo_create(info->mti_env,
 839                                     mdt_object_child(parent),
 840                                     rr->rr_name,
 841                                     mdt_object_child(child),
 842                                     &info->mti_spec,
 843                                     &info->mti_attr);
 844                 if (result == -ERESTART) {
 845                         mdt_clear_disposition(info, ldlm_rep, DISP_OPEN_CREATE);
 846                         GOTO(out_child, result);
 847                 }
 848                 else {
 849                         if (result != 0)
 850                                 GOTO(out_child, result);
 851                 }
 852                 created = 1;
 853         } else {
 854                 /* We have to get attr & lov ea for this object */
 855                 result = mo_attr_get(info->mti_env, mdt_object_child(child),
 856                                      ma);
 857                 /*
 858                  * The object is on remote node, return its FID for remote open.
 859                  */
 860                 if (result == -EREMOTE) {
 861                         int rc;
 862
 863                         /*
 864                          * Check if this lock already was sent to client and
 865                          * this is resent case. For resent case do not take lock
 866                          * again, use what is already granted.
 867                          */
 868                         LASSERT(lhc != NULL);
 869
 870                         if (lustre_handle_is_used(&lhc->mlh_reg_lh)) {
 871                                 struct ldlm_lock *lock;
 872
 873                                 LASSERT(lustre_msg_get_flags(req->rq_reqmsg) &
 874                                         MSG_RESENT);
 875
 876                                 lock = ldlm_handle2lock(&lhc->mlh_reg_lh);
 877                                 if (!lock) {
 878                                         CERROR("Invalid lock handle "LPX64"\n",
 879                                                lhc->mlh_reg_lh.cookie);
 880                                         LBUG();
 881                                 }
 882                                 LASSERT(fid_res_name_eq(mdt_object_fid(child),
 883                                                         &lock->l_resource->lr_name));
 884                                 LDLM_LOCK_PUT(lock);
 885                                 rc = 0;
 886                         } else {
 887                                 mdt_lock_handle_init(lhc);
 888                                 mdt_lock_reg_init(lhc, LCK_PR);
 889
 890                                 rc = mdt_object_lock(info, child, lhc,
 891                                                      MDS_INODELOCK_LOOKUP,
 892                                                      MDT_LOCAL_LOCK);
 893                         }
 894                         repbody->fid1 = *mdt_object_fid(child);
 895                         repbody->valid |= (OBD_MD_FLID | OBD_MD_MDS);
 896                         if (rc != 0)
 897                                 result = rc;
 898                         GOTO(out_child, result);
 899                 }
 900         }
 901
 902         /* Try to open it now. */
 903         result = mdt_mfd_open(info, parent, child, create_flags,
 904                               created, ldlm_rep);
 905         GOTO(finish_open, result);
 906
 907 finish_open:
 908         if (result != 0 && created) {
 909                 int rc2;
 910                 ma->ma_need = 0;
 911                 ma->ma_valid = 0;
 912                 ma->ma_cookie_size = 0;
 913                 rc2 = mdo_unlink(info->mti_env,
 914                                  mdt_object_child(parent),
 915                                  mdt_object_child(child),
 916                                  rr->rr_name,
 917                                  &info->mti_attr);
 918                 if (rc2 != 0)
 919                         CERROR("error in cleanup of open");
 920         }
 921 out_child:
 922         mdt_object_put(info->mti_env, child);
 923 out_parent:
 924         mdt_object_unlock_put(info, parent, lh, result);
 925 out:
 926         mdt_shrink_reply(info, DLM_REPLY_REC_OFF + 1, 1, 1);
 927         if (result)
 928                 lustre_msg_set_transno(req->rq_repmsg, 0);
 929         return result;
 930 }
 931
 932 #define MFD_CLOSED(mode) (((mode) & ~(FMODE_EPOCH | FMODE_SOM | \
 933                                       FMODE_EPOCHLCK)) == FMODE_CLOSED)
 934
 935 static int mdt_mfd_closed(struct mdt_file_data *mfd)
 936 {
 937         return ((mfd == NULL) || MFD_CLOSED(mfd->mfd_mode));
 938 }
 939
 940 int mdt_mfd_close(struct mdt_thread_info *info, struct mdt_file_data *mfd)
 941 {
 942         struct mdt_object *o = mfd->mfd_object;
 943         struct md_object *next = mdt_object_child(o);
 944         struct md_attr *ma = &info->mti_attr;
 945         int rc = 0, ret = 0;
 946         int mode;
 947         ENTRY;
 948
 949         mode = mfd->mfd_mode;
 950         mfd->mfd_mode = FMODE_CLOSED;
 951
 952         if ((mode & FMODE_WRITE) || (mode & FMODE_EPOCHLCK)) {
 953                 mdt_write_put(info->mti_mdt, o);
 954                 ret = mdt_epoch_close(info, o);
 955         } else if (mode & MDS_FMODE_EXEC) {
 956                 mdt_write_allow(info->mti_mdt, o);
 957         } else if (mode & FMODE_EPOCH) {
 958                 ret = mdt_epoch_close(info, o);
 959         }
 960
 961         ma->ma_need |= MA_INODE;
 962
 963         if (!MFD_CLOSED(mode))
 964                 rc = mo_close(info->mti_env, next, ma);
 965         else if (ret == -EAGAIN)
 966                 rc = mo_attr_get(info->mti_env, next, ma);
 967
 968         /* If the object is unlinked, do not try to re-enable SIZEONMDS */
 969         if ((ret == -EAGAIN) && (ma->ma_valid & MA_INODE) &&
 970             (ma->ma_attr.la_nlink == 0))
 971         {
 972                 ret = 0;
 973         }
 974
 975         if ((ret == -EAGAIN) || (ret == 1)) {
 976                 struct mdt_export_data *med;
 977                 /* The epoch has not closed or Size-on-MDS update is needed.
 978                  * Put mfd back into the list. */
 979                 mfd->mfd_mode = (ret == 1 ? FMODE_EPOCH : FMODE_SOM);
 980
 981                 LASSERT(mdt_info_req(info));
 982                 med = &mdt_info_req(info)->rq_export->exp_mdt_data;
 983                 spin_lock(&med->med_open_lock);
 984                 list_add(&mfd->mfd_list, &med->med_open_head);
 985                 class_handle_hash_back(&mfd->mfd_handle);
 986                 spin_unlock(&med->med_open_lock);
 987                 if (ret == 1)
 988                         ret = 0;
 989                 else {
 990                         CDEBUG(D_INODE, "Size-on-MDS attribute update is "
 991                                "needed on "DFID"\n", PFID(mdt_object_fid(o)));
 992                 }
 993         } else {
 994                 mdt_mfd_free(mfd);
 995                 mdt_object_put(info->mti_env, o);
 996         }
 997
 998         RETURN(rc ? rc : ret);
 999 }
1000
1001 int mdt_close(struct mdt_thread_info *info)
1002 {
1003         struct mdt_export_data *med;
1004         struct mdt_file_data   *mfd;
1005         struct mdt_object      *o;
1006         struct md_attr         *ma = &info->mti_attr;
1007         struct mdt_body        *repbody = NULL;
1008         int rc, ret = 0;
1009         ENTRY;
1010
1011         /* Close may come with the Size-on-MDS update. Unpack it. */
1012         rc = mdt_close_unpack(info);
1013         if (rc)
1014                 RETURN(err_serious(rc));
1015
1016         LASSERT(info->mti_epoch);
1017
1018         req_capsule_set_size(&info->mti_pill, &RMF_MDT_MD, RCL_SERVER,
1019                              info->mti_mdt->mdt_max_mdsize);
1020         req_capsule_set_size(&info->mti_pill, &RMF_LOGCOOKIES, RCL_SERVER,
1021                              info->mti_mdt->mdt_max_cookiesize);
1022         rc = req_capsule_pack(&info->mti_pill);
1023         /* Continue to close handle even if we can not pack reply */
1024         if (rc == 0) {
1025                 repbody = req_capsule_server_get(&info->mti_pill,
1026                                                  &RMF_MDT_BODY);
1027                 ma->ma_lmm = req_capsule_server_get(&info->mti_pill,
1028                                                     &RMF_MDT_MD);
1029                 ma->ma_lmm_size = req_capsule_get_size(&info->mti_pill,
1030                                                        &RMF_MDT_MD,
1031                                                        RCL_SERVER);
1032                 ma->ma_cookie = req_capsule_server_get(&info->mti_pill,
1033                                                        &RMF_LOGCOOKIES);
1034                 ma->ma_cookie_size = req_capsule_get_size(&info->mti_pill,
1035                                                           &RMF_LOGCOOKIES,
1036                                                           RCL_SERVER);
1037                 ma->ma_need = MA_INODE | MA_LOV | MA_COOKIE;
1038                 ma->ma_valid = 0;
1039                 repbody->eadatasize = 0;
1040                 repbody->aclsize = 0;
1041         } else
1042                 rc = err_serious(rc);
1043
1044         med = &mdt_info_req(info)->rq_export->exp_mdt_data;
1045
1046         spin_lock(&med->med_open_lock);
1047         mfd = mdt_handle2mfd(&(info->mti_epoch->handle));
1048         if (mdt_mfd_closed(mfd)) {
1049                 spin_unlock(&med->med_open_lock);
1050                 CDEBUG(D_INODE, "no handle for file close: fid = "DFID
1051                        ": cookie = "LPX64"\n", PFID(info->mti_rr.rr_fid1),
1052                        info->mti_epoch->handle.cookie);
1053                 rc = err_serious(-ESTALE);
1054         } else {
1055                 class_handle_unhash(&mfd->mfd_handle);
1056                 list_del_init(&mfd->mfd_list);
1057                 spin_unlock(&med->med_open_lock);
1058
1059                 /* Do not lose object before last unlink. */
1060                 o = mfd->mfd_object;
1061                 mdt_object_get(info->mti_env, o);
1062                 ret = mdt_mfd_close(info, mfd);
1063                 if (repbody != NULL)
1064                         rc = mdt_handle_last_unlink(info, o, ma);
1065                 mdt_object_put(info->mti_env, o);
1066         }
1067         if (repbody != NULL)
1068                 mdt_shrink_reply(info, REPLY_REC_OFF + 1, 0, 0);
1069
1070         if (MDT_FAIL_CHECK(OBD_FAIL_MDS_CLOSE_PACK))
1071                 RETURN(err_serious(-ENOMEM));
1072
1073         RETURN(rc ? rc : ret);
1074 }
1075
1076 int mdt_done_writing(struct mdt_thread_info *info)
1077 {
1078         struct mdt_body        *repbody = NULL;
1079         struct mdt_export_data *med;
1080         struct mdt_file_data   *mfd;
1081         int rc;
1082         ENTRY;
1083
1084         rc = req_capsule_pack(&info->mti_pill);
1085         if (rc)
1086                 RETURN(err_serious(rc));
1087
1088         repbody = req_capsule_server_get(&info->mti_pill,
1089                                          &RMF_MDT_BODY);
1090         repbody->eadatasize = 0;
1091         repbody->aclsize = 0;
1092
1093         /* Done Writing may come with the Size-on-MDS update. Unpack it. */
1094         rc = mdt_close_unpack(info);
1095         if (rc)
1096                 RETURN(err_serious(rc));
1097
1098         med = &mdt_info_req(info)->rq_export->exp_mdt_data;
1099         spin_lock(&med->med_open_lock);
1100         mfd = mdt_handle2mfd(&(info->mti_epoch->handle));
1101         if (mfd == NULL) {
1102                 spin_unlock(&med->med_open_lock);
1103                 CDEBUG(D_INODE, "no handle for file close: fid = "DFID
1104                        ": cookie = "LPX64"\n", PFID(info->mti_rr.rr_fid1),
1105                        info->mti_epoch->handle.cookie);
1106                 rc = -ESTALE;
1107         } else {
1108                 LASSERT((mfd->mfd_mode == FMODE_EPOCH) ||
1109                         (mfd->mfd_mode == FMODE_EPOCHLCK));
1110                 class_handle_unhash(&mfd->mfd_handle);
1111                 list_del_init(&mfd->mfd_list);
1112                 spin_unlock(&med->med_open_lock);
1113
1114                 /* Set EPOCH CLOSE flag if not set by client. */
1115                 info->mti_epoch->flags |= MF_EPOCH_CLOSE;
1116                 info->mti_attr.ma_valid = 0;
1117                 rc = mdt_mfd_close(info, mfd);
1118         }
1119         RETURN(rc);
1120 }