+ rc = mdd_la_get(env, mdd_sobj, so_attr, mdd_object_capa(env, mdd_sobj));
+ if (rc != 0)
+ RETURN(rc);
+
+ handle = mdd_trans_create(env, mdd);
+ if (IS_ERR(handle))
+ RETURN(PTR_ERR(handle));
+
+ rc = mdd_declare_migrate_update_name(env, mdd_pobj, mdd_sobj, mdd_tobj,
+ lname, so_attr, p_la, handle);
+ if (rc != 0) {
+ /* If the migration can not be fit in one transaction, just
+ * leave it in the original MDT */
+ if (rc == -E2BIG)
+ GOTO(stop_trans, rc = 0);
+ else
+ GOTO(stop_trans, rc);
+ }
+
+ CDEBUG(D_INFO, "%s: update "DFID"/"DFID" with %s:"DFID"\n",
+ mdd2obd_dev(mdd)->obd_name, PFID(mdd_object_fid(mdd_pobj)),
+ PFID(mdd_object_fid(mdd_sobj)), lname->ln_name,
+ PFID(mdd_object_fid(mdd_tobj)));
+
+ rc = mdd_trans_start(env, mdd, handle);
+ if (rc != 0)
+ GOTO(stop_trans, rc);
+
+ /* Revert IMMUTABLE flag */
+ la_flag->la_valid = LA_FLAGS;
+ la_flag->la_flags = so_attr->la_flags & ~LUSTRE_IMMUTABLE_FL;
+ mdd_flags_xlate(mdd_sobj, la_flag->la_flags);
+ rc = mdo_attr_set(env, mdd_sobj, la_flag, handle,
+ mdd_object_capa(env, mdd_pobj));
+ if (rc != 0)
+ GOTO(stop_trans, rc);
+
+ /* Remove source name from source directory */
+ rc = __mdd_index_delete(env, mdd_pobj, name, is_dir, handle,
+ mdd_object_capa(env, mdd_pobj));
+ if (rc != 0)
+ GOTO(stop_trans, rc);
+
+ rc = mdd_update_linkea(env, mdd_pobj, mdd_sobj, mdd_tobj,
+ handle, lname);
+ if (rc != 0)
+ GOTO(stop_trans, rc);
+
+ if (S_ISREG(so_attr->la_mode)) {
+ if (so_attr->la_nlink == 1) {
+ rc = mdo_xattr_del(env, mdd_sobj, XATTR_NAME_LOV,
+ handle,
+ mdd_object_capa(env, mdd_sobj));
+ if (rc != 0 && rc != -ENODATA)
+ GOTO(stop_trans, rc);
+ }
+ }
+
+ /* Insert new fid with target name into target dir */
+ rc = __mdd_index_insert(env, mdd_pobj, mdd_object_fid(mdd_tobj),
+ mdd_object_type(mdd_tobj), name,
+ handle, mdd_object_capa(env, mdd_pobj));
+ if (rc != 0)
+ GOTO(stop_trans, rc);
+
+ rc = mdd_links_add(env, mdd_tobj, mdo2fid(mdd_pobj), lname, handle,
+ NULL, 1);
+ if (rc != 0)
+ GOTO(stop_trans, rc);
+
+ mdd_write_lock(env, mdd_sobj, MOR_SRC_CHILD);
+ mdo_ref_del(env, mdd_sobj, handle);
+ if (is_dir)
+ mdo_ref_del(env, mdd_sobj, handle);
+
+ /* Get the attr again after ref_del */
+ rc = mdd_la_get(env, mdd_sobj, so_attr,
+ mdd_object_capa(env, mdd_sobj));
+ if (rc != 0)
+ GOTO(stop_trans, rc);
+ ma->ma_attr = *so_attr;
+ ma->ma_valid |= MA_INODE;
+ rc = mdd_finish_unlink(env, mdd_sobj, ma, mdd_pobj, lname, handle);
+ if (rc != 0)
+ GOTO(stop_trans, rc);
+
+ rc = mdd_attr_set_internal(env, mdd_pobj, p_la, handle, 0);
+ if (rc != 0)
+ GOTO(stop_trans, rc);
+
+ mdd_write_unlock(env, mdd_sobj);
+
+stop_trans:
+ mdd_trans_stop(env, mdd, rc, handle);
+
+ RETURN(rc);
+}
+
+/**
+ * Check whether we should migrate the file/dir
+ * return val
+ * < 0 permission check failed or other error.
+ * = 0 the file can be migrated.
+ * > 0 the file does not need to be migrated, mostly
+ * for multiple link file
+ **/
+static int mdd_migrate_sanity_check(const struct lu_env *env,
+ struct mdd_object *pobj,
+ const struct lu_attr *pattr,
+ struct mdd_object *sobj,
+ struct lu_attr *sattr)
+{
+ struct mdd_thread_info *info = mdd_env_info(env);
+ struct linkea_data *ldata = &info->mti_link_data;
+ int mgr_easize;
+ struct lu_buf *mgr_buf;
+ int count;
+ int rc;
+
+ ENTRY;
+
+ mgr_easize = lmv_mds_md_size(2, LMV_MAGIC_V1);
+ mgr_buf = lu_buf_check_and_alloc(&info->mti_big_buf, mgr_easize);
+ if (mgr_buf->lb_buf == NULL)
+ RETURN(-ENOMEM);
+
+ rc = mdo_xattr_get(env, sobj, mgr_buf, XATTR_NAME_LMV,
+ mdd_object_capa(env, sobj));
+ if (rc > 0) {
+ union lmv_mds_md *lmm = mgr_buf->lb_buf;
+
+ /* If the object has migrateEA, it means IMMUTE flag
+ * is being set by previous migration process, so it
+ * needs to override the IMMUTE flag, otherwise the
+ * following sanity check will fail */
+ if (le32_to_cpu(lmm->lmv_md_v1.lmv_hash_type) &
+ LMV_HASH_FLAG_MIGRATION) {
+ struct mdd_device *mdd = mdo2mdd(&sobj->mod_obj);
+
+ sattr->la_flags &= ~LUSTRE_IMMUTABLE_FL;
+ sobj->mod_flags &= ~IMMUTE_OBJ;
+ CDEBUG(D_HA, "%s: "DFID" override IMMUTE FLAG\n",
+ mdd2obd_dev(mdd)->obd_name,
+ PFID(mdd_object_fid(sobj)));
+ }
+ }
+
+ rc = mdd_rename_sanity_check(env, pobj, pattr, pobj, pattr,
+ sobj, sattr, NULL, NULL);
+ if (rc != 0)
+ RETURN(rc);
+
+ /* Then it will check if the file should be migrated. If the file
+ * has mulitple links, we only need migrate the file if all of its
+ * entries has been migrated to the remote MDT */
+ if (!S_ISREG(sattr->la_mode) || sattr->la_nlink < 2)
+ RETURN(0);
+
+ rc = mdd_links_read(env, sobj, ldata);
+ if (rc != 0) {
+ /* For multiple links files, if there are no linkEA data at all,
+ * means the file might be created before linkEA is enabled, and
+ * all all of its links should not be migrated yet, otherwise
+ * it should have some linkEA there */
+ if (rc == -ENOENT || rc == -ENODATA)
+ RETURN(1);
+ RETURN(rc);
+ }
+
+ /* If it is mulitple links file, we need update the name entry for
+ * all parent */
+ LASSERT(ldata->ld_leh != NULL);
+ ldata->ld_lee = (struct link_ea_entry *)(ldata->ld_leh + 1);
+ for (count = 0; count < ldata->ld_leh->leh_reccount; count++) {
+ struct mdd_device *mdd = mdo2mdd(&sobj->mod_obj);
+ struct mdd_object *lpobj;
+ struct lu_name lname;
+ struct lu_fid fid;
+
+ linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen,
+ &lname, &fid);
+ ldata->ld_lee = (struct link_ea_entry *)((char *)ldata->ld_lee +
+ ldata->ld_reclen);
+ lpobj = mdd_object_find(env, mdd, &fid);
+ if (IS_ERR(lpobj)) {
+ CWARN("%s: cannot find obj "DFID": rc = %ld\n",
+ mdd2obd_dev(mdd)->obd_name, PFID(&fid),
+ PTR_ERR(lpobj));
+ continue;
+ }
+
+ if (!mdd_object_exists(lpobj) || mdd_object_remote(lpobj)) {
+ CDEBUG(D_INFO, DFID"%.*s: is on remote MDT.\n",
+ PFID(&fid), lname.ln_namelen, lname.ln_name);
+ mdd_object_put(env, lpobj);
+ continue;
+ }
+
+ CDEBUG(D_INFO, DFID"still has local entry %.*s "DFID"\n",
+ PFID(mdd_object_fid(sobj)), lname.ln_namelen,
+ lname.ln_name, PFID(&fid));
+ mdd_object_put(env, lpobj);
+ rc = 1;
+ break;
+ }
+
+ RETURN(rc);
+}
+
+static int mdd_migrate(const struct lu_env *env, struct md_object *pobj,
+ struct md_object *sobj, const struct lu_name *lname,
+ struct md_object *tobj, struct md_attr *ma)
+{
+ struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
+ struct mdd_device *mdd = mdo2mdd(pobj);
+ struct mdd_object *mdd_sobj = md2mdd_obj(sobj);
+ struct mdd_object *mdd_tobj = NULL;
+ struct lu_attr *so_attr = MDD_ENV_VAR(env, cattr);
+ struct lu_attr *pattr = MDD_ENV_VAR(env, pattr);
+ int rc;
+
+ ENTRY;
+ /* If the file will being migrated, it will check whether
+ * the file is being opened by someone else right now */
+ mdd_read_lock(env, mdd_sobj, MOR_SRC_CHILD);
+ if (mdd_sobj->mod_count >= 1) {
+ CERROR("%s: "DFID"%s is already opened count %d: rc = %d\n",
+ mdd2obd_dev(mdd)->obd_name,
+ PFID(mdd_object_fid(mdd_sobj)), lname->ln_name,
+ mdd_sobj->mod_count, -EBUSY);
+ mdd_read_unlock(env, mdd_sobj);
+ GOTO(put, rc = -EBUSY);
+ }
+ mdd_read_unlock(env, mdd_sobj);
+
+ rc = mdd_la_get(env, mdd_sobj, so_attr, mdd_object_capa(env, mdd_sobj));
+ if (rc != 0)
+ GOTO(put, rc);
+
+ rc = mdd_la_get(env, mdd_pobj, pattr, BYPASS_CAPA);
+ if (rc != 0)
+ GOTO(put, rc);
+
+ rc = mdd_migrate_sanity_check(env, mdd_pobj, pattr, mdd_sobj, so_attr);
+ if (rc != 0) {
+ if (rc > 0)
+ rc = 0;
+ GOTO(put, rc);
+ }
+
+ /* Sigh, it is impossible to finish all of migration in a single
+ * transaction, for example migrating big directory entries to the
+ * new MDT, it needs insert all of name entries of children in the
+ * new directory.
+ *
+ * So migration will be done in multiple steps and transactions.
+ *
+ * 1. create an orphan object on the remote MDT in one transaction.
+ * 2. migrate extend attributes to the new target file/directory.
+ * 3. For directory, migrate the entries to the new MDT and update
+ * linkEA of each children. Because we can not migrate all entries
+ * in a single transaction, so the migrating directory will become
+ * a striped directory during migration, so once the process is
+ * interrupted, the directory is still accessible. (During lookup,
+ * client will locate the name by searching both original and target
+ * object).
+ * 4. Finally, update the name/FID to point to the new file/directory
+ * in a separate transaction.
+ */
+
+ /* step 1: Check whether the orphan object has been created, and create
+ * orphan object on the remote MDT if needed */
+ mdd_tobj = md2mdd_obj(tobj);
+ if (!mdd_object_exists(mdd_tobj)) {
+ rc = mdd_migrate_create(env, mdd_pobj, mdd_sobj, mdd_tobj,
+ so_attr);
+ if (rc != 0)
+ GOTO(put, rc);
+ }
+
+ /* step 2: migrate xattr */
+ rc = mdd_migrate_xattrs(env, mdd_sobj, mdd_tobj);
+ if (rc != 0)
+ GOTO(put, rc);
+
+ /* step 3: migrate name entries to the orphan object */
+ if (S_ISDIR(lu_object_attr(&mdd_sobj->mod_obj.mo_lu))) {
+ rc = mdd_migrate_entries(env, mdd_sobj, mdd_tobj);
+ if (rc != 0)
+ GOTO(put, rc);
+ if (unlikely(OBD_FAIL_CHECK_RESET(OBD_FAIL_MIGRATE_NET_REP,
+ OBD_FAIL_MDS_REINT_NET_REP)))
+ GOTO(put, rc = 0);
+ }
+
+ /* step 4: update name entry to the new object */
+ rc = mdd_migrate_update_name(env, mdd_pobj, mdd_sobj, mdd_tobj, lname,
+ ma);
+ if (rc != 0)
+ GOTO(put, rc);
+put:
+ RETURN(rc);