Whamcloud - gitweb
Branch HEAD
[fs/lustre-release.git] / lustre / mdt / mdt_handler.c
index 2212bff..c299486 100644 (file)
@@ -165,6 +165,8 @@ static struct mdt_opc_slice mdt_fld_handlers[];
 static struct mdt_device *mdt_dev(struct lu_device *d);
 static int mdt_regular_handle(struct ptlrpc_request *req);
 static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags);
+static int mdt_fid2path(const struct lu_env *env, struct mdt_device *mdt,
+                        struct getinfo_fid2path *fp);
 
 static const struct lu_object_operations mdt_obj_ops;
 
@@ -349,7 +351,7 @@ static int mdt_statfs(struct mdt_thread_info *info)
         /* This will trigger a watchdog timeout */
         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_STATFS_LCW_SLEEP,
                          (MDT_SERVICE_WATCHDOG_FACTOR *
-                          at_get(&svc->srv_at_estimate) / 1000) + 1);
+                          at_get(&svc->srv_at_estimate)) + 1);
 
         rc = mdt_check_ucred(info);
         if (rc)
@@ -1113,9 +1115,13 @@ static int mdt_set_info(struct mdt_thread_info *info)
                 spin_unlock(&req->rq_export->exp_lock);
 
         } else if (KEY_IS(KEY_CHANGELOG_CLEAR)) {
+                struct changelog_setinfo *cs =
+                        (struct changelog_setinfo *)val;
+                if (vallen != sizeof(*cs)) {
+                        CERROR("Bad changelog_clear setinfo size %d\n", vallen);
+                        RETURN(-EINVAL);
+                }
                 if (lustre_msg_swabbed(req->rq_reqmsg)) {
-                        struct changelog_setinfo *cs =
-                                (struct changelog_setinfo *)val;
                         __swab64s(&cs->cs_recno);
                         __swab32s(&cs->cs_id);
                 }
@@ -1509,6 +1515,8 @@ static int mdt_reint_internal(struct mdt_thread_info *info,
                 GOTO(out_shrink, rc = err_serious(rc));
         }
 
+        OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_REINT_DELAY, 10);
+
         /* for replay no cookkie / lmm need, because client have this already */
         if (info->mti_spec.no_create == 1)  {
                 if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER))
@@ -1757,6 +1765,13 @@ static int mdt_quotactl_handle(struct mdt_thread_info *info)
 
         switch (oqctl->qc_cmd) {
         case Q_QUOTAON:
+                if (info->mti_mdt->mdt_som_conf) {
+                        /* Quota cannot be used together with SOM while
+                         * SOM stored blocks in i_blocks but not in SOM EA. */
+                        LCONSOLE_ERROR("Fail to turn Quota on: SOM is enabled "
+                                       "and temporary conflicts with quota.\n");
+                        RETURN(-ENOTSUPP);
+                }
                 rc = mqo->mqo_on(info->mti_env, next, oqctl->qc_type);
                 break;
         case Q_QUOTAOFF:
@@ -2585,7 +2600,7 @@ static int mdt_req_handle(struct mdt_thread_info *info,
         }
 
         /* If we're DISCONNECTing, the mdt_export_data is already freed */
-        if (likely(rc == 0 && h->mh_opc != MDS_DISCONNECT))
+        if (likely(rc == 0 && req->rq_export && h->mh_opc != MDS_DISCONNECT))
                 target_committed_to_req(req);
 
         if (unlikely((lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) &&
@@ -2652,6 +2667,10 @@ static void mdt_thread_info_init(struct ptlrpc_request *req,
 
         info->mti_fail_id = OBD_FAIL_MDS_ALL_REPLY_NET;
         info->mti_transno = lustre_msg_get_transno(req->rq_reqmsg);
+        info->mti_mos[0] = NULL;
+        info->mti_mos[1] = NULL;
+        info->mti_mos[2] = NULL;
+        info->mti_mos[3] = NULL;
 
         memset(&info->mti_attr, 0, sizeof(info->mti_attr));
         info->mti_body = NULL;
@@ -2673,6 +2692,11 @@ static void mdt_thread_info_fini(struct mdt_thread_info *info)
 
         req_capsule_fini(info->mti_pill);
         if (info->mti_object != NULL) {
+                /*
+                 * freeing an object may lead to OSD level transaction, do not
+                 * let it mess with MDT. bz19385.
+                 */
+                info->mti_no_need_trans = 1;
                 mdt_object_put(info->mti_env, info->mti_object);
                 info->mti_object = NULL;
         }
@@ -2832,6 +2856,7 @@ static int mdt_msg_check_version(struct lustre_msg *msg)
         case MDS_GETXATTR:
         case MDS_SETXATTR:
         case MDS_SET_INFO:
+        case MDS_GET_INFO:
         case MDS_QUOTACHECK:
         case MDS_QUOTACTL:
         case QUOTA_DQACQ:
@@ -3354,14 +3379,15 @@ static int mdt_intent_reint(enum mdt_it_code opcode,
         }
         rep->lock_policy_res2 = clear_serious(rc);
 
-        lhc->mlh_reg_lh.cookie = 0ull;
-        if (rc == -ENOTCONN || rc == -ENODEV) {
+        if (rc == -ENOTCONN || rc == -ENODEV ||
+            rc == -EOVERFLOW) { /**< if VBR failure then return error */
                 /*
                  * If it is the disconnect error (ENODEV & ENOCONN), the error
                  * will be returned by rq_status, and client at ptlrpc layer
                  * will detect this, then disconnect, reconnect the import
                  * immediately, instead of impacting the following the rpc.
                  */
+                lhc->mlh_reg_lh.cookie = 0ull;
                 RETURN(rc);
         } else {
                 /*
@@ -3372,7 +3398,14 @@ static int mdt_intent_reint(enum mdt_it_code opcode,
                   * FIXME: when open lock is finished, that should be
                   * checked here.
                   */
-                RETURN(ELDLM_LOCK_ABORTED);
+                if (lustre_handle_is_used(&lhc->mlh_reg_lh)) {
+                        rep->lock_policy_res2 = 0;
+                        rc = mdt_intent_lock_replace(info, lockp, NULL, lhc, flags);
+                        RETURN(rc);
+                } else {
+                        lhc->mlh_reg_lh.cookie = 0ull;
+                        RETURN(ELDLM_LOCK_ABORTED);
+                }
         }
 }
 
@@ -4295,33 +4328,12 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m)
         struct lu_device  *d    = &m->mdt_md_dev.md_lu_dev;
         struct lu_site    *ls   = d->ld_site;
         struct obd_device *obd = mdt2obd_dev(m);
-        int                waited = 0;
         ENTRY;
 
-        /* At this point, obd exports might still be on the "obd_zombie_exports"
-         * list, and obd_zombie_impexp_thread() is trying to destroy them.
-         * We wait a little bit until all exports (except the self-export)
-         * have been destroyed, because the whole mdt stack might be accessed
-         * in mdt_destroy_export(). This will not be a long time, maybe one or
-         * two seconds are enough. This is not a problem while umounting.
-         *
-         * The three references that should be remaining are the
-         * obd_self_export and the attach and setup references.
-         */
-        while (atomic_read(&obd->obd_refcount) > 3) {
-                cfs_schedule_timeout(CFS_TASK_UNINT, cfs_time_seconds(1));
-                ++waited;
-                if (waited > 5 && IS_PO2(waited))
-                        LCONSOLE_WARN("Waiting for obd_zombie_impexp_thread "
-                                      "more than %d seconds to destroy all "
-                                      "the exports. The current obd refcount ="
-                                      " %d. Is it stuck there?\n",
-                                      waited, atomic_read(&obd->obd_refcount));
-        }
+        target_recovery_fini(obd);
 
         ping_evictor_stop();
 
-        target_recovery_fini(obd);
         mdt_stop_ptlrpc_service(m);
         mdt_llog_ctxt_unclone(env, m, LLOG_CHANGELOG_ORIG_CTXT);
         mdt_obd_llog_cleanup(obd);
@@ -4329,6 +4341,7 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m)
 #ifdef HAVE_QUOTA_SUPPORT
         next->md_ops->mdo_quota.mqo_cleanup(env, next);
 #endif
+        lut_fini(env, &m->mdt_lut);
         mdt_fs_cleanup(env, m);
         upcall_cache_cleanup(m->mdt_identity_cache);
         m->mdt_identity_cache = NULL;
@@ -4348,16 +4361,6 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m)
         mdt_seq_fini(env, m);
         mdt_seq_fini_cli(m);
         mdt_fld_fini(env, m);
-        mdt_procfs_fini(m);
-        if (obd->obd_proc_exports_entry) {
-                lprocfs_remove_proc_entry("clear", obd->obd_proc_exports_entry);
-                obd->obd_proc_exports_entry = NULL;
-        }
-        lprocfs_free_per_client_stats(obd);
-        lprocfs_free_obd_stats(obd);
-        ptlrpc_lprocfs_unregister_obd(d->ld_obd);
-        lprocfs_obd_cleanup(d->ld_obd);
-
         sptlrpc_rule_set_free(&m->mdt_sptlrpc_rset);
 
         next->md_ops->mdo_init_capa_ctxt(env, next, 0, 0, 0, NULL);
@@ -4369,6 +4372,16 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m)
          */
         mdt_stack_fini(env, m, md2lu_dev(m->mdt_child));
 
+        mdt_procfs_fini(m);
+        if (obd->obd_proc_exports_entry) {
+                lprocfs_remove_proc_entry("clear", obd->obd_proc_exports_entry);
+                obd->obd_proc_exports_entry = NULL;
+        }
+        lprocfs_free_per_client_stats(obd);
+        lprocfs_free_obd_stats(obd);
+        ptlrpc_lprocfs_unregister_obd(obd);
+        lprocfs_obd_cleanup(obd);
+
         if (ls) {
                 struct md_site *mite;
 
@@ -4495,6 +4508,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
 
         m->mdt_max_mdsize = MAX_MD_SIZE;
         m->mdt_max_cookiesize = sizeof(struct llog_cookie);
+        m->mdt_som_conf = 0;
 
         m->mdt_opts.mo_user_xattr = 0;
         m->mdt_opts.mo_acl = 0;
@@ -4597,10 +4611,14 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
                 GOTO(err_fini_proc, rc);
         }
 
-        rc = mdt_fld_init(env, obd->obd_name, m);
+        rc = lut_init(env, &m->mdt_lut, obd, m->mdt_bottom);
         if (rc)
                 GOTO(err_fini_stack, rc);
 
+        rc = mdt_fld_init(env, obd->obd_name, m);
+        if (rc)
+                GOTO(err_lut, rc);
+
         rc = mdt_seq_init(env, obd->obd_name, m);
         if (rc)
                 GOTO(err_fini_fld, rc);
@@ -4660,7 +4678,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
         server_put_mount_2(dev, lmi->lmi_mnt);
         lmi = NULL;
 
-        target_recovery_init(obd, mdt_recovery_handle);
+        target_recovery_init(&m->mdt_lut, mdt_recovery_handle);
 
         rc = mdt_start_ptlrpc_service(m);
         if (rc)
@@ -4710,6 +4728,8 @@ err_fini_seq:
         mdt_seq_fini(env, m);
 err_fini_fld:
         mdt_fld_fini(env, m);
+err_lut:
+        lut_fini(env, &m->mdt_lut);
 err_fini_stack:
         mdt_stack_fini(env, m, md2lu_dev(m->mdt_child));
 err_fini_proc:
@@ -4909,6 +4929,9 @@ static int mdt_connect_internal(struct obd_export *exp,
                 if (!mdt->mdt_opts.mo_user_xattr)
                         data->ocd_connect_flags &= ~OBD_CONNECT_XATTR;
 
+                if (!mdt->mdt_som_conf)
+                        data->ocd_connect_flags &= ~OBD_CONNECT_SOM;
+                
                 spin_lock(&exp->exp_lock);
                 exp->exp_connect_flags = data->ocd_connect_flags;
                 spin_unlock(&exp->exp_lock);
@@ -4931,6 +4954,14 @@ static int mdt_connect_internal(struct obd_export *exp,
                 return -EBADE;
         }
 
+        if (mdt->mdt_som_conf &&
+            !(exp->exp_connect_flags & OBD_CONNECT_MDS_MDS) &&
+            !(exp->exp_connect_flags & OBD_CONNECT_SOM)) {
+                CWARN("%s: MDS has SOM enabled, but client does not support "
+                      "it\n", mdt->mdt_md_dev.md_lu_dev.ld_obd->obd_name);
+                return -EBADE;
+        }
+
         return 0;
 }
 
@@ -5073,6 +5104,81 @@ static int mdt_obd_reconnect(const struct lu_env *env,
 
         RETURN(rc);
 }
+static int mdt_mfd_cleanup(struct obd_export *exp)
+{
+        struct mdt_export_data *med = &exp->exp_mdt_data;
+        struct obd_device      *obd = exp->exp_obd;
+        struct mdt_device      *mdt;
+        struct mdt_thread_info *info;
+        struct lu_env           env;
+        CFS_LIST_HEAD(closing_list);
+        struct mdt_file_data *mfd, *n;
+        int rc = 0;
+        ENTRY;
+
+        spin_lock(&med->med_open_lock);
+        while (!list_empty(&med->med_open_head)) {
+                struct list_head *tmp = med->med_open_head.next;
+                mfd = list_entry(tmp, struct mdt_file_data, mfd_list);
+
+                /* Remove mfd handle so it can't be found again.
+                 * We are consuming the mfd_list reference here. */
+                class_handle_unhash(&mfd->mfd_handle);
+                list_move_tail(&mfd->mfd_list, &closing_list);
+        }
+        spin_unlock(&med->med_open_lock);
+        mdt = mdt_dev(obd->obd_lu_dev);
+        LASSERT(mdt != NULL);
+
+        rc = lu_env_init(&env, LCT_MD_THREAD);
+        if (rc)
+                RETURN(rc);
+
+        info = lu_context_key_get(&env.le_ctx, &mdt_thread_key);
+        LASSERT(info != NULL);
+        memset(info, 0, sizeof *info);
+        info->mti_env = &env;
+        info->mti_mdt = mdt;
+        info->mti_exp = exp;
+
+        if (!list_empty(&closing_list)) {
+                struct md_attr *ma = &info->mti_attr;
+                int lmm_size;
+                int cookie_size;
+
+                lmm_size = mdt->mdt_max_mdsize;
+                cookie_size = mdt->mdt_max_cookiesize;
+                OBD_ALLOC(ma->ma_lmm, lmm_size);
+                if (ma->ma_lmm == NULL)
+                        GOTO(out_lmm, rc = -ENOMEM);
+                OBD_ALLOC(ma->ma_cookie, cookie_size);
+                if (ma->ma_cookie == NULL)
+                        GOTO(out_cookie, rc = -ENOMEM);
+
+                /* Close any open files (which may also cause orphan unlinking). */
+                list_for_each_entry_safe(mfd, n, &closing_list, mfd_list) {
+                        list_del_init(&mfd->mfd_list);
+                        /* TODO: if we close the unlinked file,
+                         * we need to remove its objects from OST */
+                        memset(&ma->ma_attr, 0, sizeof(ma->ma_attr));
+                        ma->ma_lmm_size = lmm_size;
+                        ma->ma_cookie_size = cookie_size;
+                        ma->ma_need = MA_LOV | MA_COOKIE;
+                        ma->ma_valid = 0;
+                        mdt_mfd_close(info, mfd);
+                }
+                info->mti_mdt = NULL;
+                OBD_FREE(ma->ma_cookie, cookie_size);
+                ma->ma_cookie = NULL;
+out_cookie:
+                OBD_FREE(ma->ma_lmm, lmm_size);
+                ma->ma_lmm = NULL;
+        }
+out_lmm:
+        lu_env_fini(&env);
+
+        RETURN(rc);
+}
 
 static int mdt_obd_disconnect(struct obd_export *exp)
 {
@@ -5107,7 +5213,7 @@ static int mdt_obd_disconnect(struct obd_export *exp)
                 spin_unlock(&svc->srv_lock);
         }
         spin_unlock(&exp->exp_lock);
-
+        rc = mdt_mfd_cleanup(exp);
         class_export_put(exp);
         RETURN(rc);
 }
@@ -5139,11 +5245,6 @@ static int mdt_destroy_export(struct obd_export *export)
         struct mdt_device      *mdt;
         struct mdt_thread_info *info;
         struct lu_env           env;
-        struct md_attr         *ma;
-        int lmm_size;
-        int cookie_size;
-        CFS_LIST_HEAD(closing_list);
-        struct mdt_file_data *mfd, *n;
         int rc = 0;
         ENTRY;
 
@@ -5154,6 +5255,8 @@ static int mdt_destroy_export(struct obd_export *export)
         target_destroy_export(export);
         ldlm_destroy_export(export);
 
+        LASSERT(list_empty(&export->exp_outstanding_replies));
+        LASSERT(list_empty(&med->med_open_head));
         if (obd_uuid_equals(&export->exp_client_uuid, &obd->obd_uuid))
                 RETURN(0);
 
@@ -5168,62 +5271,12 @@ static int mdt_destroy_export(struct obd_export *export)
         LASSERT(info != NULL);
         memset(info, 0, sizeof *info);
         info->mti_env = &env;
-        info->mti_mdt = mdt;
         info->mti_exp = export;
-
-        ma = &info->mti_attr;
-        lmm_size = ma->ma_lmm_size = mdt->mdt_max_mdsize;
-        cookie_size = ma->ma_cookie_size = mdt->mdt_max_cookiesize;
-        OBD_ALLOC(ma->ma_lmm, lmm_size);
-        OBD_ALLOC(ma->ma_cookie, cookie_size);
-
-        if (ma->ma_lmm == NULL || ma->ma_cookie == NULL)
-                GOTO(out, rc = -ENOMEM);
-        ma->ma_need = MA_LOV | MA_COOKIE;
-        ma->ma_valid = 0;
-        /* Close any open files (which may also cause orphan unlinking). */
-        spin_lock(&med->med_open_lock);
-        while (!list_empty(&med->med_open_head)) {
-                struct list_head *tmp = med->med_open_head.next;
-                mfd = list_entry(tmp, struct mdt_file_data, mfd_list);
-
-                /* Remove mfd handle so it can't be found again.
-                 * We are consuming the mfd_list reference here. */
-                class_handle_unhash(&mfd->mfd_handle);
-                list_move_tail(&mfd->mfd_list, &closing_list);
-        }
-        spin_unlock(&med->med_open_lock);
-
-        list_for_each_entry_safe(mfd, n, &closing_list, mfd_list) {
-                list_del_init(&mfd->mfd_list);
-                mdt_mfd_close(info, mfd);
-                /* TODO: if we close the unlinked file,
-                 * we need to remove its objects from OST */
-                memset(&ma->ma_attr, 0, sizeof(ma->ma_attr));
-                spin_lock(&med->med_open_lock);
-                ma->ma_lmm_size = lmm_size;
-                ma->ma_cookie_size = cookie_size;
-                ma->ma_need = MA_LOV | MA_COOKIE;
-                ma->ma_valid = 0;
-                spin_unlock(&med->med_open_lock);
-        }
-
         info->mti_mdt = NULL;
         mdt_client_del(&env, mdt);
 
-        EXIT;
-out:
-        if (lmm_size) {
-                OBD_FREE(ma->ma_lmm, lmm_size);
-                ma->ma_lmm = NULL;
-        }
-        if (cookie_size) {
-                OBD_FREE(ma->ma_cookie, cookie_size);
-                ma->ma_cookie = NULL;
-        }
         lu_env_fini(&env);
-
-        return rc;
+        RETURN(rc);
 }
 
 static void mdt_allow_cli(struct mdt_device *m, unsigned int flag)
@@ -5241,7 +5294,7 @@ static void mdt_allow_cli(struct mdt_device *m, unsigned int flag)
 }
 
 static int mdt_upcall(const struct lu_env *env, struct md_device *md,
-                      enum md_upcall_event ev)
+                      enum md_upcall_event ev, void *data)
 {
         struct mdt_device *m = mdt_dev(&md->md_lu_dev);
         struct md_device  *next  = m->mdt_child;
@@ -5257,10 +5310,8 @@ static int mdt_upcall(const struct lu_env *env, struct md_device *md,
                         CDEBUG(D_INFO, "get max mdsize %d max cookiesize %d\n",
                                      m->mdt_max_mdsize, m->mdt_max_cookiesize);
                         mdt_allow_cli(m, CONFIG_SYNC);
-#ifdef HAVE_QUOTA_SUPPORT
-                        if (md->md_lu_dev.ld_obd->obd_recovering == 0)
-                                next->md_ops->mdo_quota.mqo_recovery(env, next);
-#endif
+                        if (data)
+                                (*(__u64 *)data) = m->mdt_mount_count;
                         break;
                 case MD_NO_TRANS:
                         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
@@ -5271,6 +5322,13 @@ static int mdt_upcall(const struct lu_env *env, struct md_device *md,
                         /* Check that MDT is not yet configured */
                         LASSERT(!m->mdt_fl_cfglog);
                         break;
+#ifdef HAVE_QUOTA_SUPPORT
+                case MD_LOV_QUOTA:
+                        if (md->md_lu_dev.ld_obd->obd_recovering == 0 &&
+                            likely(md->md_lu_dev.ld_obd->obd_stopping == 0))
+                                next->md_ops->mdo_quota.mqo_recovery(env, next);
+                        break;
+#endif
                 default:
                         CERROR("invalid event\n");
                         rc = -EINVAL;
@@ -5305,48 +5363,45 @@ static int mdt_obd_notify(struct obd_device *host,
         RETURN(0);
 }
 
-static int mdt_ioc_fid2path(struct lu_env *env, struct mdt_device *mdt,
-                            struct obd_ioctl_data *data)
+static int mdt_rpc_fid2path(struct mdt_thread_info *info, void *key,
+                            void *val, int vallen)
 {
-        struct lu_context  ioctl_session;
-        struct mdt_object *obj;
-        struct lu_fid     *fid;
-        char  *path = NULL;
-        __u64  recno;
-        int    pathlen = data->ioc_plen1;
-        int    linkno;
-        int    rc;
-        ENTRY;
+        struct mdt_device *mdt = mdt_dev(info->mti_exp->exp_obd->obd_lu_dev);
+        struct getinfo_fid2path *fpout, *fpin;
+        int rc = 0;
 
+        fpin = key + size_round(sizeof(KEY_FID2PATH));
+        fpout = val;
 
-        fid = (struct lu_fid *)data->ioc_inlbuf1;
-        memcpy(&recno, data->ioc_inlbuf2, sizeof(recno));
-        memcpy(&linkno, data->ioc_inlbuf3, sizeof(linkno));
-        CDEBUG(D_IOCTL, "path get "DFID" from "LPU64" #%d\n",
-               PFID(fid), recno, linkno);
+        if (lustre_msg_swabbed(mdt_info_req(info)->rq_reqmsg))
+                lustre_swab_fid2path(fpin);
 
-        if (!fid_is_sane(fid))
+        memcpy(fpout, fpin, sizeof(*fpin));
+        if (fpout->gf_pathlen != vallen - sizeof(*fpin))
                 RETURN(-EINVAL);
 
-        if (pathlen < 3)
-                RETURN(-EOVERFLOW);
+        rc = mdt_fid2path(info->mti_env, mdt, fpout);
+        RETURN(rc);
+}
 
-        rc = lu_context_init(&ioctl_session, LCT_SESSION);
-        if (rc)
-                RETURN(rc);
-        ioctl_session.lc_thread = (struct ptlrpc_thread *)cfs_current();
-        lu_context_enter(&ioctl_session);
-        env->le_ses = &ioctl_session;
+static int mdt_fid2path(const struct lu_env *env, struct mdt_device *mdt,
+                        struct getinfo_fid2path *fp)
+{
+        struct mdt_object *obj;
+        int    rc;
+        ENTRY;
+
+        CDEBUG(D_IOCTL, "path get "DFID" from "LPU64" #%d\n",
+               PFID(&fp->gf_fid), fp->gf_recno, fp->gf_linkno);
 
-        OBD_ALLOC(path, pathlen);
-        if (path == NULL)
-                GOTO(out_context, rc = -ENOMEM);
+        if (!fid_is_sane(&fp->gf_fid))
+                RETURN(-EINVAL);
 
-        obj = mdt_object_find(env, mdt, fid);
+        obj = mdt_object_find(env, mdt, &fp->gf_fid);
         if (obj == NULL || IS_ERR(obj)) {
-                CDEBUG(D_IOCTL, "no object "DFID": %ld\n", PFID(fid),
+                CDEBUG(D_IOCTL, "no object "DFID": %ld\n",PFID(&fp->gf_fid),
                        PTR_ERR(obj));
-                GOTO(out_free, rc = -EINVAL);
+                RETURN(-EINVAL);
         }
 
         rc = lu_object_exists(&obj->mot_obj.mo_lu);
@@ -5356,30 +5411,59 @@ static int mdt_ioc_fid2path(struct lu_env *env, struct mdt_device *mdt,
                 else
                         rc = -ENOENT;
                 mdt_object_put(env, obj);
-                CDEBUG(D_IOCTL, "nonlocal object "DFID": %d\n", PFID(fid),
-                       rc);
-                GOTO(out_free, rc);
+                CDEBUG(D_IOCTL, "nonlocal object "DFID": %d\n",
+                       PFID(&fp->gf_fid), rc);
+                RETURN(rc);
         }
 
-        rc = mo_path(env, md_object_next(&obj->mot_obj), path, pathlen, &recno,
-                     &linkno);
+        rc = mo_path(env, md_object_next(&obj->mot_obj), fp->gf_path,
+                     fp->gf_pathlen, &fp->gf_recno, &fp->gf_linkno);
         mdt_object_put(env, obj);
-        if (rc)
-               GOTO(out_free, rc);
 
-        if (copy_to_user(data->ioc_pbuf1, path, pathlen))
-                rc = -EFAULT;
+        RETURN(rc);
+}
+
+static int mdt_get_info(struct mdt_thread_info *info)
+{
+        struct ptlrpc_request *req = mdt_info_req(info);
+        char *key;
+        int keylen;
+        __u32 *vallen;
+        void *valout;
+        int rc;
+        ENTRY;
+
+        key = req_capsule_client_get(info->mti_pill, &RMF_GETINFO_KEY);
+        if (key == NULL) {
+                CDEBUG(D_IOCTL, "No GETINFO key");
+                RETURN(-EFAULT);
+        }
+        keylen = req_capsule_get_size(info->mti_pill, &RMF_GETINFO_KEY,
+                                      RCL_CLIENT);
 
-        memcpy(data->ioc_inlbuf2, &recno, sizeof(recno));
-        memcpy(data->ioc_inlbuf3, &linkno, sizeof(linkno));
+        vallen = req_capsule_client_get(info->mti_pill, &RMF_GETINFO_VALLEN);
+        if (vallen == NULL) {
+                CDEBUG(D_IOCTL, "Unable to get RMF_GETINFO_VALLEN buffer");
+                RETURN(-EFAULT);
+        }
 
-        EXIT;
-out_free:
-        OBD_FREE(path, pathlen);
-out_context:
-        lu_context_exit(&ioctl_session);
-        lu_context_fini(&ioctl_session);
-        return rc;
+        req_capsule_set_size(info->mti_pill, &RMF_GETINFO_VAL, RCL_SERVER,
+                             *vallen);
+        rc = req_capsule_server_pack(info->mti_pill);
+        valout = req_capsule_server_get(info->mti_pill, &RMF_GETINFO_VAL);
+        if (valout == NULL) {
+                CDEBUG(D_IOCTL, "Unable to get get-info RPC out buffer");
+                RETURN(-EFAULT);
+        }
+
+        if (KEY_IS(KEY_FID2PATH))
+                rc = mdt_rpc_fid2path(info, key, valout, *vallen);
+        else
+                rc = -EINVAL;
+
+        lustre_msg_set_status(req->rq_repmsg, rc);
+
+        RETURN(rc);
 }
 
 /* Pass the ioc down */
@@ -5434,9 +5518,6 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
                 target_stop_recovery_thread(obd);
                 rc = 0;
                 break;
-        case OBD_IOC_FID2PATH:
-                rc = mdt_ioc_fid2path(&env, mdt, karg);
-                break;
         case OBD_IOC_CHANGELOG_REG:
         case OBD_IOC_CHANGELOG_DEREG:
         case OBD_IOC_CHANGELOG_CLEAR:
@@ -5468,7 +5549,8 @@ int mdt_postrecov(const struct lu_env *env, struct mdt_device *mdt)
 
         rc = ld->ld_ops->ldo_recovery_complete(env, ld);
 #ifdef HAVE_QUOTA_SUPPORT
-        next->md_ops->mdo_quota.mqo_recovery(env, next);
+        if (likely(obd->obd_stopping == 0))
+                next->md_ops->mdo_quota.mqo_recovery(env, next);
 #endif
         RETURN(rc);
 }
@@ -5690,6 +5772,7 @@ static struct mdt_handler mdt_mds_ops[] = {
 DEF_MDT_HNDL_F(0,                         CONNECT,      mdt_connect),
 DEF_MDT_HNDL_F(0,                         DISCONNECT,   mdt_disconnect),
 DEF_MDT_HNDL_F(0,                         SET_INFO,     mdt_set_info),
+DEF_MDT_HNDL_F(0,                         GET_INFO,     mdt_get_info),
 DEF_MDT_HNDL_F(0           |HABEO_REFERO, GETSTATUS,    mdt_getstatus),
 DEF_MDT_HNDL_F(HABEO_CORPUS,              GETATTR,      mdt_getattr),
 DEF_MDT_HNDL_F(HABEO_CORPUS|HABEO_REFERO, GETATTR_NAME, mdt_getattr_name),