LU-2430 mdt: Add global rename lock.

[fs/lustre-release.git] / lustre / lod / lod_object.c

diff --git a/lustre/lod/lod_object.c b/lustre/lod/lod_object.c
index ef9781d..02c14c2 100644 (file)
--- a/lustre/lod/lod_object.c
+++ b/lustre/lod/lod_object.c
@@ -629,6 +629,9 @@ int lod_parse_dir_striping(const struct lu_env *env, struct lod_object *lo,
        int                     rc = 0;
        ENTRY;
 
+       if (le32_to_cpu(lmv1->lmv_magic) == LMV_MAGIC_MIGRATE)
+               RETURN(0);
+
        if (le32_to_cpu(lmv1->lmv_magic) != LMV_MAGIC_V1)
                RETURN(-EINVAL);
 
@@ -861,7 +864,8 @@ static int lod_prep_md_striped_create(const struct lu_env *env,
                        GOTO(out_put, rc);
        }
 
-       rc = dt_declare_xattr_set(env, dt, &lmv_buf, XATTR_NAME_LMV, 0, th);
+       rc = dt_declare_xattr_set(env, dt_object_child(dt), &lmv_buf,
+                                 XATTR_NAME_LMV, 0, th);
        if (rc != 0)
                GOTO(out_put, rc);
 
@@ -1203,7 +1207,8 @@ static int lod_xattr_set_lmv(const struct lu_env *env, struct dt_object *dt,
                                  capa);
        }
 
-       rc = dt_xattr_set(env, dt, &lmv_buf, XATTR_NAME_LMV, fl, th, capa);
+       rc = dt_xattr_set(env, dt_object_child(dt), &lmv_buf, XATTR_NAME_LMV,
+                         fl, th, capa);
 
        RETURN(rc);
 }
@@ -1352,6 +1357,11 @@ static int lod_cache_parent_lov_striping(const struct lu_env *env,
        if (v1->lmm_pattern != LOV_PATTERN_RAID0 && v1->lmm_pattern != 0)
                GOTO(unlock, rc = 0);
 
+       CDEBUG(D_INFO, DFID" stripe_count=%d stripe_size=%d stripe_offset=%d\n",
+              PFID(lu_object_fid(&lp->ldo_obj.do_lu)),
+              (int)v1->lmm_stripe_count,
+              (int)v1->lmm_stripe_size, (int)v1->lmm_stripe_offset);
+
        lp->ldo_def_stripenr = v1->lmm_stripe_count;
        lp->ldo_def_stripe_size = v1->lmm_stripe_size;
        lp->ldo_def_stripe_offset = v1->lmm_stripe_offset;
@@ -1681,22 +1691,33 @@ int lod_declare_striped_object(const struct lu_env *env, struct dt_object *dt,
                GOTO(out, rc = -ENOMEM);
        }
 
-       /* choose OST and generate appropriate objects */
-       rc = lod_qos_prep_create(env, lo, attr, lovea, th);
-       if (rc) {
-               /* failed to create striping, let's reset
-                * config so that others don't get confused */
-               lod_object_free_striping(env, lo);
-               GOTO(out, rc);
-       }
+       if (!dt_object_remote(next)) {
+               /* choose OST and generate appropriate objects */
+               rc = lod_qos_prep_create(env, lo, attr, lovea, th);
+               if (rc) {
+                       /* failed to create striping, let's reset
+                        * config so that others don't get confused */
+                       lod_object_free_striping(env, lo);
+                       GOTO(out, rc);
+               }
 
-       /*
-        * declare storage for striping data
-        */
-       info->lti_buf.lb_len = lov_mds_md_size(lo->ldo_stripenr,
+               /*
+                * declare storage for striping data
+                */
+               info->lti_buf.lb_len = lov_mds_md_size(lo->ldo_stripenr,
                                lo->ldo_pool ?  LOV_MAGIC_V3 : LOV_MAGIC_V1);
-       rc = dt_declare_xattr_set(env, next, &info->lti_buf, XATTR_NAME_LOV,
-                                 0, th);
+       } else {
+               /* LOD can not choose OST objects for remote objects, i.e.
+                * stripes must be ready before that. Right now, it can only
+                * happen during migrate, i.e. migrate process needs to create
+                * remote regular file (mdd_migrate_create), then the migrate
+                * process will provide stripeEA. */
+               LASSERT(lovea != NULL);
+               info->lti_buf = *lovea;
+       }
+
+       rc = dt_declare_xattr_set(env, next, &info->lti_buf,
+                                 XATTR_NAME_LOV, 0, th);
        if (rc)
                GOTO(out, rc);
 
@@ -1892,7 +1913,11 @@ static int lod_declare_object_create(const struct lu_env *env,
                        rc = lod_declare_striped_object(env, dt, attr,
                                                        NULL, th);
        } else if (dof->dof_type == DFT_DIR) {
-               rc = lod_declare_dir_striping_create(env, dt, attr, dof, th);
+               /* Orphan object (like migrating object) does not have
+                * lod_dir_stripe, see lod_ah_init */
+               if (lo->ldo_dir_stripe != NULL)
+                       rc = lod_declare_dir_striping_create(env, dt, attr,
+                                                            dof, th);
        }
 out:
        RETURN(rc);
@@ -1936,7 +1961,8 @@ static int lod_object_create(const struct lu_env *env, struct dt_object *dt,
        rc = dt_create(env, next, attr, hint, dof, th);
 
        if (rc == 0) {
-               if (S_ISDIR(dt->do_lu.lo_header->loh_attr))
+               if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
+                   lo->ldo_dir_stripe != NULL)
                        rc = lod_dir_striping_create(env, dt, attr, dof, th);
                else if (lo->ldo_stripe && dof->u.dof_reg.striped != 0)
                        rc = lod_striping_create(env, dt, attr, dof, th);
@@ -2180,6 +2206,12 @@ static int lod_object_lock(const struct lu_env *env,
        /* striped directory lock */
        for (i = 0; i < lo->ldo_stripenr; i++) {
                struct lustre_handle    lockh;
+               struct ldlm_res_id      *res_id;
+
+               res_id = &lod_env_info(env)->lti_res_id;
+               fid_build_reg_res_name(lu_object_fid(&lo->ldo_stripe[i]->do_lu),
+                                      res_id);
+               einfo->ei_res_id = res_id;
 
                LASSERT(lo->ldo_stripe[i]);
                rc = dt_object_lock(env, lo->ldo_stripe[i], &lockh, einfo,
@@ -2244,11 +2276,11 @@ static ssize_t lod_read(const struct lu_env *env, struct dt_object *dt,
 
 static ssize_t lod_declare_write(const struct lu_env *env,
                                 struct dt_object *dt,
-                                const loff_t size, loff_t pos,
+                                const struct lu_buf *buf, loff_t pos,
                                 struct thandle *th)
 {
        return dt_declare_record_write(env, dt_object_child(dt),
-                                      size, pos, th);
+                                      buf, pos, th);
 }
 
 static ssize_t lod_write(const struct lu_env *env, struct dt_object *dt,