LU-307 Send parent FID from client to MDS on NFS open for striping info

Which is useful for delay created cases. and it is mainly for Lustre re-export
through NFS. If client does not give such hint, MDS will try to find parent by
the first fid in the link EA. Another fixes are about the protection of delay
creating OSTs objects concurrently.

Change-Id: I5dd72e0452303ca5ab7bff035ff85f83b5862d66
Signed-off-by: nasf <yong.fan@whamcloud.com>
Reviewed-on: http://review.whamcloud.com/557
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>
Tested-by: Hudson
Reviewed-by: wangdi <di.wang@whamcloud.com>
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/linux/lustre_compat25.h b/lustre/include/linux/lustre_compat25.h
index 9df5862..a1291c9 100644 (file)
--- a/lustre/include/linux/lustre_compat25.h
+++ b/lustre/include/linux/lustre_compat25.h
@@ -452,8 +452,10 @@ static inline struct dentry *d_obtain_alias(struct inode *inode)
 {
        struct dentry *anon = d_alloc_anon(inode);
 
 {
        struct dentry *anon = d_alloc_anon(inode);
 

-       if (anon == NULL)
+       if (anon == NULL) {

                iput(inode);
                iput(inode);

+                anon = ERR_PTR(-ENOMEM);
+        }

 
        return anon;
 }
 
        return anon;
 }

diff --git a/lustre/include/lustre_mds.h b/lustre/include/lustre_mds.h
index 2a586ff..f0ceb00 100644 (file)
--- a/lustre/include/lustre_mds.h
+++ b/lustre/include/lustre_mds.h
@@ -87,7 +87,7 @@ int mds_log_op_unlink(struct obd_device *, struct lov_mds_md *, int,
 #define MDD_OBD_UUID     "mdd_obd_uuid"
 #define MDD_OBD_TYPE     "mds"
 
 #define MDD_OBD_UUID     "mdd_obd_uuid"
 #define MDD_OBD_TYPE     "mds"
 

-static inline int md_should_create(__u32 flags)
+static inline int md_should_create(__u64 flags)

 {
        return !(flags & MDS_OPEN_DELAY_CREATE ||
                !(flags & FMODE_WRITE));
 {
        return !(flags & MDS_OPEN_DELAY_CREATE ||
                !(flags & FMODE_WRITE));

diff --git a/lustre/include/md_object.h b/lustre/include/md_object.h
index 9a7e3ee..fbdd34f 100644 (file)
--- a/lustre/include/md_object.h
+++ b/lustre/include/md_object.h
@@ -126,7 +126,8 @@ enum ma_valid {
         MA_LOV_DEF   = (1 << 6),
 /* (Layout lock will used #7 here) */
         MA_HSM       = (1 << 8),
         MA_LOV_DEF   = (1 << 6),
 /* (Layout lock will used #7 here) */
         MA_HSM       = (1 << 8),

-        MA_SOM       = (1 << 9)
+        MA_SOM       = (1 << 9),
+        MA_PFID      = (1 << 10)

 };
 
 typedef enum {
 };
 
 typedef enum {


@@ -177,6 +178,7 @@ struct md_attr {
         struct lustre_capa     *ma_capa;
         struct md_hsm           ma_hsm;
         struct md_som_data     *ma_som;
         struct lustre_capa     *ma_capa;
         struct md_hsm           ma_hsm;
         struct md_som_data     *ma_som;

+        struct lu_fid           ma_pfid;

 };
 
 /** Additional parameters for create */
 };
 
 /** Additional parameters for create */

diff --git a/lustre/llite/file.c b/lustre/llite/file.c
index b6062ed..c186b5a 100644 (file)
--- a/lustre/llite/file.c
+++ b/lustre/llite/file.c
@@ -346,6 +346,7 @@ static int ll_intent_file_open(struct file *file, void *lmm,
         const int len = file->f_dentry->d_name.len;
         struct md_op_data *op_data;
         struct ptlrpc_request *req;
         const int len = file->f_dentry->d_name.len;
         struct md_op_data *op_data;
         struct ptlrpc_request *req;

+        __u32 opc = LUSTRE_OPC_ANY;

         int rc;
         ENTRY;
 
         int rc;
         ENTRY;
 


@@ -361,12 +362,15 @@ static int ll_intent_file_open(struct file *file, void *lmm,
          * makes a good candidate for using OPEN lock */
         /* If lmmsize & lmm are not 0, we are just setting stripe info
          * parameters. No need for the open lock */
          * makes a good candidate for using OPEN lock */
         /* If lmmsize & lmm are not 0, we are just setting stripe info
          * parameters. No need for the open lock */

-        if (!lmm && !lmmsize)
+        if (lmm == NULL && lmmsize == 0) {

                 itp->it_flags |= MDS_OPEN_LOCK;
                 itp->it_flags |= MDS_OPEN_LOCK;

+                if (itp->it_flags & FMODE_WRITE)
+                        opc = LUSTRE_OPC_CREATE;
+        }

 
         op_data  = ll_prep_md_op_data(NULL, parent->d_inode,
                                       file->f_dentry->d_inode, name, len,
 
         op_data  = ll_prep_md_op_data(NULL, parent->d_inode,
                                       file->f_dentry->d_inode, name, len,

-                                      O_RDWR, LUSTRE_OPC_ANY, NULL);
+                                      O_RDWR, opc, NULL);

         if (IS_ERR(op_data))
                 RETURN(PTR_ERR(op_data));
 
         if (IS_ERR(op_data))
                 RETURN(PTR_ERR(op_data));
 


@@ -1460,12 +1464,13 @@ static int ll_lov_setstripe(struct inode *inode, struct file *file,
 static int ll_lov_getstripe(struct inode *inode, unsigned long arg)
 {
         struct lov_stripe_md *lsm = ll_i2info(inode)->lli_smd;
 static int ll_lov_getstripe(struct inode *inode, unsigned long arg)
 {
         struct lov_stripe_md *lsm = ll_i2info(inode)->lli_smd;

+        int rc = -ENODATA;
+        ENTRY;

 
 

-        if (!lsm)
-                RETURN(-ENODATA);
-
-        return obd_iocontrol(LL_IOC_LOV_GETSTRIPE, ll_i2dtexp(inode), 0, lsm,
-                            (void *)arg);
+        if (lsm != NULL)
+                rc = obd_iocontrol(LL_IOC_LOV_GETSTRIPE, ll_i2dtexp(inode), 0,
+                                   lsm, (void *)arg);
+        RETURN(rc);

 }
 
 int ll_get_grouplock(struct inode *inode, struct file *file, unsigned long arg)
 }
 
 int ll_get_grouplock(struct inode *inode, struct file *file, unsigned long arg)

diff --git a/lustre/llite/llite_internal.h b/lustre/llite/llite_internal.h
index 766fd29..7d0f68d 100644 (file)
--- a/lustre/llite/llite_internal.h
+++ b/lustre/llite/llite_internal.h
@@ -167,6 +167,9 @@ struct ll_inode_info {
 
         /* identifying fields for both metadata and data stacks. */
         struct lu_fid           lli_fid;
 
         /* identifying fields for both metadata and data stacks. */
         struct lu_fid           lli_fid;

+        /* Parent fid for accessing default stripe data on parent directory
+         * for allocating OST objects after a mknod() and later open-by-FID. */
+        struct lu_fid           lli_pfid;

         struct lov_stripe_md   *lli_smd;
 
         /* fid capability */
         struct lov_stripe_md   *lli_smd;
 
         /* fid capability */

diff --git a/lustre/llite/llite_lib.c b/lustre/llite/llite_lib.c
index 6a86df3..3ee3ca0 100644 (file)
--- a/lustre/llite/llite_lib.c
+++ b/lustre/llite/llite_lib.c
@@ -828,6 +828,7 @@ void ll_lli_init(struct ll_inode_info *lli)
         CFS_INIT_LIST_HEAD(&lli->lli_oss_capas);
         cfs_spin_lock_init(&lli->lli_sa_lock);
         cfs_sema_init(&lli->lli_readdir_sem, 1);
         CFS_INIT_LIST_HEAD(&lli->lli_oss_capas);
         cfs_spin_lock_init(&lli->lli_sa_lock);
         cfs_sema_init(&lli->lli_readdir_sem, 1);

+        fid_zero(&lli->lli_pfid);

 }
 
 static inline int ll_bdi_register(struct backing_dev_info *bdi)
 }
 
 static inline int ll_bdi_register(struct backing_dev_info *bdi)


@@ -1518,7 +1519,9 @@ void ll_update_inode(struct inode *inode, struct lustre_md *md)
                          * but lov raid0 is not setup yet and parallel e.g.
                          * glimpse would try to use uninitialized lov */
                         cl_inode_init(inode, md);
                          * but lov raid0 is not setup yet and parallel e.g.
                          * glimpse would try to use uninitialized lov */
                         cl_inode_init(inode, md);

+                        cfs_spin_lock(&lli->lli_lock);

                         lli->lli_smd = lsm;
                         lli->lli_smd = lsm;

+                        cfs_spin_unlock(&lli->lli_lock);

                         cfs_up(&lli->lli_och_sem);
                         lli->lli_maxbytes = lsm->lsm_maxbytes;
                         if (lli->lli_maxbytes > PAGE_CACHE_MAXBYTES)
                         cfs_up(&lli->lli_och_sem);
                         lli->lli_maxbytes = lsm->lsm_maxbytes;
                         if (lli->lli_maxbytes > PAGE_CACHE_MAXBYTES)


@@ -2129,6 +2132,21 @@ struct md_op_data * ll_prep_md_op_data(struct md_op_data *op_data,
         op_data->op_mds = 0;
         op_data->op_data = data;
 
         op_data->op_mds = 0;
         op_data->op_data = data;
 

+        /* If the file is being opened after mknod() (normally due to NFS)
+         * try to use the default stripe data from parent directory for
+         * allocating OST objects.  Try to pass the parent FID to MDS. */
+        if (opc == LUSTRE_OPC_CREATE && i1 == i2 && S_ISREG(i2->i_mode) &&
+            ll_i2info(i2)->lli_smd == NULL) {
+                struct ll_inode_info *lli = ll_i2info(i2);
+
+                cfs_spin_lock(&lli->lli_lock);
+                if (likely(lli->lli_smd == NULL &&
+                           !fid_is_zero(&lli->lli_pfid)))
+                        op_data->op_fid1 = lli->lli_pfid;
+                cfs_spin_unlock(&lli->lli_lock);
+                /** We ignore parent's capability temporary. */
+        }
+

         return op_data;
 }
 
         return op_data;
 }
 

diff --git a/lustre/llite/llite_nfs.c b/lustre/llite/llite_nfs.c
index ffb8a07..c3e327c 100644 (file)
--- a/lustre/llite/llite_nfs.c
+++ b/lustre/llite/llite_nfs.c
@@ -114,8 +114,13 @@ static struct inode *search_inode_for_lustre(struct super_block *sb,
         RETURN(inode);
 }
 
         RETURN(inode);
 }
 

-static struct dentry *ll_iget_for_nfs(struct super_block *sb,
-                                      const struct lu_fid *fid)
+struct lustre_nfs_fid {
+        struct lu_fid   lnf_child;
+        struct lu_fid   lnf_parent;
+};
+
+static struct dentry *
+ll_iget_for_nfs(struct super_block *sb, struct lu_fid *fid, struct lu_fid *parent)

 {
         struct inode  *inode;
         struct dentry *result;
 {
         struct inode  *inode;
         struct dentry *result;


@@ -131,15 +136,26 @@ static struct dentry *ll_iget_for_nfs(struct super_block *sb,
 
         if (is_bad_inode(inode)) {
                 /* we didn't find the right inode.. */
 
         if (is_bad_inode(inode)) {
                 /* we didn't find the right inode.. */

-                CERROR("can't get inode by fid "DFID"\n",
-                       PFID(fid));

                 iput(inode);
                 RETURN(ERR_PTR(-ESTALE));
         }
 
                 iput(inode);
                 RETURN(ERR_PTR(-ESTALE));
         }
 

+        /**
+         * It is an anonymous dentry without OST objects created yet.
+         * We have to find the parent to tell MDS how to init lov objects.
+         */
+        if (S_ISREG(inode->i_mode) && ll_i2info(inode)->lli_smd == NULL &&
+            parent != NULL) {
+                struct ll_inode_info *lli = ll_i2info(inode);
+
+                cfs_spin_lock(&lli->lli_lock);
+                lli->lli_pfid = *parent;
+                cfs_spin_unlock(&lli->lli_lock);
+        }
+

         result = d_obtain_alias(inode);
         result = d_obtain_alias(inode);

-        if (!result)
-                RETURN(ERR_PTR(-ENOMEM));
+        if (IS_ERR(result))
+                RETURN(result);

 
         ll_dops_init(result, 1, 0);
 
 
         ll_dops_init(result, 1, 0);
 


@@ -148,11 +164,6 @@ static struct dentry *ll_iget_for_nfs(struct super_block *sb,
 
 #define LUSTRE_NFS_FID          0x97
 
 
 #define LUSTRE_NFS_FID          0x97
 

-struct lustre_nfs_fid {
-        struct lu_fid   lnf_child;
-        struct lu_fid   lnf_parent;
-};
-

 /**
  * \a connectable - is nfsd will connect himself or this should be done
  *                  at lustre
 /**
  * \a connectable - is nfsd will connect himself or this should be done
  *                  at lustre


@@ -250,7 +261,7 @@ static struct dentry *ll_fh_to_dentry(struct super_block *sb, struct fid *fid,
         if (fh_type != LUSTRE_NFS_FID)
                 RETURN(ERR_PTR(-EPROTO));
 
         if (fh_type != LUSTRE_NFS_FID)
                 RETURN(ERR_PTR(-EPROTO));
 

-        RETURN(ll_iget_for_nfs(sb, &nfs_fid->lnf_child));
+        RETURN(ll_iget_for_nfs(sb, &nfs_fid->lnf_child, &nfs_fid->lnf_parent));

 }
 
 static struct dentry *ll_fh_to_parent(struct super_block *sb, struct fid *fid,
 }
 
 static struct dentry *ll_fh_to_parent(struct super_block *sb, struct fid *fid,


@@ -261,7 +272,7 @@ static struct dentry *ll_fh_to_parent(struct super_block *sb, struct fid *fid,
         if (fh_type != LUSTRE_NFS_FID)
                 RETURN(ERR_PTR(-EPROTO));
 
         if (fh_type != LUSTRE_NFS_FID)
                 RETURN(ERR_PTR(-EPROTO));
 

-        RETURN(ll_iget_for_nfs(sb, &nfs_fid->lnf_parent));
+        RETURN(ll_iget_for_nfs(sb, &nfs_fid->lnf_parent, NULL));

 }
 
 #else
 }
 
 #else


@@ -293,11 +304,10 @@ static struct dentry *ll_decode_fh(struct super_block *sb, __u32 *fh, int fh_len
 
 static struct dentry *ll_get_dentry(struct super_block *sb, void *data)
 {
 
 static struct dentry *ll_get_dentry(struct super_block *sb, void *data)
 {

-        struct lustre_nfs_fid *fid = data;
-        struct dentry      *entry;
+        struct dentry *entry;

         ENTRY;
 
         ENTRY;
 

-        entry = ll_iget_for_nfs(sb, &fid->lnf_child);
+        entry = ll_iget_for_nfs(sb, data, NULL);

         RETURN(entry);
 }
 #endif
         RETURN(entry);
 }
 #endif


@@ -338,7 +348,7 @@ static struct dentry *ll_get_parent(struct dentry *dchild)
         CDEBUG(D_INFO, "parent for "DFID" is "DFID"\n",
                 PFID(ll_inode2fid(dir)), PFID(&body->fid1));
 
         CDEBUG(D_INFO, "parent for "DFID" is "DFID"\n",
                 PFID(ll_inode2fid(dir)), PFID(&body->fid1));
 

-        result = ll_iget_for_nfs(dir->i_sb, &body->fid1);
+        result = ll_iget_for_nfs(dir->i_sb, &body->fid1, NULL);

 
         ptlrpc_req_finished(req);
         RETURN(result);
 
         ptlrpc_req_finished(req);
         RETURN(result);

diff --git a/lustre/mdd/mdd_lov.c b/lustre/mdd/mdd_lov.c
index c6bdb36..4d2fed6 100644 (file)
--- a/lustre/mdd/mdd_lov.c
+++ b/lustre/mdd/mdd_lov.c
@@ -387,7 +387,7 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd,
         struct obdo           *oa;
         struct lov_stripe_md  *lsm = NULL;
         const void            *eadata = spec->u.sp_ea.eadata;
         struct obdo           *oa;
         struct lov_stripe_md  *lsm = NULL;
         const void            *eadata = spec->u.sp_ea.eadata;

-        __u32                  create_flags = spec->sp_cr_flags;
+        __u64                  create_flags = spec->sp_cr_flags;

         struct obd_trans_info *oti = &mdd_env_info(env)->mti_oti;
         int                    rc = 0;
         ENTRY;
         struct obd_trans_info *oti = &mdd_env_info(env)->mti_oti;
         int                    rc = 0;
         ENTRY;

diff --git a/lustre/mdd/mdd_object.c b/lustre/mdd/mdd_object.c
index 4fe42ff..553e04a 100644 (file)
--- a/lustre/mdd/mdd_object.c
+++ b/lustre/mdd/mdd_object.c
@@ -638,6 +638,34 @@ static int __mdd_lmm_get(const struct lu_env *env,
         RETURN(rc);
 }
 
         RETURN(rc);
 }
 

+/* get the first parent fid from link EA */
+static int mdd_pfid_get(const struct lu_env *env,
+                        struct mdd_object *mdd_obj, struct md_attr *ma)
+{
+        struct lu_buf *buf;
+        struct link_ea_header *leh;
+        struct link_ea_entry *lee;
+        struct lu_fid *pfid = &ma->ma_pfid;
+        ENTRY;
+
+        if (ma->ma_valid & MA_PFID)
+                RETURN(0);
+
+        buf = mdd_links_get(env, mdd_obj);
+        if (IS_ERR(buf))
+                RETURN(PTR_ERR(buf));
+
+        leh = buf->lb_buf;
+        lee = (struct link_ea_entry *)(leh + 1);
+        memcpy(pfid, &lee->lee_parent_fid, sizeof(*pfid));
+        fid_be_to_cpu(pfid, pfid);
+        ma->ma_valid |= MA_PFID;
+        if (buf->lb_len > OBD_ALLOC_BIG)
+                /* if we vmalloced a large buffer drop it */
+                mdd_buf_put(buf);
+        RETURN(0);
+}
+

 int mdd_lmm_get_locked(const struct lu_env *env, struct mdd_object *mdd_obj,
                        struct md_attr *ma)
 {
 int mdd_lmm_get_locked(const struct lu_env *env, struct mdd_object *mdd_obj,
                        struct md_attr *ma)
 {


@@ -734,6 +762,10 @@ int mdd_attr_get_internal(const struct lu_env *env, struct mdd_object *mdd_obj,
                     S_ISDIR(mdd_object_type(mdd_obj)))
                         rc = __mdd_lmm_get(env, mdd_obj, ma);
         }
                     S_ISDIR(mdd_object_type(mdd_obj)))
                         rc = __mdd_lmm_get(env, mdd_obj, ma);
         }

+        if (rc == 0 && ma->ma_need & MA_PFID && !(ma->ma_valid & MA_LOV)) {
+                if (S_ISREG(mdd_object_type(mdd_obj)))
+                        rc = mdd_pfid_get(env, mdd_obj, ma);
+        }

         if (rc == 0 && ma->ma_need & MA_LMV) {
                 if (S_ISDIR(mdd_object_type(mdd_obj)))
                         rc = __mdd_lmv_get(env, mdd_obj, ma);
         if (rc == 0 && ma->ma_need & MA_LMV) {
                 if (S_ISDIR(mdd_object_type(mdd_obj)))
                         rc = __mdd_lmv_get(env, mdd_obj, ma);


@@ -758,7 +790,7 @@ int mdd_attr_get_internal_locked(const struct lu_env *env,
 {
         int rc;
         int needlock = ma->ma_need &
 {
         int rc;
         int needlock = ma->ma_need &

-                       (MA_LOV | MA_LMV | MA_ACL_DEF | MA_HSM | MA_SOM);
+                       (MA_LOV | MA_LMV | MA_ACL_DEF | MA_HSM | MA_SOM | MA_PFID);

 
         if (needlock)
                 mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 
         if (needlock)
                 mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 094c6d2..ef2542b 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -4812,6 +4812,7 @@ static struct lu_object *mdt_object_alloc(const struct lu_env *env,
                 lu_object_add_top(h, o);
                 o->lo_ops = &mdt_obj_ops;
                 cfs_sema_init(&mo->mot_ioepoch_sem, 1);
                 lu_object_add_top(h, o);
                 o->lo_ops = &mdt_obj_ops;
                 cfs_sema_init(&mo->mot_ioepoch_sem, 1);

+                cfs_sema_init(&mo->mot_lov_sem, 1);

                 RETURN(o);
         } else
                 RETURN(NULL);
                 RETURN(o);
         } else
                 RETURN(NULL);

diff --git a/lustre/mdt/mdt_internal.h b/lustre/mdt/mdt_internal.h
index 7210e57..1df6c83 100644 (file)
--- a/lustre/mdt/mdt_internal.h
+++ b/lustre/mdt/mdt_internal.h
@@ -185,7 +185,9 @@ struct mdt_object {
         int                     mot_ioepoch_count;
         int                     mot_writecount;
         /* Lock to protect object's IO epoch. */
         int                     mot_ioepoch_count;
         int                     mot_writecount;
         /* Lock to protect object's IO epoch. */

-        cfs_semaphore_t        mot_ioepoch_sem;
+        cfs_semaphore_t         mot_ioepoch_sem;
+        /* Lock to protect create_data */
+        cfs_semaphore_t         mot_lov_sem;

 };
 
 enum mdt_object_flags {
 };
 
 enum mdt_object_flags {


@@ -204,6 +206,8 @@ enum mdt_object_flags {
         MOF_SOM_RECOV   = (1 << 1),
         /** File has been just created. */
         MOF_SOM_CREATED = (1 << 2),
         MOF_SOM_RECOV   = (1 << 1),
         /** File has been just created. */
         MOF_SOM_CREATED = (1 << 2),

+        /** lov object has been created. */
+        MOF_LOV_CREATED = (1 << 3),

 };
 
 struct mdt_lock_handle {
 };
 
 struct mdt_lock_handle {

diff --git a/lustre/mdt/mdt_open.c b/lustre/mdt/mdt_open.c
index 3f1e526..47a7093 100644 (file)
--- a/lustre/mdt/mdt_open.c
+++ b/lustre/mdt/mdt_open.c
@@ -107,8 +107,8 @@ static int mdt_create_data(struct mdt_thread_info *info,
                            struct mdt_object *p, struct mdt_object *o)
 {
         struct md_op_spec     *spec = &info->mti_spec;
                            struct mdt_object *p, struct mdt_object *o)
 {
         struct md_op_spec     *spec = &info->mti_spec;

-        struct md_attr        *ma = &info->mti_attr;
-        int rc;
+        struct md_attr        *ma   = &info->mti_attr;
+        int                    rc   = 0;

         ENTRY;
 
         if (!md_should_create(spec->sp_cr_flags))
         ENTRY;
 
         if (!md_should_create(spec->sp_cr_flags))


@@ -116,9 +116,15 @@ static int mdt_create_data(struct mdt_thread_info *info,
 
         ma->ma_need = MA_INODE | MA_LOV;
         ma->ma_valid = 0;
 
         ma->ma_need = MA_INODE | MA_LOV;
         ma->ma_valid = 0;

-        rc = mdo_create_data(info->mti_env,
-                             p ? mdt_object_child(p) : NULL,
-                             mdt_object_child(o), spec, ma);
+        cfs_down(&o->mot_lov_sem);
+        if (!(o->mot_flags & MOF_LOV_CREATED)) {
+                rc = mdo_create_data(info->mti_env,
+                                     p ? mdt_object_child(p) : NULL,
+                                     mdt_object_child(o), spec, ma);
+                if (rc == 0 && ma->ma_valid & MA_LOV)
+                        o->mot_flags |= MOF_LOV_CREATED;
+        }
+        cfs_up(&o->mot_lov_sem);

         RETURN(rc);
 }
 
         RETURN(rc);
 }
 


@@ -551,10 +557,8 @@ static void mdt_empty_transno(struct mdt_thread_info* info)
 
         ENTRY;
         /* transaction has occurred already */
 
         ENTRY;
         /* transaction has occurred already */

-        if (lustre_msg_get_transno(req->rq_repmsg) != 0) {
-                EXIT;
-                return;
-        }
+        if (lustre_msg_get_transno(req->rq_repmsg) != 0)
+                RETURN_EXIT;

 
         cfs_spin_lock(&mdt->mdt_lut.lut_translock);
         if (info->mti_transno == 0) {
 
         cfs_spin_lock(&mdt->mdt_lut.lut_translock);
         if (info->mti_transno == 0) {


@@ -930,12 +934,11 @@ void mdt_reconstruct_open(struct mdt_thread_info *info,
                         rc = PTR_ERR(parent);
                         LCONSOLE_WARN("Parent "DFID" lookup error %d."
                                       " Evicting client %s with export %s.\n",
                         rc = PTR_ERR(parent);
                         LCONSOLE_WARN("Parent "DFID" lookup error %d."
                                       " Evicting client %s with export %s.\n",

-                                      PFID(mdt_object_fid(parent)), rc,
+                                      PFID(rr->rr_fid1), rc,

                                       obd_uuid2str(&exp->exp_client_uuid),
                                       obd_export_nid2str(exp));
                         mdt_export_evict(exp);
                                       obd_uuid2str(&exp->exp_client_uuid),
                                       obd_export_nid2str(exp));
                         mdt_export_evict(exp);

-                        EXIT;
-                        return;
+                        RETURN_EXIT;

                 }
                 child = mdt_object_find(env, mdt, rr->rr_fid2);
                 if (IS_ERR(child)) {
                 }
                 child = mdt_object_find(env, mdt, rr->rr_fid2);
                 if (IS_ERR(child)) {


@@ -947,8 +950,7 @@ void mdt_reconstruct_open(struct mdt_thread_info *info,
                                       obd_export_nid2str(exp));
                         mdt_object_put(env, parent);
                         mdt_export_evict(exp);
                                       obd_export_nid2str(exp));
                         mdt_object_put(env, parent);
                         mdt_export_evict(exp);

-                        EXIT;
-                        return;
+                        RETURN_EXIT;

                 }
                 rc = mdt_object_exists(child);
                 if (rc > 0) {
                 }
                 rc = mdt_object_exists(child);
                 if (rc > 0) {


@@ -1026,19 +1028,37 @@ static int mdt_open_by_fid(struct mdt_thread_info* info,
         RETURN(rc);
 }
 
         RETURN(rc);
 }
 

-static int mdt_open_anon_by_fid(struct mdt_thread_info* info,
+static int mdt_open_anon_by_fid(struct mdt_thread_info *info,

                                 struct ldlm_reply *rep, 
                                 struct mdt_lock_handle *lhc)
 {
                                 struct ldlm_reply *rep, 
                                 struct mdt_lock_handle *lhc)
 {

+        const struct lu_env     *env   = info->mti_env;
+        struct mdt_device       *mdt   = info->mti_mdt;

         __u32                    flags = info->mti_spec.sp_cr_flags;
         __u32                    flags = info->mti_spec.sp_cr_flags;

-        struct mdt_reint_record *rr = &info->mti_rr;
-        struct md_attr          *ma = &info->mti_attr;
+        struct mdt_reint_record *rr    = &info->mti_rr;
+        struct md_attr          *ma    = &info->mti_attr;
+        struct mdt_object       *parent= NULL;

         struct mdt_object       *o;
         int                      rc;
         ldlm_mode_t              lm;
         ENTRY;
 
         struct mdt_object       *o;
         int                      rc;
         ldlm_mode_t              lm;
         ENTRY;
 

-        o = mdt_object_find(info->mti_env, info->mti_mdt, rr->rr_fid2);
+        if (md_should_create(flags)) {
+                if (!lu_fid_eq(rr->rr_fid1, rr->rr_fid2)) {
+                        parent = mdt_object_find(env, mdt, rr->rr_fid1);
+                        if (IS_ERR(parent)) {
+                                CDEBUG(D_INODE, "Fail to find parent "DFID
+                                       " for anonymous created %ld, try to"
+                                       " use server-side parent.\n",
+                                       PFID(rr->rr_fid1), PTR_ERR(parent));
+                                parent = NULL;
+                        }
+                }
+                if (parent == NULL)
+                        ma->ma_need |= MA_PFID;
+        }
+
+        o = mdt_object_find(env, mdt, rr->rr_fid2);

         if (IS_ERR(o))
                 RETURN(rc = PTR_ERR(o));
 
         if (IS_ERR(o))
                 RETURN(rc = PTR_ERR(o));
 


@@ -1051,7 +1071,6 @@ static int mdt_open_anon_by_fid(struct mdt_thread_info* info,
                 CERROR("NFS remote open shouldn't happen.\n");
                 GOTO(out, rc);
         }
                 CERROR("NFS remote open shouldn't happen.\n");
                 GOTO(out, rc);
         }

-

         mdt_set_disposition(info, rep, (DISP_IT_EXECD |
                                         DISP_LOOKUP_EXECD |
                                         DISP_LOOKUP_POS));
         mdt_set_disposition(info, rep, (DISP_IT_EXECD |
                                         DISP_LOOKUP_EXECD |
                                         DISP_LOOKUP_POS));


@@ -1071,20 +1090,33 @@ static int mdt_open_anon_by_fid(struct mdt_thread_info* info,
         if (rc)
                 GOTO(out, rc);
 
         if (rc)
                 GOTO(out, rc);
 

-        rc = mo_attr_get(info->mti_env, mdt_object_child(o), ma);
+        rc = mo_attr_get(env, mdt_object_child(o), ma);

         if (rc)
                 GOTO(out, rc);
 
         if (rc)
                 GOTO(out, rc);
 

+        if (ma->ma_valid & MA_PFID) {
+                parent = mdt_object_find(env, mdt, &ma->ma_pfid);
+                if (IS_ERR(parent)) {
+                        CDEBUG(D_INODE, "Fail to find parent "DFID
+                               " for anonymous created %ld, try to"
+                               " use system default.\n",
+                               PFID(&ma->ma_pfid), PTR_ERR(parent));
+                        parent = NULL;
+                }
+        }
+

         if (flags & MDS_OPEN_LOCK)
                 mdt_set_disposition(info, rep, DISP_OPEN_LOCK);
         if (flags & MDS_OPEN_LOCK)
                 mdt_set_disposition(info, rep, DISP_OPEN_LOCK);

-        rc = mdt_finish_open(info, NULL, o, flags, 0, rep);
+        rc = mdt_finish_open(info, parent, o, flags, 0, rep);

 
         if (!(flags & MDS_OPEN_LOCK) || rc)
                 mdt_object_unlock(info, o, lhc, 1);
 
         GOTO(out, rc);
 out:
 
         if (!(flags & MDS_OPEN_LOCK) || rc)
                 mdt_object_unlock(info, o, lhc, 1);
 
         GOTO(out, rc);
 out:

-        mdt_object_put(info->mti_env, o);
+        mdt_object_put(env, o);
+        if (parent != NULL)
+                mdt_object_put(env, parent);

         return rc;
 }
 
         return rc;
 }