- remove wrong assert from mdt_cross_open

- add mdc_attr_get() to retrieve attr for remote object while rename
- cmm changes to prepare src object type before processing with rename
- mdd don't use mdd_object for src while rename, because it can be remote
  object, instead of that it uses its fid and type to do operations and checks

diff --git a/lustre/cmm/cmm_object.c b/lustre/cmm/cmm_object.c
index ccbc23b..420a86f 100644 (file)
--- a/lustre/cmm/cmm_object.c
+++ b/lustre/cmm/cmm_object.c
@@ -407,14 +407,47 @@ static int cml_unlink(const struct lu_context *ctx, struct md_object *mo_p,
 }
 
 /* rename is split to local/remote by location of new parent dir */
+struct md_object *md_object_find(const struct lu_context *ctx,
+                                  struct md_device *md,
+                                  const struct lu_fid *f)
+{
+        struct lu_object *o;
+        struct md_object *m;
+        ENTRY;
+
+        o = lu_object_find(ctx, md2lu_dev(md)->ld_site, f);
+        if (IS_ERR(o))
+                m = (struct md_object *)o;
+        else {
+                o = lu_object_locate(o->lo_header, md2lu_dev(md)->ld_type);
+                m = o ? lu2md(o) : NULL;
+        }
+        RETURN(m);
+}
+
 static int cml_rename(const struct lu_context *ctx, struct md_object *mo_po,
                        struct md_object *mo_pn, const struct lu_fid *lf,
                        const char *s_name, struct md_object *mo_t,
                        const char *t_name, struct md_attr *ma)
 {
+        struct cmm_thread_info *cmi;
+        struct md_object *mo_s = md_object_find(ctx, md_obj2dev(mo_po), lf);
+        struct md_attr *tmp_ma;
         int rc;
         ENTRY;
 
+        cmi = lu_context_key_get(ctx, &cmm_thread_key);
+        LASSERT(cmi);
+        tmp_ma = &cmi->cmi_ma;
+        tmp_ma->ma_need = MA_INODE;
+        
+        /* get type from src, can be remote req */
+        rc = mo_attr_get(ctx, md_object_next(mo_s), tmp_ma);
+        if (rc != 0)
+                RETURN(rc);
+
+        ma->ma_attr.la_mode = tmp_ma->ma_attr.la_mode;
+
         if (mo_t && lu_object_exists(&mo_t->mo_lu) < 0) {
                 /* mo_t is remote object and there is RPC to unlink it */
                 rc = mo_ref_del(ctx, md_object_next(mo_t), ma);

diff --git a/lustre/cmm/mdc_object.c b/lustre/cmm/mdc_object.c
index 2d0170a..cc87e7d 100644 (file)
--- a/lustre/cmm/mdc_object.c
+++ b/lustre/cmm/mdc_object.c
@@ -159,6 +159,34 @@ static void mdc_req2attr_update(const struct lu_context *ctx,
         mdc_body2attr(body, ma);
 }
 
+static int mdc_attr_get(const struct lu_context *ctx, struct md_object *mo,
+                        struct md_attr *ma)
+{
+        struct mdc_device *mc = md2mdc_dev(md_obj2dev(mo));
+        struct mdc_thread_info *mci;
+        int rc;
+        ENTRY;
+
+        mci = lu_context_key_get(ctx, &mdc_thread_key);
+        LASSERT(mci);
+
+        memset(&mci->mci_opdata, 0, sizeof(mci->mci_opdata));
+
+        rc = md_getattr(mc->mc_desc.cl_exp, lu_object_fid(&mo->mo_lu),
+                        OBD_MD_FLMODE | OBD_MD_FLUID | OBD_MD_FLGID,
+                        0, &mci->mci_req);
+
+        if (rc == 0) {
+                /* get attr from request */
+                mdc_req2attr_update(ctx, ma);
+        }
+
+        ptlrpc_req_finished(mci->mci_req);
+
+        RETURN(rc);
+}
+
+
 static int mdc_object_create(const struct lu_context *ctx,
                              struct md_object *mo, 
                              const struct md_create_spec *spec,
@@ -311,6 +339,7 @@ send_page:
 #endif
 
 static struct md_object_operations mdc_mo_ops = {
+        .moo_attr_get       = mdc_attr_get,
         .moo_object_create  = mdc_object_create,
         .moo_ref_add        = mdc_ref_add,
         .moo_ref_del        = mdc_ref_del,

diff --git a/lustre/mdd/mdd_handler.c b/lustre/mdd/mdd_handler.c
index 2a1d0cc..651574e 100644 (file)
--- a/lustre/mdd/mdd_handler.c
+++ b/lustre/mdd/mdd_handler.c
@@ -1352,13 +1352,13 @@ static int mdd_parent_fid(const struct lu_context *ctxt,
 }
 
 /*
- * return 0: if p2 is the parent of p1
+ * return 0: if lf is the fid of the ancestor of p1
  * otherwise: other_value
  */
 static int mdd_is_parent(const struct lu_context *ctxt,
                          struct mdd_device *mdd,
                          struct mdd_object *p1,
-                         struct mdd_object *p2)
+                         const struct lu_fid *lf)
 {
         struct lu_fid * pfid;
         struct mdd_object *parent = NULL;
@@ -1372,7 +1372,7 @@ static int mdd_is_parent(const struct lu_context *ctxt,
                 rc = mdd_parent_fid(ctxt, p1, pfid);
                 if (rc)
                         GOTO(out, rc);
-                if (lu_fid_eq(pfid, mdo2fid(p2)))
+                if (lu_fid_eq(pfid, lf))
                         GOTO(out, rc = 0);
                 if (lu_fid_eq(pfid, &mdd->mdd_root_fid))
                         GOTO(out, rc = 1);
@@ -1412,7 +1412,7 @@ static int mdd_rename_lock(const struct lu_context *ctxt,
                 RETURN(0);
         }
 
-        if (!mdd_is_parent(ctxt, mdd, src_pobj, tgt_pobj)) {
+        if (!mdd_is_parent(ctxt, mdd, src_pobj, mdo2fid(tgt_pobj))) {
                 mdd_lock2(ctxt, tgt_pobj, src_pobj);
                 RETURN(0);
         }
@@ -1434,17 +1434,16 @@ static void mdd_rename_unlock(const struct lu_context *ctxt,
 static int mdd_rename_sanity_check(const struct lu_context *ctxt,
                                    struct mdd_object *src_pobj,
                                    struct mdd_object *tgt_pobj,
-                                   struct mdd_object *sobj,
+                                   const struct lu_fid *sfid,
+                                   int src_is_dir,
                                    struct mdd_object *tobj)
 {
         struct mdd_device *mdd =mdo2mdd(&src_pobj->mod_obj);
-        int rc = 0, src_is_dir, tgt_is_dir;
+        int rc = 0, tgt_is_dir;
         ENTRY;
 
-        src_is_dir = S_ISDIR(mdd_object_type(sobj));
-        rc = mdd_may_delete(ctxt, src_pobj, sobj, src_is_dir);
-        if (rc)
-                GOTO(out, rc);
+        if (mdd_is_dead_obj(src_pobj))
+                RETURN(-ENOENT);
 
         if (!tobj) {
                 rc = mdd_may_create(ctxt, tgt_pobj, NULL);
@@ -1457,17 +1456,15 @@ static int mdd_rename_sanity_check(const struct lu_context *ctxt,
                 }
         }
         if (rc)
-                GOTO(out, rc);
+                RETURN(rc);
 
         /* source should not be ancestor of target dir */
-        if (src_is_dir && !mdd_is_parent(ctxt, mdd, tgt_pobj, sobj))
+        if (src_is_dir && !mdd_is_parent(ctxt, mdd, tgt_pobj, sfid))
                 rc = -EINVAL;
 
-out:
-        mdd_object_put(ctxt, sobj);
         RETURN(rc);
 }
-
+/* src object can be remote that is why we use only fid and type of object */
 static int mdd_rename(const struct lu_context *ctxt, struct md_object *src_pobj,
                       struct md_object *tgt_pobj, const struct lu_fid *lf,
                       const char *sname, struct md_object *tobj,
@@ -1484,18 +1481,15 @@ static int mdd_rename(const struct lu_context *ctxt, struct md_object *src_pobj,
         int rc;
         ENTRY;
 
-        /* the source can be remote one, not supported yet */
-        if (mdd_sobj == NULL)
-                RETURN(-EOPNOTSUPP);
-
-        is_dir = S_ISDIR(mdd_object_type(mdd_sobj));
+        LASSERT(ma->ma_attr.la_mode & S_IFMT);
+        is_dir = S_ISDIR(ma->ma_attr.la_mode);
         
         if (tobj)
                 mdd_tobj = md2mdd_obj(tobj);
 
         /*XXX: shouldn't this check be done under lock below? */
         rc = mdd_rename_sanity_check(ctxt, mdd_spobj, mdd_tpobj,
-                                     mdd_sobj, mdd_tobj);
+                                     lf, is_dir, mdd_tobj);
         if (rc)
                 RETURN(rc);
 
@@ -1529,10 +1523,12 @@ static int mdd_rename(const struct lu_context *ctxt, struct md_object *src_pobj,
 
         *la_copy = ma->ma_attr;
         la_copy->la_valid = LA_CTIME;
-        rc = mdd_attr_set_internal_locked(ctxt, mdd_sobj, la_copy, handle);
-        if (rc)
-                GOTO(cleanup, rc);
-
+        if (mdd_sobj) {
+                /*XXX: how to update ctime for remote sobj? */
+                rc = mdd_attr_set_internal_locked(ctxt, mdd_sobj, la_copy, handle);
+                if (rc)
+                        GOTO(cleanup, rc);
+        }
         if (tobj && lu_object_exists(&tobj->mo_lu)) {
                 mdd_write_lock(ctxt, mdd_tobj);
                 __mdd_ref_del(ctxt, mdd_tobj, handle);

diff --git a/lustre/mdt/mdt_open.c b/lustre/mdt/mdt_open.c
index f45d1a4..088fcb9 100644 (file)
--- a/lustre/mdt/mdt_open.c
+++ b/lustre/mdt/mdt_open.c
@@ -661,8 +661,6 @@ int mdt_cross_open(struct mdt_thread_info* info, const struct lu_fid *fid,
         int                rc;
         ENTRY;
 
-        LASSERTF(!(flags & MDS_OPEN_CREAT), "Cross-ref open+create!\n");
-
         o = mdt_object_find(info->mti_ctxt, info->mti_mdt, fid);
         if (IS_ERR(o)) 
                 RETURN(rc = PTR_ERR(o));