- take reference on export for commit callback

- move mfd clenanup to mds_disconnect() as mds_export_destroy() may be delayed
  and busy inodes may remain during umount
b:18734
i:adilger,shadow

diff --git a/lustre/include/obd_class.h b/lustre/include/obd_class.h
index 66a8918..459e470 100644 (file)
--- a/lustre/include/obd_class.h
+++ b/lustre/include/obd_class.h
@@ -192,6 +192,7 @@ void class_del_profiles(void);
         exp_;                                                                  \
 })
 
+/* class_export_put() is non-blocking */
 #define class_export_put(exp)                                                  \
 do {                                                                           \
         LASSERT((exp) != NULL);                                                \

diff --git a/lustre/mds/handler.c b/lustre/mds/handler.c
index a7b20ac..7603f75 100644 (file)
--- a/lustre/mds/handler.c
+++ b/lustre/mds/handler.c
@@ -406,10 +406,22 @@ int mds_init_export(struct obd_export *exp)
         RETURN(ldlm_init_export(exp));
 }
 
-static int mds_destroy_export(struct obd_export *export)
+static int mds_destroy_export(struct obd_export *exp)
+{
+        int rc = 0;
+        ENTRY;
+
+        target_destroy_export(exp);
+        ldlm_destroy_export(exp);
+
+        LASSERT(list_empty(&exp->exp_mds_data.med_open_head));
+        mds_client_free(exp);
+}
+
+static int mds_cleanup_mfd(struct obd_export *exp)
 {
         struct mds_export_data *med;
-        struct obd_device *obd = export->exp_obd;
+        struct obd_device *obd = exp->exp_obd;
         struct mds_obd *mds = &obd->u.mds;
         struct lvfs_run_ctxt saved;
         struct lov_mds_md *lmm;
@@ -420,12 +432,26 @@ static int mds_destroy_export(struct obd_export *export)
         int rc = 0;
         ENTRY;
 
-        med = &export->exp_mds_data;
-        target_destroy_export(export);
-        ldlm_destroy_export(export);
+        med = &exp->exp_mds_data;
 
-        if (obd_uuid_equals(&export->exp_client_uuid, &obd->obd_uuid))
+        spin_lock(&med->med_open_lock);
+        if (list_empty(&med->med_open_head)) {
+                spin_unlock(&med->med_open_lock);
                 RETURN(0);
+        }
+
+        CFS_INIT_LIST_HEAD(&closing_list);
+        while (!list_empty(&med->med_open_head)) {
+                struct list_head *tmp = med->med_open_head.next;
+                struct mds_file_data *mfd =
+                        list_entry(tmp, struct mds_file_data, mfd_list);
+
+                /* Remove mfd handle so it can't be found again.
+                 * We are consuming the mfd_list reference here. */
+                mds_mfd_unlink(mfd, 0);
+                list_add_tail(&mfd->mfd_list, &closing_list);
+        }
+        spin_unlock(&med->med_open_lock);
 
         lmm_sz = mds->mds_max_mdsize;
         OBD_ALLOC(lmm, lmm_sz);
@@ -446,20 +472,6 @@ static int mds_destroy_export(struct obd_export *export)
 
         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
         /* Close any open files (which may also cause orphan unlinking). */
-        CFS_INIT_LIST_HEAD(&closing_list);
-        spin_lock(&med->med_open_lock);
-        while (!list_empty(&med->med_open_head)) {
-                struct list_head *tmp = med->med_open_head.next;
-                struct mds_file_data *mfd =
-                        list_entry(tmp, struct mds_file_data, mfd_list);
-
-                /* Remove mfd handle so it can't be found again.
-                 * We are consuming the mfd_list reference here. */
-                mds_mfd_unlink(mfd, 0);
-                list_add_tail(&mfd->mfd_list, &closing_list);
-        }
-        spin_unlock(&med->med_open_lock);
-
         list_for_each_entry_safe(mfd, n, &closing_list, mfd_list) {
                 int lmm_size = lmm_sz;
                 umode_t mode = mfd->mfd_dentry->d_inode->i_mode;
@@ -485,7 +497,7 @@ static int mds_destroy_export(struct obd_export *export)
 
                 list_del_init(&mfd->mfd_list);
                 rc = mds_mfd_close(NULL, REQ_REC_OFF, obd, mfd,
-                                   !(export->exp_flags & OBD_OPT_FAILOVER),
+                                   !(exp->exp_flags & OBD_OPT_FAILOVER),
                                    lmm, lmm_size, logcookies,
                                    mds->mds_max_cookiesize,
                                    &valid);
@@ -505,14 +517,10 @@ static int mds_destroy_export(struct obd_export *export)
                 }
 
         }
-
+        pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
         OBD_FREE(logcookies, cookie_sz);
         OBD_FREE(lmm, lmm_sz);
-
-        pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
-        mds_client_free(export);
-
- out:
+out:
         RETURN(rc);
 }
 
@@ -546,6 +554,7 @@ static int mds_disconnect(struct obd_export *exp)
                 spin_unlock(&svc->srv_lock);
         }
         spin_unlock(&exp->exp_lock);
+        rc = mds_cleanup_mfd(exp);
 
         class_export_put(exp);
         RETURN(rc);

diff --git a/lustre/mds/mds_reint.c b/lustre/mds/mds_reint.c
index 381b443..d347a7c 100644 (file)
--- a/lustre/mds/mds_reint.c
+++ b/lustre/mds/mds_reint.c
@@ -65,7 +65,11 @@ void mds_commit_cb(struct obd_device *obd, __u64 transno, void *data,
                    int error)
 {
         struct obd_export *exp = data;
+        LASSERTF(exp->exp_obd == obd,
+                 "%s: bad export (%p), obd (%p) != exp->exp_obd (%p)\n",
+                 obd->obd_name, exp, obd, exp->exp_obd);
         obd_transno_commit_cb(obd, transno, exp, error);
+        class_export_put(exp);
 }
 
 struct mds_logcancel_data {
@@ -319,13 +323,13 @@ int mds_finish_transno(struct mds_obd *mds, struct inode **inodes, void *handle,
                 if (!force_sync)
                         force_sync = fsfilt_add_journal_cb(obd, transno,
                                                            handle, mds_commit_cb,
-                                                           exp);
+                                                           class_export_get(exp));
 
                 err = fsfilt_write_record(obd, mds->mds_rcvd_filp, lcd,
                                           sizeof(*lcd), &off,
                                           force_sync | exp->exp_need_sync);
                 if (force_sync)
-                        mds_commit_cb(obd, transno, exp, err);
+                        mds_commit_cb(obd, transno, class_export_get(exp), err);
         }
 
         if (err) {

diff --git a/lustre/obdclass/genops.c b/lustre/obdclass/genops.c
index ba6164e..cdae4f4 100644 (file)
--- a/lustre/obdclass/genops.c
+++ b/lustre/obdclass/genops.c
@@ -644,6 +644,8 @@ static void export_handle_addref(void *export)
         class_export_get(export);
 }
 
+/* called from mds_commit_cb() in context of journal commit callback
+ * and cannot call any blocking functions. */
 void __class_export_put(struct obd_export *exp)
 {
         if (atomic_dec_and_test(&exp->exp_refcount)) {

diff --git a/lustre/obdfilter/filter.c b/lustre/obdfilter/filter.c
index 80a5d1c..5f49442 100644 (file)
--- a/lustre/obdfilter/filter.c
+++ b/lustre/obdfilter/filter.c
@@ -84,7 +84,11 @@ static void filter_commit_cb(struct obd_device *obd, __u64 transno,
                              void *cb_data, int error)
 {
         struct obd_export *exp = cb_data;
+        LASSERTF(exp->exp_obd == obd,
+                 "%s: bad export (%p), obd (%p) != exp->exp_obd (%p)\n",
+                 obd->obd_name, exp, obd, exp->exp_obd);
         obd_transno_commit_cb(obd, transno, exp, error);
+        class_export_put(exp);
 }
 
 int filter_version_get_check(struct obd_export *exp,
@@ -167,13 +171,13 @@ int filter_finish_transno(struct obd_export *exp, struct inode *inode,
                                                            last_rcvd,
                                                            oti->oti_handle,
                                                            filter_commit_cb,
-                                                           exp);
+                                                           class_export_get(exp));
 
                 err = fsfilt_write_record(exp->exp_obd, filter->fo_rcvd_filp,
                                           lcd, sizeof(*lcd), &off,
                                           force_sync | exp->exp_need_sync);
                 if (force_sync)
-                        filter_commit_cb(exp->exp_obd, last_rcvd, exp, err);
+                        filter_commit_cb(exp->exp_obd, last_rcvd, class_export_get(exp), err);
         }
         if (err) {
                 log_pri = D_ERROR;