Branch HEAD

b=19550
i=rread
i=zam

Move unlinked exports to the new list until they become 'zombie'. Wait for these
exports during umount. Output in dump_exports() is exteneded and contain more
fields to help with debugging. The new exp_cb_count was added to track number of
callbacks keeping the export references.

diff --git a/lustre/include/lustre_export.h b/lustre/include/lustre_export.h
index f284bab..d78a26d 100644 (file)
--- a/lustre/include/lustre_export.h
+++ b/lustre/include/lustre_export.h
@@ -144,6 +144,7 @@ struct obd_export {
         struct portals_handle     exp_handle;
         atomic_t                  exp_refcount;
         atomic_t                  exp_rpc_count;
         struct portals_handle     exp_handle;
         atomic_t                  exp_refcount;
         atomic_t                  exp_rpc_count;

+        atomic_t                  exp_cb_count;

         struct obd_uuid           exp_client_uuid;
         struct list_head          exp_obd_chain;
         struct hlist_node         exp_uuid_hash; /* uuid-export hash*/
         struct obd_uuid           exp_client_uuid;
         struct list_head          exp_obd_chain;
         struct hlist_node         exp_uuid_hash; /* uuid-export hash*/

diff --git a/lustre/include/obd.h b/lustre/include/obd.h
index 10da129..a7c53bc 100644 (file)
--- a/lustre/include/obd.h
+++ b/lustre/include/obd.h
@@ -1047,6 +1047,7 @@ struct obd_device {
         atomic_t                obd_refcount;
         cfs_waitq_t             obd_refcount_waitq;
         struct list_head        obd_exports;
         atomic_t                obd_refcount;
         cfs_waitq_t             obd_refcount_waitq;
         struct list_head        obd_exports;

+        struct list_head        obd_unlinked_exports;

         struct list_head        obd_delayed_exports;
         int                     obd_num_exports;
         spinlock_t              obd_nid_lock;
         struct list_head        obd_delayed_exports;
         int                     obd_num_exports;
         spinlock_t              obd_nid_lock;

diff --git a/lustre/include/obd_class.h b/lustre/include/obd_class.h
index 07f783d..86690a2 100644 (file)
--- a/lustre/include/obd_class.h
+++ b/lustre/include/obd_class.h
@@ -101,6 +101,7 @@ int obd_zombie_impexp_init(void);
 void obd_zombie_impexp_stop(void);
 void obd_zombie_impexp_cull(void);
 void obd_zombie_barrier(void);
 void obd_zombie_impexp_stop(void);
 void obd_zombie_impexp_cull(void);
 void obd_zombie_barrier(void);

+void obd_exports_barrier(struct obd_device *obd);

 
 /* obd_config.c */
 int class_process_config(struct lustre_cfg *lcfg);
 
 /* obd_config.c */
 int class_process_config(struct lustre_cfg *lcfg);


@@ -114,6 +115,7 @@ struct obd_device *class_incref(struct obd_device *obd,
                                 const char *scope, const void *source);
 void class_decref(struct obd_device *obd,
                   const char *scope, const void *source);
                                 const char *scope, const void *source);
 void class_decref(struct obd_device *obd,
                   const char *scope, const void *source);

+void dump_exports(struct obd_device *obd);

 
 /*obdecho*/
 #ifdef LPROCFS
 
 /*obdecho*/
 #ifdef LPROCFS

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 66bd145..7178463 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -4337,6 +4337,7 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m)
         mdt_stop_ptlrpc_service(m);
         mdt_llog_ctxt_unclone(env, m, LLOG_CHANGELOG_ORIG_CTXT);
         mdt_obd_llog_cleanup(obd);
         mdt_stop_ptlrpc_service(m);
         mdt_llog_ctxt_unclone(env, m, LLOG_CHANGELOG_ORIG_CTXT);
         mdt_obd_llog_cleanup(obd);

+        obd_exports_barrier(obd);

         obd_zombie_barrier();
 #ifdef HAVE_QUOTA_SUPPORT
         next->md_ops->mdo_quota.mqo_cleanup(env, next);
         obd_zombie_barrier();
 #ifdef HAVE_QUOTA_SUPPORT
         next->md_ops->mdo_quota.mqo_cleanup(env, next);

diff --git a/lustre/mdt/mdt_recovery.c b/lustre/mdt/mdt_recovery.c
index 810d252..21fe440 100644 (file)
--- a/lustre/mdt/mdt_recovery.c
+++ b/lustre/mdt/mdt_recovery.c
@@ -926,6 +926,7 @@ static int mdt_txn_stop_cb(const struct lu_env *env,
          * export as parameter */
         mdt_trans_add_cb(txn, lut_cb_last_committed,
                          class_export_get(mti->mti_exp));
          * export as parameter */
         mdt_trans_add_cb(txn, lut_cb_last_committed,
                          class_export_get(mti->mti_exp));

+        atomic_inc(&mti->mti_exp->exp_cb_count);

 
         return mdt_last_rcvd_update(mti, txn);
 }
 
         return mdt_last_rcvd_update(mti, txn);
 }

diff --git a/lustre/obdclass/genops.c b/lustre/obdclass/genops.c
index a74ee18..4af4a61 100644 (file)
--- a/lustre/obdclass/genops.c
+++ b/lustre/obdclass/genops.c
@@ -756,6 +756,7 @@ void class_export_put(struct obd_export *exp)
         LASSERT(atomic_read(&exp->exp_refcount) < 0x5a5a5a);
 
         if (atomic_dec_and_test(&exp->exp_refcount)) {
         LASSERT(atomic_read(&exp->exp_refcount) < 0x5a5a5a);
 
         if (atomic_dec_and_test(&exp->exp_refcount)) {

+                LASSERT(!list_empty(&exp->exp_obd_chain));

                 CDEBUG(D_IOCTL, "final put %p/%s\n",
                        exp, exp->exp_client_uuid.uuid);
                 obd_zombie_export_add(exp);
                 CDEBUG(D_IOCTL, "final put %p/%s\n",
                        exp, exp->exp_client_uuid.uuid);
                 obd_zombie_export_add(exp);


@@ -780,6 +781,7 @@ struct obd_export *class_new_export(struct obd_device *obd,
         export->exp_lock_hash = NULL;
         atomic_set(&export->exp_refcount, 2);
         atomic_set(&export->exp_rpc_count, 0);
         export->exp_lock_hash = NULL;
         atomic_set(&export->exp_refcount, 2);
         atomic_set(&export->exp_rpc_count, 0);

+        atomic_set(&export->exp_cb_count, 0);

         export->exp_obd = obd;
         CFS_INIT_LIST_HEAD(&export->exp_outstanding_replies);
         spin_lock_init(&export->exp_uncommitted_replies_lock);
         export->exp_obd = obd;
         CFS_INIT_LIST_HEAD(&export->exp_outstanding_replies);
         spin_lock_init(&export->exp_uncommitted_replies_lock);


@@ -835,7 +837,7 @@ void class_unlink_export(struct obd_export *exp)
                                 &exp->exp_client_uuid,
                                 &exp->exp_uuid_hash);
 
                                 &exp->exp_client_uuid,
                                 &exp->exp_uuid_hash);
 

-        list_del_init(&exp->exp_obd_chain);
+        list_move(&exp->exp_obd_chain, &exp->exp_obd->obd_unlinked_exports);

         list_del_init(&exp->exp_obd_chain_timed);
         exp->exp_obd->obd_num_exports--;
         spin_unlock(&exp->exp_obd->obd_dev_lock);
         list_del_init(&exp->exp_obd_chain_timed);
         exp->exp_obd->obd_num_exports--;
         spin_unlock(&exp->exp_obd->obd_dev_lock);


@@ -1299,6 +1301,72 @@ int obd_export_evict_by_uuid(struct obd_device *obd, const char *uuid)
 }
 EXPORT_SYMBOL(obd_export_evict_by_uuid);
 
 }
 EXPORT_SYMBOL(obd_export_evict_by_uuid);
 

+static void print_export_data(struct obd_export *exp, const char *status)
+{
+        struct ptlrpc_reply_state *rs;
+        struct ptlrpc_reply_state *first_reply = NULL;
+        int nreplies = 0;
+
+        spin_lock(&exp->exp_lock);
+        list_for_each_entry (rs, &exp->exp_outstanding_replies, rs_exp_list) {
+                if (nreplies == 0)
+                        first_reply = rs;
+                nreplies++;
+        }
+        spin_unlock(&exp->exp_lock);
+
+        CDEBUG(D_HA, "%s: %s %p %s %s %d %d %d: %p %s %d %d %d %d "LPU64"\n",
+               exp->exp_obd->obd_name, status, exp, exp->exp_client_uuid.uuid,
+               obd_export_nid2str(exp), atomic_read(&exp->exp_refcount),
+               exp->exp_failed, nreplies, first_reply,
+               nreplies > 3 ? "..." : "",
+               atomic_read(&exp->exp_rpc_count),
+               atomic_read(&exp->exp_cb_count),
+               exp->exp_disconnected, exp->exp_delayed,
+               exp->exp_last_committed);
+}
+
+void dump_exports(struct obd_device *obd)
+{
+        struct obd_export *exp;
+
+        spin_lock(&obd->obd_dev_lock);
+        list_for_each_entry(exp, &obd->obd_exports, exp_obd_chain)
+                print_export_data(exp, "ACTIVE");
+        list_for_each_entry(exp, &obd->obd_unlinked_exports, exp_obd_chain)
+                print_export_data(exp, "UNLINKED");
+        list_for_each_entry(exp, &obd->obd_delayed_exports, exp_obd_chain)
+                print_export_data(exp, "DELAYED");
+        spin_unlock(&obd->obd_dev_lock);
+        spin_lock(&obd_zombie_impexp_lock);
+        list_for_each_entry(exp, &obd_zombie_exports, exp_obd_chain)
+                print_export_data(exp, "ZOMBIE");
+        spin_unlock(&obd_zombie_impexp_lock);
+}
+EXPORT_SYMBOL(dump_exports);
+
+void obd_exports_barrier(struct obd_device *obd)
+{
+        int waited = 2;
+        LASSERT(list_empty(&obd->obd_exports));
+        spin_lock(&obd->obd_dev_lock);
+        while (!list_empty(&obd->obd_unlinked_exports)) {
+                spin_unlock(&obd->obd_dev_lock);
+                cfs_schedule_timeout(CFS_TASK_UNINT, cfs_time_seconds(waited));
+                if (waited > 5 && IS_PO2(waited)) {
+                        LCONSOLE_WARN("Waiting for obd_unlinked_exports "
+                                      "more than %d seconds. "
+                                      "The obd refcount = %d. Is it stuck?\n",
+                                      waited, atomic_read(&obd->obd_refcount));
+                        dump_exports(obd);
+                }
+                waited *= 2;
+                spin_lock(&obd->obd_dev_lock);
+        }
+        spin_unlock(&obd->obd_dev_lock);
+}
+EXPORT_SYMBOL(obd_exports_barrier);
+

 /**
  * kill zombie imports and exports
  */
 /**
  * kill zombie imports and exports
  */


@@ -1369,8 +1437,11 @@ static int obd_zombie_impexp_check(void *arg)
  * Add export to the obd_zombe thread and notify it.
  */
 static void obd_zombie_export_add(struct obd_export *exp) {
  * Add export to the obd_zombe thread and notify it.
  */
 static void obd_zombie_export_add(struct obd_export *exp) {

+        spin_lock(&exp->exp_obd->obd_dev_lock);
+        LASSERT(!list_empty(&exp->exp_obd_chain));
+        list_del_init(&exp->exp_obd_chain);
+        spin_unlock(&exp->exp_obd->obd_dev_lock);

         spin_lock(&obd_zombie_impexp_lock);
         spin_lock(&obd_zombie_impexp_lock);

-        LASSERT(list_empty(&exp->exp_obd_chain));

         list_add(&exp->exp_obd_chain, &obd_zombie_exports);
         spin_unlock(&obd_zombie_impexp_lock);
 
         list_add(&exp->exp_obd_chain, &obd_zombie_exports);
         spin_unlock(&obd_zombie_impexp_lock);
 

diff --git a/lustre/obdclass/obd_config.c b/lustre/obdclass/obd_config.c
index a6917f1..2a0b0f8 100644 (file)
--- a/lustre/obdclass/obd_config.c
+++ b/lustre/obdclass/obd_config.c
@@ -255,6 +255,7 @@ int class_attach(struct lustre_cfg *lcfg)
         obd->obd_pool_slv = 0;
 
         CFS_INIT_LIST_HEAD(&obd->obd_exports);
         obd->obd_pool_slv = 0;
 
         CFS_INIT_LIST_HEAD(&obd->obd_exports);

+        CFS_INIT_LIST_HEAD(&obd->obd_unlinked_exports);

         CFS_INIT_LIST_HEAD(&obd->obd_delayed_exports);
         CFS_INIT_LIST_HEAD(&obd->obd_exports_timed);
         CFS_INIT_LIST_HEAD(&obd->obd_nid_stats);
         CFS_INIT_LIST_HEAD(&obd->obd_delayed_exports);
         CFS_INIT_LIST_HEAD(&obd->obd_exports_timed);
         CFS_INIT_LIST_HEAD(&obd->obd_nid_stats);


@@ -452,35 +453,6 @@ int class_detach(struct obd_device *obd, struct lustre_cfg *lcfg)
         RETURN(0);
 }
 
         RETURN(0);
 }
 

-static void dump_exports(struct obd_device *obd)
-{
-        struct obd_export *exp;
-
-        spin_lock(&obd->obd_dev_lock);
-        list_for_each_entry(exp, &obd->obd_exports, exp_obd_chain) {
-                struct ptlrpc_reply_state *rs;
-                struct ptlrpc_reply_state *first_reply = NULL;
-                int                        nreplies = 0;
-
-                spin_lock(&exp->exp_lock);
-                list_for_each_entry (rs, &exp->exp_outstanding_replies,
-                                     rs_exp_list) {
-                        if (nreplies == 0)
-                                first_reply = rs;
-                        nreplies++;
-                }
-                spin_unlock(&exp->exp_lock);
-
-                CDEBUG(D_IOCTL, "%s: %p %s %s %d %d %d: %p %s\n",
-                       obd->obd_name, exp, exp->exp_client_uuid.uuid,
-                       obd_export_nid2str(exp),
-                       atomic_read(&exp->exp_refcount),
-                       exp->exp_failed, nreplies, first_reply,
-                       nreplies > 3 ? "..." : "");
-        }
-        spin_unlock(&obd->obd_dev_lock);
-}
-

 int class_cleanup(struct obd_device *obd, struct lustre_cfg *lcfg)
 {
         int err = 0;
 int class_cleanup(struct obd_device *obd, struct lustre_cfg *lcfg)
 {
         int err = 0;

diff --git a/lustre/obdfilter/filter.c b/lustre/obdfilter/filter.c
index 21c24a7..c0b4997 100644 (file)
--- a/lustre/obdfilter/filter.c
+++ b/lustre/obdfilter/filter.c
@@ -90,6 +90,7 @@ static void filter_commit_cb(struct obd_device *obd, __u64 transno,
         struct obd_export *exp = cb_data;
         LASSERT(exp->exp_obd == obd);
         obd_transno_commit_cb(obd, transno, exp, error);
         struct obd_export *exp = cb_data;
         LASSERT(exp->exp_obd == obd);
         obd_transno_commit_cb(obd, transno, exp, error);

+        atomic_dec(&exp->exp_cb_count);

         class_export_put(exp);
 }
 
         class_export_put(exp);
 }
 


@@ -167,6 +168,7 @@ int filter_finish_transno(struct obd_export *exp, struct inode *inode,
                 err = -EINVAL;
         } else {
                 class_export_get(exp); /* released when the cb is called */
                 err = -EINVAL;
         } else {
                 class_export_get(exp); /* released when the cb is called */

+                atomic_inc(&exp->exp_cb_count);

                 if (!force_sync)
                         force_sync = fsfilt_add_journal_cb(exp->exp_obd,
                                                            last_rcvd,
                 if (!force_sync)
                         force_sync = fsfilt_add_journal_cb(exp->exp_obd,
                                                            last_rcvd,


@@ -2637,8 +2639,7 @@ static int filter_precleanup(struct obd_device *obd,
                 break;
         case OBD_CLEANUP_EXPORTS:
                 /* Stop recovery before namespace cleanup. */
                 break;
         case OBD_CLEANUP_EXPORTS:
                 /* Stop recovery before namespace cleanup. */

-                target_stop_recovery_thread(obd);
-                target_cleanup_recovery(obd);
+                target_recovery_fini(obd);

                 rc = filter_llog_preclean(obd);
                 break;
         }
                 rc = filter_llog_preclean(obd);
                 break;
         }


@@ -2654,14 +2655,8 @@ static int filter_cleanup(struct obd_device *obd)
                 LCONSOLE_WARN("%s: shutting down for failover; client state "
                               "will be preserved.\n", obd->obd_name);
 
                 LCONSOLE_WARN("%s: shutting down for failover; client state "
                               "will be preserved.\n", obd->obd_name);
 

-        if (!list_empty(&obd->obd_exports)) {
-                CERROR("%s: still has clients!\n", obd->obd_name);
-                class_disconnect_exports(obd);
-                if (!list_empty(&obd->obd_exports)) {
-                        CERROR("still has exports after forced cleanup?\n");
-                        RETURN(-EBUSY);
-                }
-        }
+        obd_exports_barrier(obd);
+        obd_zombie_barrier();

 
         lprocfs_remove_proc_entry("clear", obd->obd_proc_exports_entry);
         lprocfs_free_per_client_stats(obd);
 
         lprocfs_remove_proc_entry("clear", obd->obd_proc_exports_entry);
         lprocfs_free_per_client_stats(obd);

diff --git a/lustre/ptlrpc/target.c b/lustre/ptlrpc/target.c
index 3dc9675..ecf1e30 100644 (file)
--- a/lustre/ptlrpc/target.c
+++ b/lustre/ptlrpc/target.c
@@ -309,6 +309,8 @@ void lut_cb_last_committed(struct lu_target *lut, __u64 transno,
         } else {
                 spin_unlock(&lut->lut_translock);
         }
         } else {
                 spin_unlock(&lut->lut_translock);
         }

+        LASSERT(atomic_read(&exp->exp_cb_count) > 0);
+        atomic_dec(&exp->exp_cb_count);

         class_export_put(exp);
         if (transno)
                 CDEBUG(D_HA, "%s: transno "LPD64" is committed\n",
         class_export_put(exp);
         if (transno)
                 CDEBUG(D_HA, "%s: transno "LPD64" is committed\n",